Sortie structurée

Le SDK peut renvoyer la réponse finale de l'agent sous forme d'instance de votre propre modèle Pydantic. Passez le modèle via output_type= : le champ output de l'enregistrement agents.Completed contient alors la réponse analysée, typée selon votre modèle. Une réponse non conforme est renvoyée sous forme de valeur agents.InvalidOutput, et non d'exception.

Utilisez cette fonctionnalité lorsque votre code traite la réponse : extraire les champs d'une facture pour remplir une ligne de base de données, classer un ticket d'assistance dans une catégorie avec une priorité, ou obtenir une décision d'approbation ou de refus accompagnée de son motif.

Définissez output_type= sur l'Agent, ou lors d'un appel à agent.run dans le même processus comme ci-dessous, pour demander au modèle de respecter ce format. Avec session.run, turn.result et un appel à agent.run sur un agent déployé, ce paramètre sert uniquement à analyser la réponse. Définissez-le sur l'Agent lorsque toutes les réponses ont la même structure.

import asyncio

from pydantic import BaseModel

from mistralai import agents


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


agent = agents.Agent(model="mistral-medium-latest")


async def main() -> None:
    match await agent.run("Where is the Eiffel Tower?", output_type=City):
        case agents.Completed(output=City() as city):
            print(city)
        case agents.Completed(output=agents.InvalidOutput(error=error)):
            print(f"not a City: {error}")
        case state:
            print(f"no answer: {state}")


asyncio.run(main())

import asyncio

from pydantic import BaseModel

from mistralai import agents


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


agent = agents.Agent(model="mistral-medium-latest")


async def main() -> None:
    match await agent.run("Where is the Eiffel Tower?", output_type=City):
        case agents.Completed(output=City() as city):
            print(city)
        case agents.Completed(output=agents.InvalidOutput(error=error)):
            print(f"not a City: {error}")
        case state:
            print(f"no answer: {state}")


asyncio.run(main())

name='Paris' country='France'

name='Paris' country='France'

Avec output_type=City, le champ output d'un enregistrement Completed est de type City | None | agents.InvalidOutput :

• une instance de City si la réponse est valide ;
• une valeur agents.InvalidOutput dans le cas contraire : raw contient le texte et error indique la cause ;
• None si le tour s'est terminé sans réponse finale, avec stop_reason égal à "iteration_limit", "skipped", "rejected" ou "output_limit".

Lorsqu'aucun type de sortie n'est défini sur l'appel ou sur l'Agent, output contient le texte de la réponse, ou "" si le tour n'en contient pas, quelle que soit la valeur de stop_reason.

text contient la réponse brute dans tous les cas. Les autres enregistrements, tels que Blocked ou Failed, n'ont pas de champ output et sont renvoyés tels quels.

Définissez output_type= sur l'Agent pour imposer un format à la réponse finale de chaque tour, dans chaque session ouverte par l'agent. C'est ainsi que vous définissez un type de sortie pour un agent déployé :

import asyncio

from pydantic import BaseModel

from mistralai import agents


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


agent = agents.Agent(model="mistral-medium-latest", output_type=City)


async def main() -> None:
    state = await agent.run("Where is the Eiffel Tower?")
    if isinstance(state, agents.Completed):
        print(repr(state.output))
        print(state.text)


asyncio.run(main())

import asyncio

from pydantic import BaseModel

from mistralai import agents


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


agent = agents.Agent(model="mistral-medium-latest", output_type=City)


async def main() -> None:
    state = await agent.run("Where is the Eiffel Tower?")
    if isinstance(state, agents.Completed):
        print(repr(state.output))
        print(state.text)


asyncio.run(main())

City(name='Paris', country='France')
{"name": "Paris", "country": "France"}

City(name='Paris', country='France')
{"name": "Paris", "country": "France"}

Dans le même processus, agent.run(..., output_type=SomeModel) remplace le type de sortie de l'Agent pour cette exécution.

Sans output_type= sur l'appel, le type statique de output est Any. Passez également le modèle à run() pour que votre vérificateur de types reconnaisse le type City.

agents serve, agents deploy et agent.to_workflow() utilisent le type de sortie de l'Agent. Un sous-agent n'en hérite jamais.

Un handle obtenu via agent.start_session() ou agent.session(session_id) conserve le paramètre output_type de l'Agent, tout comme ses tours : session.run() et turn.result() l'utilisent pour analyser la réponse si vous n'en passez aucun, et turn.get() l'utilise également.

session.run() et turn.result() acceptent aussi output_type= (voir Session.run), qui remplace celui de l'Agent pour cet appel. Utilisez-le lorsqu'un seul tour d'une session nécessite une réponse structurée.

Le modèle ne reçoit pas ce paramètre output_type : la session a été configurée à son démarrage. Demandez plutôt du JSON dans le prompt. À la fin du tour, le SDK tente d'analyser la réponse selon votre modèle. Si elle est valide, output contient une instance de ce modèle. Sinon, output contient une valeur agents.InvalidOutput, et text conserve la réponse :

import asyncio

from pydantic import BaseModel

from mistralai import agents


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


agent = agents.Agent(model="mistral-medium-latest")


async def main() -> None:
    session = await agent.start_session()
    state = await session.run(
        "Where is Big Ben? Answer only with a JSON object with name and country.", output_type=City
    )
    if isinstance(state, agents.Completed) and isinstance(state.output, City):
        print(f"{state.output.name} is in {state.output.country}.")
    elif isinstance(state, agents.Completed):
        print(f"No city in the answer: {state.text}")
    else:
        print(f"No answer: {state}")
    await session.cancel()


asyncio.run(main())

import asyncio

from pydantic import BaseModel

from mistralai import agents


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


agent = agents.Agent(model="mistral-medium-latest")


async def main() -> None:
    session = await agent.start_session()
    state = await session.run(
        "Where is Big Ben? Answer only with a JSON object with name and country.", output_type=City
    )
    if isinstance(state, agents.Completed) and isinstance(state.output, City):
        print(f"{state.output.name} is in {state.output.country}.")
    elif isinstance(state, agents.Completed):
        print(f"No city in the answer: {state.text}")
    else:
        print(f"No answer: {state}")
    await session.cancel()


asyncio.run(main())

Un modèle sans contrainte de format peut malgré tout placer son JSON dans un bloc de code, ce qui fait échouer l'analyse. Conservez donc le mécanisme de repli. Si tous les tours nécessitent la même structure, définissez output_type sur l'Agent : le modèle reçoit alors une demande de respecter ce format.

Un type de sortie n'empêche pas l'agent d'utiliser ses outils. Le modèle peut d'abord appeler des outils, puis fournir sa réponse finale selon votre type de sortie, à partir des résultats des outils. Par exemple, un agent doté d'un outil get_weather et de output_type=Forecast appelle l'outil, puis répond avec une instance de Forecast.

Cela s'applique à un type de sortie défini sur l'Agent ou lors d'un appel à agent.run dans le même processus : le SDK l'envoie avec les outils à chaque appel du modèle au cours du tour.

Pour que l'agent corrige une réponse non valide, ajoutez agents.pattern.hook.StructuredOutputHook[YourModel]() à son paramètre hooks=[...] : ce hook vérifie le JSON final et demande au modèle de réessayer.

Environnement : donnez à l'agent un ordinateur sur lequel exécuter des commandes.

Voir aussi :

• Exécution d'un agent : les enregistrements renvoyés par agent.run.
• Tours : lisez un tour avec turn.result.