Valider une sortie structurée avec un hook

agents.pattern.hook.StructuredOutputHook valide la réponse finale de chaque tour avec votre modèle Pydantic. Si la réponse n'est pas valide, le hook transmet l'erreur à l'agent, qui corrige sa réponse.

Le hook ne garantit pas une réponse valide. Il cesse de demander des corrections après max_retries, ou lorsque le tour a épuisé les itérations autorisées par max_iterations. Il accepte alors la dernière réponse, ou fait échouer le tour si vous définissez on_limit="reject".

Cet exemple présente : un hook qui vérifie la réponse finale de chaque tour et demande à l'agent de la corriger, output_type=, et une application qui désactive le hook.

Exécuter : placez le bloc de code ci-dessous dans un fichier, structured_output_hook.py, puis exécutez python structured_output_hook.py (configuration).

Avec les fournisseurs "mistral" et "openai", output_type= impose déjà une réponse au format JSON conforme au schéma de votre modèle. Utilisez le hook dans les cas suivants :

• Votre modèle comporte des règles qu'un schéma JSON ne peut pas exprimer, comme un @model_validator.
• Vous validez sans output_type= : avec le fournisseur "anthropic", un fournisseur de complétion Personnalisé, ou un sous-agent.

Le hook ne lit pas output_type=. Vous pouvez l'utiliser avec ou sans cet argument.

Indiquez votre modèle Pydantic entre crochets et passez les autres paramètres au constructeur : agents.pattern.hook.StructuredOutputHook[City](max_retries=2).

Les Références de StructuredOutputHook présentent sa signature complète.

Avec output_type=, définissez schema_in_prompt=False : le fournisseur transmet déjà le schéma au modèle. Si vous utilisez uniquement le hook, conservez True, ou définissez False et intégrez le schéma à vos propres instructions.

Avec "accept", l'exécution renvoie la dernière réponse, et votre application décide comment la traiter. Avec output_type=, agent.run() valide également cette réponse. Si elle n'est pas valide, le champ output de l'enregistrement Completed contient un agents.InvalidOutput avec la réponse et l'erreur Pydantic.

Avec "reject", le tour échoue : agent.run() renvoie agents.Failed avec le code capability_execution_failed, avec ou sans output_type=. Le message indique quelle vérification a échoué et fournit l'erreur Pydantic ainsi que les 1 000 premiers caractères de la dernière réponse. La session reste utilisable pour le tour suivant.

Après le déploiement de l'agent avec Studio (déploiements Auto-hébergés ou gérés par Mistral), l'historique de session affiche aussi chaque réponse refusée par le hook, sous la forme d'un message de l'assistant identique aux autres.

Créer un seul objet hook et conserver une référence à cet objet permet à votre application de le désactiver et de le réactiver avec enabled. Dans le même processus, la modification s'applique au prochain appel à agent.run() ou à la prochaine session : une session conserve les valeurs définies à son démarrage. Un agent déployé utilise la valeur définie lors de son déploiement.

L'exemple utilise on_limit="reject" : si la réponse n'est toujours pas un City valide, le résultat renvoyé est agents.Failed :

"""Validate structured output with a hook that the application can switch off.

Set MISTRAL_API_KEY, then run: python structured_output_hook.py
"""

import asyncio

from pydantic import BaseModel

from mistralai import agents


class City(BaseModel):
    name: str
    country: str


@agents.tool
async def lookup_landmark(name: str) -> dict[str, str]:
    """Find where a landmark is."""
    return {"landmark": name, "location": "Champ de Mars, 7th arrondissement, Paris, France"}


async def main() -> None:
    # The hook makes the agent correct a response that is not a valid City. output_type= below
    # already gives the model the schema, so the hook does not add it to the prompt.
    check_city = agents.pattern.hook.StructuredOutputHook[City](
        max_retries=2, on_limit="reject", schema_in_prompt=False
    )
    agent = agents.Agent(
        model="mistral-medium-latest",
        name="structured-output-hook-example",
        instructions="Use lookup_landmark to find where a landmark is.",
        max_iterations=6,
        tools=[lookup_landmark],
        hooks=[check_city],
    )

    # output_type=City parses the final response into Completed.output.
    match await agent.run("Which city and country contain the Eiffel Tower?", output_type=City):
        case agents.Completed(output=City() as city):
            print(city.model_dump())
        case agents.Failed(error=error) if error.code == "capability_execution_failed":
            # on_limit="reject": still not a valid City after max_retries corrections.
            raise RuntimeError(f"No valid City: {error.message}")
        case state:
            raise RuntimeError(f"The run did not complete: {state!r}")

    # The application switches the hook off for a free-text answer.
    check_city.enabled = False
    match await agent.run("Describe the Eiffel Tower in one sentence."):
        case agents.Completed(text=text) if text:
            print(text)
        case state:
            raise RuntimeError(f"The run did not complete: {state!r}")


if __name__ == "__main__":
    asyncio.run(main())

"""Validate structured output with a hook that the application can switch off.

Set MISTRAL_API_KEY, then run: python structured_output_hook.py
"""

import asyncio

from pydantic import BaseModel

from mistralai import agents


class City(BaseModel):
    name: str
    country: str


@agents.tool
async def lookup_landmark(name: str) -> dict[str, str]:
    """Find where a landmark is."""
    return {"landmark": name, "location": "Champ de Mars, 7th arrondissement, Paris, France"}


async def main() -> None:
    # The hook makes the agent correct a response that is not a valid City. output_type= below
    # already gives the model the schema, so the hook does not add it to the prompt.
    check_city = agents.pattern.hook.StructuredOutputHook[City](
        max_retries=2, on_limit="reject", schema_in_prompt=False
    )
    agent = agents.Agent(
        model="mistral-medium-latest",
        name="structured-output-hook-example",
        instructions="Use lookup_landmark to find where a landmark is.",
        max_iterations=6,
        tools=[lookup_landmark],
        hooks=[check_city],
    )

    # output_type=City parses the final response into Completed.output.
    match await agent.run("Which city and country contain the Eiffel Tower?", output_type=City):
        case agents.Completed(output=City() as city):
            print(city.model_dump())
        case agents.Failed(error=error) if error.code == "capability_execution_failed":
            # on_limit="reject": still not a valid City after max_retries corrections.
            raise RuntimeError(f"No valid City: {error.message}")
        case state:
            raise RuntimeError(f"The run did not complete: {state!r}")

    # The application switches the hook off for a free-text answer.
    check_city.enabled = False
    match await agent.run("Describe the Eiffel Tower in one sentence."):
        case agents.Completed(text=text) if text:
            print(text)
        case state:
            raise RuntimeError(f"The run did not complete: {state!r}")


if __name__ == "__main__":
    asyncio.run(main())

La première exécution a renvoyé un City valide. Avec le hook désactivé, la seconde a renvoyé du texte libre.

• Sortie structurée : l'argument output_type=.
• API Hooks : points d'exécution des hooks, valeurs de retour possibles pour chacun et état.