Azure OpenAI

A multi-agent pipeline using Azure OpenAI Service with the standard OpenAI Python client pointed at Azure-specific endpoints. The orchestrator uses three-tier model selection (mini/standard/premium), dispatches a classifier child agent using gpt-4o-mini and a synthesizer child agent using the tier-selected model, then evaluates response quality with a reasoning decorator. Child agents are defined inside the main function for session ID closure capture.

Environment variables

This example requires OPENAI_API_KEY, OPENAI_API_BASE (Azure endpoint), WAXELL_API_KEY, and WAXELL_API_URL. Use --dry-run to run without any API keys.

Architecture

Key Code

Three-tier model selection

The decision decorator maps query complexity to Azure model tiers: mini for simple queries, standard for moderate, and premium for complex analysis.

@waxell_observe.decision(name="choose_model_tier", options=["mini", "standard", "premium"])
async def choose_model_tier(query_analysis: dict) -> dict:
    is_complex = query_analysis.get("is_complex", False)
    word_count = query_analysis.get("word_count", 0)

    if is_complex and word_count > 20:
        chosen = "premium"
        reasoning = f"Query is complex ({word_count} words). Using premium tier (gpt-4o)."
        confidence = 0.90
    elif is_complex:
        chosen = "standard"
        reasoning = f"Moderate complexity ({word_count} words). Using standard tier (gpt-4o)."
        confidence = 0.85
    else:
        chosen = "mini"
        reasoning = f"Straightforward ({word_count} words). Mini tier (gpt-4o-mini) sufficient."
        confidence = 0.92
    return {"chosen": chosen, "reasoning": reasoning, "confidence": confidence}

Child agents defined inside main for session closure

Child agents capture session_id and user context from the enclosing scope, allowing the @observe decorator to propagate context correctly.

@observe(agent_name="azure-classifier", workflow_name="classify-query",
         session_id=session, user_id=user_id, user_group=user_group,
         client=observe_client, enforce_policy=False)
async def azure_classifier(query: str) -> dict:
    waxell_observe.tag("provider", "azure_openai")
    waxell_observe.tag("role", "classifier")
    waxell_observe.metadata("model", "gpt-4o-mini")

    classify_response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Classify the query into: technical, general, opinion, or research."},
            {"role": "user", "content": query},
        ],
    )
    classification = classify_response.choices[0].message.content
    waxell_observe.score("classification_tokens",
                         float(classify_response.usage.prompt_tokens + classify_response.usage.completion_tokens))
    return {"classification": classification, "model": classify_response.model}

What this demonstrates

• @waxell.observe -- parent-child hierarchy with closure-captured session context
• @waxell.step_dec -- query preprocessing with complexity detection
• @waxell.decision -- three-tier model selection (mini/standard/premium)
• @waxell.reasoning_dec -- cross-agent response quality evaluation
• waxell.tag() -- Azure-specific provider tagging
• waxell.score() -- token count and quality scores
• waxell.metadata() -- model tier, token counts, and response details
• Azure OpenAI endpoints -- standard OpenAI client with Azure base_url and api_version

Run it

cd dev/waxell-dev
python -m app.demos.azure_openai_agent --dry-run

Source

dev/waxell-dev/app/demos/azure_openai_agent.py