Skip to main content

Mistral

A multi-model classify-and-synthesize pipeline using Mistral's chat completion API. The orchestrator dispatches two child agents: a classifier backed by mistral-small-latest for fast intent detection, and a synthesizer backed by mistral-large-latest for detailed answer generation. Demonstrates tiered model selection via the @decision decorator.

Environment variables

This example requires MISTRAL_API_KEY, WAXELL_API_KEY, and WAXELL_API_URL. Use --dry-run to run without any API keys.

Architecture​

Key Code​

Tiered model selection with @decision​

The decision decorator records the model tier choice (small, large, or both) based on query complexity.

@waxell.decision(name="choose_model_tier", options=["small", "large", "both"])
def choose_model_tier(query_info: dict) -> dict:
    wc = query_info.get("word_count", 0)
    if wc < 8:
        chosen, reasoning = "small", "Simple query -- small model sufficient"
    elif wc < 20:
        chosen, reasoning = "both", "Medium complexity -- classify with small, synthesize with large"
    else:
        chosen, reasoning = "large", "Complex query -- large model needed for nuance"
    return {"chosen": chosen, "reasoning": reasoning, "confidence": 0.82}

Child agents with Mistral's async API​

Each child agent uses client.chat.complete_async() with Mistral's native Python SDK. The synthesizer evaluates its own output quality before returning.

@waxell.observe(agent_name="mistral-synthesizer", workflow_name="answer-synthesis", capture_io=True)
async def run_synthesizer(query: str, classification: str, client, *, dry_run=False, waxell_ctx=None) -> dict:
    waxell.tag("task", "synthesis")
    waxell.tag("model", "mistral-large-latest")

    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=[
            {"role": "system", "content": "You are a knowledgeable AI observability expert."},
            {"role": "user", "content": query},
        ],
    )
    answer = response.choices[0].message.content

    quality = evaluate_synthesis_quality(answer)
    waxell.score("synthesis_quality", quality["quality_score"])
    return {"answer": answer, "quality": quality, "model": response.model}

What this demonstrates​

  • @waxell.observe -- parent orchestrator with 2 child agents
  • @waxell.step_dec -- query preprocessing
  • @waxell.decision -- tiered model selection (small vs large)
  • @waxell.reasoning_dec -- synthesis quality evaluation
  • waxell.tag() -- task and model tagging
  • waxell.score() -- classification and synthesis quality scores
  • waxell.metadata() -- SDK and model metadata
  • Auto-instrumented Mistral LLM calls -- chat.complete_async traced automatically

Run it​

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

Source​

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