Tutorial: Building AIP-Equivalent AI Workflows on Azure¶

# test_openai.py
import os
from openai import AzureOpenAI

client = AzureOpenAI(
    azure_endpoint=os.environ["AOAI_ENDPOINT"],
    api_key=os.environ["AOAI_KEY"],
    api_version="2024-06-01",
)

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Summarize what Azure OpenAI is in two sentences."},
    ],
    temperature=0.3,
    max_tokens=200,
)

print(response.choices[0].message.content)

# grounded_chatbot.py
import asyncio
import os
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
from semantic_kernel.agents import ChatCompletionAgent
from semantic_kernel.contents import ChatHistory
from semantic_kernel.functions import kernel_function

# --- Plugin: Azure AI Search retrieval ---

class SearchPlugin:
    """Retrieve grounding context from Azure AI Search."""

    def __init__(self):
        from azure.search.documents import SearchClient
        from azure.core.credentials import AzureKeyCredential

        self.client = SearchClient(
            endpoint=os.environ["SEARCH_ENDPOINT"],
            index_name=os.environ["SEARCH_INDEX"],
            credential=AzureKeyCredential(os.environ["SEARCH_KEY"]),
        )

    @kernel_function(
        name="search_knowledge_base",
        description="Search the knowledge base for documents relevant to a query.",
    )
    def search(self, query: str) -> str:
        results = self.client.search(search_text=query, top=5)
        hits = []
        for r in results:
            hits.append(f"[{r['title']}]: {r['content'][:500]}")
        return "\n\n---\n\n".join(hits) if hits else "No relevant documents found."


# --- Kernel and Agent setup ---

async def main():
    kernel = Kernel()
    kernel.add_service(
        AzureChatCompletion(
            service_id="gpt4o",
            deployment_name="gpt-4o",
            endpoint=os.environ["AOAI_ENDPOINT"],
            api_key=os.environ["AOAI_KEY"],
        )
    )
    kernel.add_plugin(SearchPlugin(), plugin_name="Search")

    agent = ChatCompletionAgent(
        kernel=kernel,
        service_id="gpt4o",
        name="DataAssistant",
        instructions=(
            "You are a data platform assistant. Use the Search plugin to find "
            "relevant documents before answering. Always cite your sources. "
            "If no relevant documents are found, say so."
        ),
    )

    chat = ChatHistory()
    print("Grounded Chatbot (type 'quit' to exit)")
    print("-" * 50)

    while True:
        user_input = input("\nYou: ").strip()
        if user_input.lower() in ("quit", "exit"):
            break
        chat.add_user_message(user_input)
        async for response in agent.invoke(chat):
            print(f"\nAssistant: {response.content}")

asyncio.run(main())

# create_vector_index.py
import os
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import (
    SearchIndex,
    SearchField,
    SearchFieldDataType,
    SimpleField,
    SearchableField,
    VectorSearch,
    HnswAlgorithmConfiguration,
    VectorSearchProfile,
    SemanticConfiguration,
    SemanticSearch,
    SemanticPrioritizedFields,
    SemanticField,
)
from azure.core.credentials import AzureKeyCredential

client = SearchIndexClient(
    endpoint=os.environ["SEARCH_ENDPOINT"],
    credential=AzureKeyCredential(os.environ["SEARCH_KEY"]),
)

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

index = SearchIndex(
    name="knowledge-base",
    fields=fields,
    vector_search=VectorSearch(
        algorithms=[HnswAlgorithmConfiguration(name="hnsw-config")],
        profiles=[VectorSearchProfile(
            name="vector-profile",
            algorithm_configuration_name="hnsw-config",
        )],
    ),
    semantic_search=SemanticSearch(configurations=[
        SemanticConfiguration(
            name="semantic-config",
            prioritized_fields=SemanticPrioritizedFields(
                title_field=SemanticField(field_name="title"),
                content_fields=[SemanticField(field_name="content")],
            ),
        ),
    ]),
)

result = client.create_or_update_index(index)
print(f"Index '{result.name}' created successfully")

# index_documents.py
import os
import hashlib
from openai import AzureOpenAI
from azure.search.documents import SearchClient
from azure.core.credentials import AzureKeyCredential

oai = AzureOpenAI(
    azure_endpoint=os.environ["AOAI_ENDPOINT"],
    api_key=os.environ["AOAI_KEY"],
    api_version="2024-06-01",
)

search = SearchClient(
    endpoint=os.environ["SEARCH_ENDPOINT"],
    index_name="knowledge-base",
    credential=AzureKeyCredential(os.environ["SEARCH_KEY"]),
)


def chunk_text(text: str, chunk_size: int = 800, overlap: int = 200) -> list[str]:
    """Split text into overlapping chunks."""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start += chunk_size - overlap
    return chunks


def embed_and_index(title: str, content: str, source: str) -> int:
    """Chunk a document, generate embeddings, and upload to the index."""
    chunks = chunk_text(content)
    documents = []
    for i, chunk in enumerate(chunks):
        doc_id = hashlib.md5(f"{title}_{i}".encode()).hexdigest()
        embedding = oai.embeddings.create(
            model="text-embedding-3-large", input=chunk
        ).data[0].embedding
        documents.append({
            "id": doc_id,
            "title": f"{title} (part {i + 1})",
            "content": chunk,
            "source": source,
            "content_vector": embedding,
        })
    result = search.upload_documents(documents)
    return sum(1 for r in result if r.succeeded)


# Example: index a batch of documents
sample_docs = [
    {"title": "Data Governance Policy", "content": "All data products must...", "source": "policies/governance.md"},
    {"title": "Case Management SOP", "content": "Federal cases are tracked...", "source": "sops/case-mgmt.md"},
]

for doc in sample_docs:
    count = embed_and_index(doc["title"], doc["content"], doc["source"])
    print(f"Indexed {count} chunks for '{doc['title']}'")

# rag_query.py
import os
from openai import AzureOpenAI
from azure.search.documents import SearchClient
from azure.search.documents.models import VectorizableTextQuery, QueryType
from azure.core.credentials import AzureKeyCredential

oai = AzureOpenAI(
    azure_endpoint=os.environ["AOAI_ENDPOINT"],
    api_key=os.environ["AOAI_KEY"],
    api_version="2024-06-01",
)

search = SearchClient(
    endpoint=os.environ["SEARCH_ENDPOINT"],
    index_name="knowledge-base",
    credential=AzureKeyCredential(os.environ["SEARCH_KEY"]),
)


def rag_query(question: str) -> str:
    """Search -> Augment -> Generate."""
    # 1. Hybrid search (vector + keyword + semantic reranking)
    results = search.search(
        search_text=question,
        vector_queries=[
            VectorizableTextQuery(
                text=question,
                k_nearest_neighbors=5,
                fields="content_vector",
            )
        ],
        query_type=QueryType.SEMANTIC,
        semantic_configuration_name="semantic-config",
        top=5,
    )

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

    # 3. Generate grounded answer
    response = oai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "Answer using ONLY the provided context. "
                    "Cite sources by title. If the context does not "
                    "contain the answer, say so."
                ),
            },
            {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
        ],
        temperature=0.2,
        max_tokens=1024,
    )
    return response.choices[0].message.content


# Test
answer = rag_query("What are the data governance requirements?")
print(answer)

# function_app.py -- Azure Functions v2 (Python)
import azure.functions as func
import json
import os
from openai import AzureOpenAI

app = func.FunctionApp()

client = AzureOpenAI(
    azure_endpoint=os.environ["AOAI_ENDPOINT"],
    api_key=os.environ["AOAI_KEY"],
    api_version="2024-06-01",
)

CATEGORIES = ["billing", "technical", "account", "security", "general"]


@app.function_name("ClassifyDocument")
@app.route(route="classify", methods=["POST"])
def classify_document(req: func.HttpRequest) -> func.HttpResponse:
    """Classify a document into a category using Azure OpenAI."""
    body = req.get_json()
    text = body.get("text", "")

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    f"Classify the text into exactly one category: {CATEGORIES}. "
                    "Return JSON: {\"category\": \"...\", \"confidence\": 0.0-1.0, "
                    "\"reasoning\": \"...\"}"
                ),
            },
            {"role": "user", "content": text},
        ],
        temperature=0.0,
        max_tokens=256,
        response_format={"type": "json_object"},
    )

    result = json.loads(response.choices[0].message.content)
    return func.HttpResponse(json.dumps(result), mimetype="application/json")


@app.function_name("ExtractEntities")
@app.route(route="extract", methods=["POST"])
def extract_entities(req: func.HttpRequest) -> func.HttpResponse:
    """Extract named entities from text using Azure OpenAI."""
    body = req.get_json()
    text = body.get("text", "")

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "Extract all named entities from the text. Return JSON: "
                    "{\"entities\": [{\"text\": \"...\", \"type\": \"PERSON|ORG|LOCATION|DATE\"}]}"
                ),
            },
            {"role": "user", "content": text},
        ],
        temperature=0.0,
        max_tokens=512,
        response_format={"type": "json_object"},
    )

    result = json.loads(response.choices[0].message.content)
    return func.HttpResponse(json.dumps(result), mimetype="application/json")


@app.function_name("SummarizeDocument")
@app.route(route="summarize", methods=["POST"])
def summarize_document(req: func.HttpRequest) -> func.HttpResponse:
    """Summarize a document using Azure OpenAI."""
    body = req.get_json()
    text = body.get("text", "")
    max_sentences = body.get("max_sentences", 3)

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": f"Summarize the text in {max_sentences} sentences or fewer.",
            },
            {"role": "user", "content": text},
        ],
        temperature=0.3,
        max_tokens=512,
    )

    return func.HttpResponse(
        json.dumps({"summary": response.choices[0].message.content}),
        mimetype="application/json",
    )

# evaluate_rag.py
import json
import os
from openai import AzureOpenAI

client = AzureOpenAI(
    azure_endpoint=os.environ["AOAI_ENDPOINT"],
    api_key=os.environ["AOAI_KEY"],
    api_version="2024-06-01",
)


def evaluate_response(question: str, response: str, ground_truth: str, context: str) -> dict:
    """Evaluate a RAG response on groundedness, relevance, and coherence."""
    eval_prompt = f"""Evaluate the following AI response on three metrics.
Score each metric from 1 (worst) to 5 (best).

Question: {question}
Context provided: {context}
AI Response: {response}
Expected Answer: {ground_truth}

Return JSON:
{{
    "groundedness": <1-5>,
    "relevance": <1-5>,
    "coherence": <1-5>,
    "groundedness_reason": "...",
    "relevance_reason": "...",
    "coherence_reason": "..."
}}

Scoring guide:
- Groundedness: Is the response supported by the context? (5 = fully grounded, 1 = hallucinated)
- Relevance: Does the response answer the question? (5 = fully relevant, 1 = off-topic)
- Coherence: Is the response well-structured and clear? (5 = clear, 1 = incoherent)"""

    result = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are an AI evaluation judge. Return only valid JSON."},
            {"role": "user", "content": eval_prompt},
        ],
        temperature=0.0,
        max_tokens=512,
        response_format={"type": "json_object"},
    )
    return json.loads(result.choices[0].message.content)


def run_evaluation(dataset_path: str, rag_fn) -> dict:
    """Run evaluation over a dataset and compute aggregate metrics."""
    with open(dataset_path) as f:
        test_cases = json.load(f)

    results = []
    for case in test_cases:
        # Generate a response using your RAG pipeline
        response = rag_fn(case["question"])

        # Evaluate the response
        scores = evaluate_response(
            question=case["question"],
            response=response,
            ground_truth=case["ground_truth"],
            context=case["context"],
        )
        scores["question"] = case["question"]
        results.append(scores)
        print(f"  Q: {case['question'][:60]}...")
        print(f"    Groundedness={scores['groundedness']} "
              f"Relevance={scores['relevance']} "
              f"Coherence={scores['coherence']}")

    # Aggregate metrics
    avg = lambda metric: sum(r[metric] for r in results) / len(results)
    summary = {
        "total_cases": len(results),
        "avg_groundedness": round(avg("groundedness"), 2),
        "avg_relevance": round(avg("relevance"), 2),
        "avg_coherence": round(avg("coherence"), 2),
        "details": results,
    }

    print(f"\n{'='*50}")
    print(f"Evaluation Summary ({summary['total_cases']} cases)")
    print(f"  Groundedness: {summary['avg_groundedness']}/5")
    print(f"  Relevance:    {summary['avg_relevance']}/5")
    print(f"  Coherence:    {summary['avg_coherence']}/5")

    return summary


# Usage with the RAG query function from Step 3:
# from rag_query import rag_query
# run_evaluation("eval_dataset.json", rag_query)