🌊 Build a support agent with Mistral

import os
from mistralai.client import Mistral
from mistralai.client.models import TextChunk

# Split models by task — a small fast model for short classification and extraction, a general model for chat and tools
FAST_MODEL = "ministral-3-8b-2512"
AGENT_MODEL = "mistral-small-latest"

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])


def text_of(content) -> str:
    """Return only the final answer text from a response (skips thinking chunks in reasoning mode)"""
    if isinstance(content, str):
        return content
    return "".join(chunk.text for chunk in content or [] if isinstance(chunk, TextChunk))

import time
from llm import AGENT_MODEL, FAST_MODEL, client, text_of

PROMPT = "I ordered three days ago and shipping still hasn't started. What should I do?"

for model in (FAST_MODEL, AGENT_MODEL):
    started = time.perf_counter()
    resp = client.chat.complete(
        model=model,
        messages=[{"role": "user", "content": PROMPT}],
        temperature=0.1,
        max_tokens=300,
    )
    elapsed = time.perf_counter() - started
    usage = resp.usage
    print(f"[{model}] {elapsed:.2f}s · input {usage.prompt_tokens} / output {usage.completion_tokens} tokens")
    print(text_of(resp.choices[0].message.content)[:200], "\n")

from typing import Literal
from pydantic import BaseModel, Field
from llm import FAST_MODEL, client


class Ticket(BaseModel):
    category: Literal["billing", "delivery", "account", "bug", "other"]
    urgency: Literal["low", "medium", "high"]
    order_id: str | None = Field(default=None, description="The order number if the inquiry has one, otherwise null")
    summary: str = Field(description="One-sentence summary")


SYSTEM = """You classify customer inquiries for an online store.
category: billing (payments, refunds, receipts) / delivery (shipping, returns) / account (sign-in, profile) / bug (app or web errors) / other
urgency: high (financial loss, service unusable, legal action mentioned) / medium (complaints about delays) / low (simple questions)"""


def classify(text: str, model: str = FAST_MODEL) -> Ticket:
    resp = client.chat.parse(
        model=model,
        messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": text}],
        response_format=Ticket,
        temperature=0,
    )
    return resp.choices[0].message.parsed


if __name__ == "__main__":
    for text in [
        "I was charged twice. If it isn't refunded today, I'm blocking my card.",
        "Order A-1024 has been stuck in shipping for three days",
        "I'm not getting the password reset email",
    ]:
        print(classify(text).model_dump())

import json

# In a real service these call your order DB and payment API
ORDERS = {
    "A-1024": {"status": "in transit", "carrier": "UPS", "eta": "2026-10-07"},
    "A-2048": {"status": "paid", "carrier": None, "eta": None},
}


def get_order_status(order_id: str) -> dict:
    order = ORDERS.get(order_id.upper())
    return {"order_id": order_id, **order} if order else {"error": f"Order {order_id} not found"}


def request_refund(order_id: str, reason: str) -> dict:
    return {"order_id": order_id, "refund_ticket": f"RF-{order_id}", "status": "received", "reason": reason}


TOOLS = {"get_order_status": get_order_status, "request_refund": request_refund}

TOOL_SCHEMAS = [
    {"type": "function", "function": {
        "name": "get_order_status",
        "description": "Get an order's status, carrier and estimated delivery date by order number",
        "parameters": {"type": "object",
                       "properties": {"order_id": {"type": "string", "description": "Order number (e.g. A-1024)"}},
                       "required": ["order_id"]},
    }},
    {"type": "function", "function": {
        "name": "request_refund",
        "description": "File a refund only when the customer clearly asks for one",
        "parameters": {"type": "object",
                       "properties": {"order_id": {"type": "string"}, "reason": {"type": "string", "description": "Short refund reason"}},
                       "required": ["order_id", "reason"]},
    }},
]


def run_tool(name: str, raw_args) -> str:
    """Validate and run a call the model produced. The Mistral SDK may give arguments as a dict or a JSON string"""
    if name not in TOOLS:
        return json.dumps({"error": f"unknown tool: {name}"})
    try:
        args = raw_args if isinstance(raw_args, dict) else json.loads(raw_args or "{}")
        return json.dumps(TOOLS[name](**args))
    except (json.JSONDecodeError, TypeError) as e:
        return json.dumps({"error": f"bad arguments: {e}"})

from llm import AGENT_MODEL, client, text_of
from tools import TOOL_SCHEMAS, run_tool

SYSTEM = """You are a customer-support agent for an online store.
- Always check order status with the tools, and never make up anything the tool results do not say.
- File a refund only when the customer clearly asks for one.
- Answer politely in three sentences or fewer."""


def answer(question: str, reasoning: str = "none", max_steps: int = 4) -> str:
    messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": question}]
    for _ in range(max_steps):
        resp = client.chat.complete(
            model=AGENT_MODEL,
            messages=messages,
            tools=TOOL_SCHEMAS,
            tool_choice="auto",
            parallel_tool_calls=False,    # one at a time — keeps logs and error handling simple
            reasoning_effort=reasoning,
            temperature=0.1,
        )
        msg = resp.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:
            return text_of(msg.content)

        for tc in msg.tool_calls:
            result = run_tool(tc.function.name, tc.function.arguments)
            print(f"[tool] {tc.function.name}({tc.function.arguments}) -> {result}")
            messages.append({"role": "tool", "name": tc.function.name, "content": result, "tool_call_id": tc.id})
    return "This is taking longer than expected. A support agent will follow up shortly."


if __name__ == "__main__":
    print(answer("When will order A-1024 arrive?"))

from agent import answer
from classify import Ticket, classify


def needs_reasoning(ticket: Ticket) -> bool:
    """Reasoning mode is expensive — use it only for hard inquiries such as financial loss or error analysis"""
    return ticket.urgency == "high" or ticket.category == "bug"


def handle(text: str) -> dict:
    ticket = classify(text)
    effort = "high" if needs_reasoning(ticket) else "none"
    reply = answer(text, reasoning=effort)
    return {"ticket": ticket.model_dump(), "reasoning": effort, "reply": reply}


if __name__ == "__main__":
    for text in ["When will order A-1024 arrive?", "I was charged twice. Please refund order A-2048."]:
        print(handle(text), "\n")

import base64
import mimetypes
import sys
from pydantic import BaseModel, Field
from llm import FAST_MODEL, client


class Receipt(BaseModel):
    merchant: str
    paid_at: str | None = Field(default=None, description="Payment date and time (YYYY-MM-DD HH:MM)")
    total: float = Field(description="Total amount paid")
    items: list[str]


def to_data_url(path: str) -> str:
    mime = mimetypes.guess_type(path)[0] or "image/jpeg"
    with open(path, "rb") as f:
        return f"data:{mime};base64,{base64.b64encode(f.read()).decode()}"


def read_receipt(path: str) -> Receipt:
    # 1) OCR: image → Markdown
    ocr = client.ocr.process(
        model="mistral-ocr-latest",
        document={"type": "image_url", "image_url": to_data_url(path)},
    )
    markdown = "\n\n".join(page.markdown for page in ocr.pages)

    # 2) Structure: Markdown → Receipt
    resp = client.chat.parse(
        model=FAST_MODEL,
        messages=[
            {"role": "system", "content": "Extract receipt details from OCR output. Use null for missing values."},
            {"role": "user", "content": markdown},
        ],
        response_format=Receipt,
        temperature=0,
    )
    return resp.choices[0].message.parsed


if __name__ == "__main__":
    print(read_receipt(sys.argv[1] if len(sys.argv) > 1 else "receipt.jpg").model_dump())

from ollama import chat
from classify import SYSTEM, Ticket


def classify_local(text: str, model: str = "ministral-3") -> Ticket:
    """Classify with the same Ticket schema on local Ministral 3 — inquiry text never leaves your machine"""
    resp = chat(
        model=model,
        messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": text}],
        format=Ticket.model_json_schema(),
        options={"temperature": 0},
    )
    return Ticket.model_validate_json(resp.message.content)


if __name__ == "__main__":
    print(classify_local("Order A-1024 has been stuck in shipping for three days").model_dump())

{"text": "I was charged twice. If it isn't refunded today, I'm blocking my card.", "category": "billing", "urgency": "high"}
{"text": "Order A-1024 has been stuck in shipping for three days", "category": "delivery", "urgency": "medium"}
{"text": "I'm not getting the password reset email", "category": "account", "urgency": "low"}
{"text": "The app freezes on a white screen when I tap the pay button", "category": "bug", "urgency": "high"}

import json
import statistics
import time
from classify import classify
from local import classify_local

cases = [json.loads(line) for line in open("data/labeled.jsonl", encoding="utf-8")]


def evaluate(name: str, fn) -> None:
    hits_category = hits_urgency = 0
    latencies = []
    for case in cases:
        started = time.perf_counter()
        ticket = fn(case["text"])
        latencies.append(time.perf_counter() - started)
        hits_category += ticket.category == case["category"]
        hits_urgency += ticket.urgency == case["urgency"]
        if ticket.category != case["category"]:
            print(f"  ✗ [{name}] {case['text'][:40]}… expected {case['category']} / got {ticket.category}")
    n = len(cases)
    print(f"{name}: category {hits_category}/{n}, urgency {hits_urgency}/{n}, "
          f"latency p50 {statistics.median(latencies):.2f}s / max {max(latencies):.2f}s")


evaluate("API ministral-3-8b", classify)
evaluate("local ministral-3", classify_local)

import logging
import time
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from router import handle

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("support-agent")

app = FastAPI(title="Support Agent")


class Inquiry(BaseModel):
    text: str = Field(min_length=1, max_length=2000)


@app.post("/inquiries")
def create_inquiry(inquiry: Inquiry) -> dict:
    started = time.perf_counter()
    try:
        result = handle(inquiry.text)
    except Exception:
        log.exception("inquiry failed")
        raise HTTPException(status_code=502, detail="AI processing failed. Connecting you to a human agent.")
    ticket = result["ticket"]
    log.info("category=%s urgency=%s reasoning=%s elapsed=%.2fs",
             ticket["category"], ticket["urgency"], result["reasoning"], time.perf_counter() - started)
    return result