Mistral AI agent circuit breaker: stopping loop failures in Mistral Large and Mixtral function-calling agents

import asyncio
from mistralai import Mistral
from runguard import guard_async, LoopDetectedError, BudgetExceededError

client = Mistral(api_key="YOUR_MISTRAL_API_KEY")

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_database",
            "description": "Search the product database for records matching a query.",
            "parameters": {
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "summarize_results",
            "description": "Summarize a list of database records into a human-readable report.",
            "parameters": {
                "type": "object",
                "properties": {"records": {"type": "array", "items": {"type": "string"}}},
                "required": ["records"],
            },
        },
    },
]


async def _call_mistral(messages: list, **kwargs) -> dict:
    response = await client.chat.complete_async(
        model="mistral-large-latest",
        messages=messages,
        tools=TOOLS,
        tool_choice="auto",
        **kwargs,
    )
    choice = response.choices[0]
    usage = response.usage

    # Mistral Large 2 pricing: $2/$6 per 1M input/output tokens
    input_tokens = usage.prompt_tokens if usage else 0
    output_tokens = usage.completion_tokens if usage else 0
    usd = (input_tokens * 2.0 + output_tokens * 6.0) / 1_000_000

    # Signature: first tool-call name, or "end_turn"
    tool_calls = getattr(choice.message, "tool_calls", None) or []
    sig = tool_calls[0].function.name if tool_calls else "end_turn"

    return {"choice": choice, "usd": usd, "sig": sig}


# Wrap once — reuse across iterations of the agent loop
guarded_mistral = guard_async(
    _call_mistral,
    budget={"max_usd": 2.0},
    loop={"repeats": 3, "max_cycle_len": 4},
)


async def run_mistral_agent(user_query: str) -> str:
    messages = [
        {"role": "system", "content": "You are a research assistant. Use the provided tools to answer questions."},
        {"role": "user", "content": user_query},
    ]

    while True:
        try:
            result = await guarded_mistral(messages)
        except LoopDetectedError as e:
            return f"Loop detected after {len(messages)} messages. Pattern: {e.pattern!r}. Aborting."
        except BudgetExceededError as e:
            return f"Budget cap reached (${e.spent:.4f}). Aborting after {len(messages)} messages."

        choice = result["choice"]
        messages.append(choice.message)

        tool_calls = getattr(choice.message, "tool_calls", None) or []
        if not tool_calls:
            # No more tool calls — return final answer
            return choice.message.content or ""

        for tc in tool_calls:
            fn_name = tc.function.name
            fn_args = tc.function.arguments  # JSON string from Mistral

            try:
                tool_result = await dispatch_tool(fn_name, fn_args)
            except Exception as e:
                # Raise exceptions as tool results, never return error strings
                tool_result = f"__tool_error__: {type(e).__name__}: {e}"

            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": str(tool_result),
            })


async def dispatch_tool(name: str, args_json: str) -> str:
    import json
    args = json.loads(args_json)
    if name == "search_database":
        return await search_database(args["query"])
    elif name == "summarize_results":
        return await summarize_results(args["records"])
    raise ValueError(f"Unknown tool: {name}")


# Stub implementations
async def search_database(query: str) -> str: ...
async def summarize_results(records: list) -> str: ...