Cortex AI Migration Guide¶

from openai import AzureOpenAI
import os

client = AzureOpenAI(
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
    api_version="2024-12-01-preview",
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"]
)

def summarize_report(document_text: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are a federal analyst. Summarize government reports concisely."},
            {"role": "user", "content": f"Summarize in 3 bullet points:\n\n{document_text}"}
        ],
        max_tokens=500,
        temperature=0.3
    )
    return response.choices[0].message.content

# RAG pipeline: Azure AI Search + Azure OpenAI
from azure.search.documents import SearchClient
from azure.search.documents.models import VectorizedQuery
from openai import AzureOpenAI
import os

# 1. Search for relevant documents
search_client = SearchClient(
    endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
    index_name="government-documents",
    credential=os.environ["AZURE_SEARCH_KEY"]
)

openai_client = AzureOpenAI(
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
    api_version="2024-12-01-preview",
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"]
)

def ask_question(question: str) -> str:
    # Generate embedding for the question
    embedding = openai_client.embeddings.create(
        model="text-embedding-3-large",
        input=question
    ).data[0].embedding

    # Search for relevant documents
    results = search_client.search(
        search_text=question,
        vector_queries=[
            VectorizedQuery(
                vector=embedding,
                k_nearest_neighbors=5,
                fields="content_vector"
            )
        ],
        top=5
    )

    # Build context from search results
    context = "\n\n---\n\n".join([
        f"Source: {r['title']}\n{r['content']}"
        for r in results
    ])

    # Generate answer
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are a federal analyst. Answer questions using only the provided context. Cite your sources."},
            {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"}
        ],
        max_tokens=1000,
        temperature=0.1
    )

    return response.choices[0].message.content

from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex,
    SearchField,
    SearchFieldDataType,
    VectorSearch,
    HnswAlgorithmConfiguration,
    VectorSearchProfile,
    SemanticConfiguration,
    SemanticSearch,
    SemanticPrioritizedFields,
    SemanticField
)

index_client = SearchIndexClient(
    endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
    credential=os.environ["AZURE_SEARCH_KEY"]
)

fields = [
    SearchField(name="id", type=SearchFieldDataType.String, key=True),
    SearchField(name="title", type=SearchFieldDataType.String, searchable=True),
    SearchField(name="content", type=SearchFieldDataType.String, searchable=True),
    SearchField(
        name="content_vector",
        type=SearchFieldDataType.Collection(SearchFieldDataType.Single),
        searchable=True,
        vector_search_dimensions=3072,
        vector_search_profile_name="vector-profile"
    ),
    SearchField(name="category", type=SearchFieldDataType.String, filterable=True),
    SearchField(name="published_date", type=SearchFieldDataType.DateTimeOffset, filterable=True, sortable=True)
]

vector_search = VectorSearch(
    algorithms=[HnswAlgorithmConfiguration(name="hnsw-config")],
    profiles=[VectorSearchProfile(name="vector-profile", algorithm_configuration_name="hnsw-config")]
)

semantic_config = SemanticConfiguration(
    name="semantic-config",
    prioritized_fields=SemanticPrioritizedFields(
        content_fields=[SemanticField(field_name="content")]
    )
)

index = SearchIndex(
    name="government-documents",
    fields=fields,
    vector_search=vector_search,
    semantic_search=SemanticSearch(configurations=[semantic_config])
)

index_client.create_or_update_index(index)

from azure.search.documents import SearchClient

search_client = SearchClient(
    endpoint=os.environ["AZURE_SEARCH_ENDPOINT"],
    index_name="government-documents",
    credential=os.environ["AZURE_SEARCH_KEY"]
)

# Generate embeddings and upload documents
documents = []
for row in spark.table("analytics_prod.raw.documents").collect():
    embedding = openai_client.embeddings.create(
        model="text-embedding-3-large",
        input=row["content"]
    ).data[0].embedding

    documents.append({
        "id": row["document_id"],
        "title": row["title"],
        "content": row["content"],
        "content_vector": embedding,
        "category": row["category"],
        "published_date": row["published_date"].isoformat()
    })

# Upload in batches
search_client.upload_documents(documents)

# Create an external function for SQL access
def search_documents(query: str, top_k: int = 5) -> list:
    embedding = openai_client.embeddings.create(
        model="text-embedding-3-large",
        input=query
    ).data[0].embedding

    results = search_client.search(
        search_text=query,
        vector_queries=[
            VectorizedQuery(vector=embedding, k_nearest_neighbors=top_k, fields="content_vector")
        ],
        top=top_k
    )
    return [{"id": r["id"], "title": r["title"], "score": r["@search.score"]} for r in results]

from azure.ai.contentsafety import ContentSafetyClient
from azure.ai.contentsafety.models import AnalyzeTextOptions, TextCategory

client = ContentSafetyClient(
    endpoint=os.environ["AZURE_CONTENT_SAFETY_ENDPOINT"],
    credential=os.environ["AZURE_CONTENT_SAFETY_KEY"]
)

def check_content_safety(text: str) -> dict:
    request = AnalyzeTextOptions(text=text)
    response = client.analyze_text(request)

    results = {}
    for category_result in response.categories_analysis:
        results[category_result.category] = {
            "severity": category_result.severity
        }

    return results

# Use as a guardrail before/after LLM calls
def safe_generate(prompt: str) -> str:
    # Check input
    input_safety = check_content_safety(prompt)
    if any(r["severity"] >= 4 for r in input_safety.values()):
        return "Content blocked: input contains unsafe content"

    # Generate response
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    output = response.choices[0].message.content

    # Check output
    output_safety = check_content_safety(output)
    if any(r["severity"] >= 4 for r in output_safety.values()):
        return "Content blocked: response contains unsafe content"

    return output

from openai import AzureOpenAI
import json

client = AzureOpenAI(
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
    api_version="2024-12-01-preview",
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"]
)

# Step 1: Prepare training data (JSONL format)
training_data = [
    {
        "messages": [
            {"role": "system", "content": "You classify federal procurement documents."},
            {"role": "user", "content": "Contract for IT services with ABC Corp, value $2.5M, period of performance 2026-2028."},
            {"role": "assistant", "content": "Category: IT Services\nValue: $2,500,000\nPOP: 2026-2028\nVendor: ABC Corp"}
        ]
    },
    # ... more examples
]

with open("training_data.jsonl", "w") as f:
    for example in training_data:
        f.write(json.dumps(example) + "\n")

# Step 2: Upload training file
file = client.files.create(
    file=open("training_data.jsonl", "rb"),
    purpose="fine-tune"
)

# Step 3: Create fine-tuning job
job = client.fine_tuning.jobs.create(
    model="gpt-4o-mini-2024-07-18",
    training_file=file.id,
    hyperparameters={"n_epochs": 3}
)

# Step 4: Monitor and deploy
# Fine-tuned model available as a deployment in Azure OpenAI