Créer un agent d’IA de zéro en Python : la boucle de jeu, étape par étape
Créez un agent capable d’appeler des outils en Python pur, doté de concepts tels que les objectifs, les actions, la mémoire et l’environnement, puis ajoutez un registre d’outils, une injection de dépendances cachée et des sorties typées.
Les frameworks d’agents tels que Pydantic AI, AutoGen et Microsoft Agent Framework sont utiles pour mettre rapidement en marche un prototype. Cependant, après en avoir utilisé plusieurs, quelques points de friction apparaissent généralement.
- L’écosystème est encore en évolution. Chaque framework couvre certaines des fonctionnalités nécessaires tout en omettant d’autres, et un autre framework dispose des éléments manquants au premier. C’est normal pour un domaine encore jeune, et avec le temps la plupart d’entre eux permettront de gérer les cas d’usage courants.
- L’abstraction s’accumule. Les frameworks visent à servir une grande variété de conceptions d’agents, ce qui entraîne l’ajout de couches difficiles à suivre et encore plus difficiles à déboguer. LangChain en est un exemple typique : lorsque le système ne fonctionne pas correctement, ces couches entravent la détection de la cause du problème.
Le côté encourageant, c’est qu’un agent n’a pas besoin d’être complexe. Pour de nombreuses tâches, une implémentation simple suffit. En le développant à partir de zéro, vous avez un contrôle total sur ce qui est envoyé au modèle, sur la manière dont les décisions sont prises et sur la gestion des coûts ; en connaissant les principes de base, il devient plus facile de faire confiance à tout framework ultérieur. Ce qui suit décrit la construction étape par étape d’un agent Python fonctionnel à partir de principes fondamentaux.
L’idée centrale : la boucle de l’agent
Commencez par simplifier l’idée. Un agent IA est un grand modèle de langage capable d’effectuer des actions et d’utiliser leurs résultats pour décider de ce à faire ensuite. Dans une interaction classique avec un LLM, vous envoyez une demande et vous recevez une seule réponse. Un agent, lui, fonctionne en boucle : il examine l’état actuel, choisit une action, la met en œuvre, reçoit des retours, puis recommence jusqu’à ce que la tâche soit terminée.
GAME : une conception en quatre parties
Pour construire cette boucle, nous utiliserons la structure GAME, qui divise l’agent en Objectifs, Actions, Mémoire et Environnement. Elle offre un modèle mental compact et permet au code de rester modulaire et facile à étendre.
Objectif
L’objectif définit ce que l’agent cherche à accomplir. Il contient la description de la tâche ainsi que toutes les instructions guidant son comportement, et devient la demande système.
class Goal(BaseModel):
name: str
description: str
Action (outil)
La base de code ici l’appelle un Tool, mais ces termes signifient la même chose. Une action est quelque chose que l’agent peut faire : appeler une API, rechercher des documents, interroger une base de données ou effectuer un calcul.
class Tool(BaseModel):
name: str
description: str
parameters: Dict[str, Any]
function: Callable
tags: Optional[List[str]] = None
terminal: bool = False
Les API de modèles de langage actuels, y compris celles d’OpenAI, acceptent des définitions d’outils en même temps que la requête. Ces définitions doivent être écrites sous forme de JSON Schema, indiquant le nom de chaque outil, ses paramètres ainsi que les entrées attendues.
Comme premier outil, prenons une fonction qui additionne deux nombres :
def add(a: float, b: float) -> float:
return a + b
Ensuite, décrivez-le sous une forme que le modèle peut lire :
number_pair = {
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"},
},
"required": ["a", "b"],
}
add_tool = Tool(
name="add",
description="Adds two numbers.",
parameters=number_pair,
function=add,
)
Lorsque la requête est constituée, ces objets d’action sont transformés en définitions d’outils et envoyés au modèle. Le modèle ne lance jamais la fonction elle-même ; il se contente de déterminer quel outil appeler et de générer les arguments nécessaires.
Supposons que l’utilisateur demande : « Quel est le résultat de 10 + 20 ? » Le modèle pourrait répondre quelque chose comme ceci :
{
"name": "add",
"arguments": {
"a": 10,
"b": 20
}
}
Le choix de l’action et la fourniture de ses arguments constituent toute la tâche du modèle.
Envoi des actions au modèle
Cet outil transmet l’historique de la conversation ainsi que les actions disponibles au LLM :
def generate_response(memory: Memory, tools: List[Dict[str, Any]]) -> Any:
from litellm import completion
messages = [m.model_dump(exclude_none=True) for m in memory.get()]
return completion(
model="openai/gpt-4o",
messages=messages,
max_tokens=1024,
tools=tools,
tool_choice="required",
parallel_tool_calls=False,
)
tools = [
{
"type": "function",
"function": {
"name": add_tool.name,
"description": add_tool.description,
"parameters": add_tool.parameters,
}
}
]
result = generate_response(
memory=Memory(messages=[
Message(role=Role.SYSTEM, content="Perform arithmetic using the add and multiply tools. "),
Message(role=Role.USER, content="Add 200 and 10, then multiply the result by 2.")]),
tools=tools
)
print(json.dumps(result.choices[0].message.model_dump(exclude_none=True), indent=2))
# Result:
{
"role": "assistant",
"tool_calls": [
{
"function": {
"arguments": "{\"a\":200,\"b\":10}",
"name": "add"
},
"id": "call_ZDlLQxVB9WLY7fPHwRipeeM1",
"type": "function"
}
],
"provider_specific_fields": {
"refusal": null
},
"annotations": []
}
L’agent exécute ensuite la fonction Python de correspondance et transmet le résultat à l’étape suivante du cycle.
Mémoire
La mémoire conserve ce que les itérations précédentes ont collecté, permettant ainsi à l’agent de prendre en compte les observations antérieures lorsqu’il choisit sa prochaine action.
Pour l’instant, elle reste minimale : les nouveaux messages sont simplement ajoutés à une liste. Les fenêtres de contexte des LLM étant finies, un système réel a besoin d’un mécanisme plus intelligent pour éliminer ou résumer les informations, ce que cet exemple omet.
class Memory(BaseModel):
"""M in GAME: the agent's conversational record - the running list of
Messages the LLM sees each turn. A thin, swappable store so strategies
(windowing, summarization, persistence) can vary later without touching
Agent.run; for now it is just an append-and-read list."""
messages: List[Message] = []
def add(self, message: Message) -> None:
self.messages.append(message)
def get(self) -> List[Message]:
return self.messages
Un modèle Message représente chaque interaction au sein du cycle. Les observations, les résultats des outils, les entrées de l’utilisateur ainsi que les réponses de l’assistant se transforment tous en messages. Sa structure suit étroitement le format des messages des API de modèles de langage actuels, ce qui permet de créer des prompts à partir de l’historique de manière simple et d’étendre la mémoire de façon aisée.
from enum import Enum
class Role(str, Enum):
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"
TOOL= "tool"
class Message(BaseModel):
role: str
content: Optional[str] = None
Environnement
On pourrait se demander pourquoi un environnement existe du tout, alors que l’action pourrait simplement s’exécuter à l’intérieur de sa propre classe.
class Action:
...
def execute_action(self, **args):
return function(**args)
Cela convient bien pour des exemples pédagogiques. Cependant, cela devient limitant lorsque l’on a besoin d’injection de dépendances, de contextes cachés, de connexions à une base de données, de sessions utilisateur ou d’autres ressources externes, et les sections suivantes montrent comment cela se produit.
class Environment:
def execute_action(
self,
action: Tool,
args: dict,
) -> Any:
"""Run `action` with the LLM-provided `args`, injecting the hidden context
into whichever param the tool declares for it (by name or subclass type).
Returns the tool's native result, or {"error": ...} if it raised."""
try:
call_args = dict(args)
return action.function(**call_args)
except Exception as e:
return {"error": str(e)}
Construction du cycle de l’agent
Avec les quatre composants du jeu définis, le cycle principal peut être écrit. Il assemble une requête, demande une réponse au LLM, exécute les actions choisies, collecte les observations et répète le processus jusqu’à ce qu’une condition d’arrêt soit remplie.
class Agent:
def __init__(self, goal: Goal, actions: List[Action], environment: Optional[Environment] = None):
self.goal = goal
self.actions = {a.name: a for a in actions}
self.environment = environment or Environment()
self.memory = Memory()
self.memory.add(Message(role=Role.SYSTEM, content=goal.description))
def tools_schema(self) -> List[Dict[str, Any]]:
"""Describe every action in the function-calling format the LLM expects."""
return [
{
"type": "function",
"function": {
"name": a.name,
"description": a.description,
"parameters": a.parameters,
},
}
for a in self.actions.values()
]
def run(self, user_input: str, max_iterations: int = 15) -> Any:
self.memory.add(Message(role=Role.USER, content=user_input))
for _ in range(max_iterations):
# 1. Ask the LLM what to do next, given the conversation + tools.
response = generate_response(self.memory, tools=self.tools_schema())
assistant_msg = response.choices[0].message
# 2. Record the assistant's turn (text and/or tool calls).
self.memory.add(
Message(
role=Role.ASSISTANT,
content=assistant_msg.content,
tool_calls=[tc.model_dump() for tc in assistant_msg.tool_calls]
if assistant_msg.tool_calls
else None,
)
)
# 3. If it asked for tools, run them and feed the results back.
if assistant_msg.tool_calls:
for call in assistant_msg.tool_calls:
action = self.actions[call.function.name]
args = json.loads(call.function.arguments)
result = self.environment.execute_action(action, args)
self.memory.add(
Message(
role=Role.TOOL,
tool_call_id=call.id,
name=action.name,
content=str(result),
)
)
# 4. A terminal action ends the loop and carries the result.
if action.terminal:
return result
return None
if __name__ == "__main__":
number_pair = {
"type": "object",
"properties": {"a": {"type": "number"}, "b": {"type": "number"}},
"required": ["a", "b"],
}
actions = [
Action(
name="add",
description="Adds two numbers.",
parameters=number_pair,
function=add,
),
Action(
name="multiply",
description="Multiplies two numbers.",
parameters=number_pair,
function=multiply,
),
Action(
name="finish",
description="Provide the final answer to the user and end the task.",
parameters={
"type": "object",
"properties": {"answer": {"type": "string"}},
"required": ["answer"],
},
function=finish,
terminal=True, # this one ends the loop
),
]
agent = Agent(
goal=Goal(
name="calculator",
description=(
"Perform arithmetic using the add and multiply tools. "
"When you have the final answer, call finish with it."
),
),
actions=actions,
)
result = agent.run("Add 200 and 10, then multiply the result by 2.")
print("\nresult:", result)
print(json.dumps([m.model_dump(exclude_none=True) for m in agent.memory.get()], indent=2))
## Result
result: 420
[
{
"role": "system",
"content": "Perform arithmetic using the add and multiply tools. When you have the final answer, call finish with it."
},
{
"role": "user",
"content": "Add 200 and 10, then multiply the result by 2."
},
{
"role": "assistant",
"tool_calls": [
{
"function": {
"arguments": "{\"a\":200,\"b\":10}",
"name": "add"
},
"id": "call_wYPzGCwLuc71cK7q2Ok81OzF",
"type": "function"
}
]
},
{
"role": "tool",
"content": "210",
"tool_call_id": "call_wYPzGCwLuc71cK7q2Ok81OzF",
"name": "add"
},
{
"role": "assistant",
"tool_calls": [
{
"function": {
"arguments": "{\"a\":210,\"b\":2}",
"name": "multiply"
},
"id": "call_5JD4McfNm0m68Poedsm8uODO",
"type": "function"
}
]
},
{
"role": "tool",
"content": "420",
"tool_call_id": "call_5JD4McfNm0m68Poedsm8uODO",
"name": "multiply"
},
{
"role": "assistant",
"tool_calls": [
{
"function": {
"arguments": "{\"answer\":\"420\"}",
"name": "finish"
},
"id": "call_nbIGeKvgWtXPh4YlioZxfiyj",
"type": "function"
}
]
},
{
"role": "tool",
"content": "420",
"tool_call_id": "call_nbIGeKvgWtXPh4YlioZxfiyj",
"name": "finish"
}
]
Quelques remarques sur les détails. Le format du message pourrait être organisé différemment par rapport au rôle "tool" utilisé ici, mais la structure nécessiterait de légères modifications. La plupart des API de LLM suivent une convention selon laquelle le résultat d’une outil doit être associé à l’appel d’outil qui le précède, généralement via un tool_id ; ainsi, le même identifiant apparaît deux fois dans l’historique : une fois dans la demande et une fois dans le résultat correspondant.
Les modèles modernes sont également fortement entraînés sur des flux d’appels d’outils. Réutiliser les rôles et les schémas de messages qu’ils ont vus pendant l’entraînement correspond à leurs attentes et permet généralement des appels d’outils plus stables ainsi qu’une meilleure performance globale.
Le outil spécial finish mérite également d’être mentionné. Il permet au modèle de signaler explicitement que la tâche est terminée, afin qu’il ne reste pas bloqué à appeler des outils indéfiniment. Il est enregistré comme n’importe quel autre outil, sur chaque agent, ce qui constitue un aspect du design qui sera amélioré prochainement.
Les limites de la première version
Cet agent fonctionne et expose tout le cycle sans rien cacher derrière un framework, ce qui est idéal pour apprendre. Au-delà des exemples simplifiés, trois limites apparaissent rapidement.
- Les définitions d’outils ne sont pas évolutives. Chaque action nécessite un JSON Schema rédigé manuellement, à synchroniser avec sa fonction Python. À mesure que le nombre d’outils augmente, cela devient répétitif et sujet aux erreurs ; de plus, un petit changement dans la signature peut rendre le schéma obsolète, ce qui amène le modèle à effectuer des appels invalides.
finish arrête la boucle et renvoie un résultat, mais chaque agent doit le déclarer, et il ne peut renvoyer qu’une valeur simple. Les agents plus performants souhaiteront des règles de cessation plus sophistiquées ainsi qu’une sortie structurée.Chacun de ces aspects est abordé à tour de rôle, en commençant par la gestion et l’enregistrement des outils.
Problème 1 : un registre d’outils
Commençons par le plus problématique : il faut synchroniser manuellement le schéma et les fonctions. Les API de complétion exigent une liste de schémas JSON décrivant les outils disponibles.
Dans la première version, chaque outil était constitué de deux éléments écrits séparément : une fonction Python qui effectuait le travail et un schéma JSON indiquant au LLM comment l’appeler :
def add(a: float, b: float) -> float:
return a + b
add_tool = {
"type": "function",
"function": {
"name": "add",
"description": "Adds two floats.",
"parameters": {
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"},
},
"required": ["a", "b"],
},
},
}
tools = [add_tool]
...
completion(
model="openai/gpt-4o",
messages=messages,
tools=tools,
)
La solution consiste à générer le schéma à partir de la fonction. La fonction devient la seule source de vérité, et le schéma est dérivé automatiquement lorsque l’outil est enregistré.
JSON_TYPE_MAP = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
}
def get_json_type(python_type: Any) -> str:
origin = get_origin(python_type)
if python_type in (list, List) or origin in (list, List):
return "array"
if python_type in (dict, Dict) or origin in (dict, Dict):
return "object"
if origin in (Union, types.UnionType):
non_none = [a for a in get_args(python_type) if a is not type(None)]
return get_json_type(non_none[0]) if non_none else "string"
return JSON_TYPE_MAP.get(python_type, "string")
def build_tool(func: Callable, ...) -> Tool:
name = name or func.__name__
description = description or (func.__doc__.strip() if func.__doc__ else "")
signature = inspect.signature(func)
type_hints = get_type_hints(func)
args_schema = {"type": "object", "properties": {}, "required": []}
for param_name, param in signature.parameters.items():
param_type = type_hints.get(param_name, str)
args_schema["properties"][param_name] = {"type": get_json_type(param_type)}
if param.default == inspect.Parameter.empty: # no default ⇒ required
args_schema["required"].append(param_name)
return Tool(name=name, description=description,
parameters=args_schema, function=func)
Ensuite, ce constructeur devrait s’exécuter automatiquement chaque fois qu’une fonction d’outil est définie, c’est pourquoi un décorateur s’en charge :
@tool()
def add(a: float, b: float) -> float:
"""Adds two floats."""
return a + b
Le décorateur est délibérément sobre. Il crée l’objet Tool, le lie à l’objet fonction et renvoie la fonction telle quelle :
def tool(name=None, description=None, parameters_override=None, tags=None):
def decorator(func):
func._tool_metadata = build_tool(func, name, description,
parameters_override, tags)
return func
return decorator
# The decorator automatically add a new attribute _tool_metadata contains the
# information of the tool (Tool)
Essai :
serializable@tool()
def add(a: float, b: float) -> float:
"""Adds two floats."""
return a + b
if __name__ == "__main__":
# _tool_metadata (Tool) will be added as an attribute of the function after the decorator call
tool_metadata = add._tool_metadata
# exclude function as it can not be serializable
print(json.dumps(tool_metadata.model_dump(exclude={"function"}), indent=2)) #
# Result
{
"name": "add",
"description": "Adds two floats.",
"parameters": {
"type": "object",
"properties": {
"a": {
"type": "number"
},
"b": {
"type": "number"
}
},
"required": [
"a",
"b"
]
},
"tags": null,
"terminal": false
}
Lorsqu’il y a de nombreux outils et plusieurs agents, et que ces derniers partagent des outils, un endroit central pour les gérer est utile. Un registre d’actions remplit cette fonction, en suivant quels outils un agent donné peut utiliser.
class ActionRegistry:
def __init__(self):
self.tools: Dict[str, Tool] = {}
def register(self, action: Callable | Tool, terminal: bool = False):
# accept a raw @tool-decorated function or a pre-built Tool
if isinstance(action, Tool):
tool = action
else:
tool = getattr(action, "_tool_metadata", None) or build_tool(action)
if terminal != tool.terminal:
tool = tool.model_copy(update={"terminal": terminal})
self.tools[tool.name] = tool
def get_tool(self, name: str) -> Optional[Tool]:
return self.tools.get(name)
def get_tools_schema(self) -> List[Dict[str, Any]]:
"""Tool schemas in the OpenAI / litellm function-calling format."""
return [
{"type": "function",
"function": {"name": t.name,
"description": t.description,
"parameters": t.parameters}}
for t in self.tools.values()
]
Enregistrer une fonction se fait alors par une seule appel :
registry.register(add)
Le registre lit les métadonnées attachées par @tool et les transforme en une instance Tool. Pour une fonction qui n’a pas été décorée, il recourt automatiquement à l’appel de build_tool(). Il s’agit d’une valeur par défaut pratique, et ceux qui souhaitent un contrôle total peuvent toujours créer des objets Tool manuellement.
Remarquez également que terminal est défini lors de l’enregistrement et non sur l’outil lui-même :
registry.register(tool, terminal=True)
Le fait qu’un outil mette fin à la boucle est une propriété de l’agent, et non de la fonction. Le même outil peut être terminal pour un agent et ordinaire pour un autre. Lorsqu’une valeur terminal différente est demandée, le registre copie l’objet Tool au lieu de modifier l’original, de sorte que la configuration ne s’échappe pas entre les registres et qu’une définition peut être réutilisée en toute sécurité entre différents agents.
Problème 2 : injection de dépendances
Les outils fonctionnent actuellement, mais avec une restriction importante : ils ne voient que les arguments générés par le LLM.
Les outils pratiques ont souvent besoin de ressources qui ne devraient jamais être montrées au modèle, notamment des connexions de base de données, des clients API, des objets de configuration, des sessions utilisateur et l’état de l’application. Certains doivent également accéder à la mémoire de l’agent. Imaginez un outil qui enregistre des articles de recherche dans une base de données :
@tool()
def save_paper(
title: str,
content: str,
database_url: str,
api_key: str,
):
...
Pour le formuler plus clairement, une classe contient le contexte de l’action :
class ActionContext(BaseModel):
database_url: str,
api_key: str
# Then the tool becomes:
@tool()
def save_paper(
title: str,
content: str,
action_context: ActionContext
):
...
Si le contexte n’était qu’un autre argument d’outil, le LLM devrait inventer ces valeurs, ce qui n’a aucun sens, car elles doivent être injectées. C’est là que Environment vient en aide : les paramètres de contexte restent cachés du modèle et sont remplis manuellement au moment de l’exécution.
Puisque certains outils ont besoin de contexte et d’autres non, la première étape consiste en une fonction qui vérifie, à partir des annotations de type, si une fonction prend un ActionContext ou une sous-classe de celui-ci :
def context_param(func: Callable) -> Optional[tuple[str, type]]:
try:
hints = get_type_hints(func)
except Exception:
hints = {}
for name in inspect.signature(func).parameters:
ann = hints.get(name)
if isinstance(ann, type) and issubclass(ann, ActionContext):
return name, ann
return None
# Now we modify the `build_tool` function
def build_tool(
func: Callable,
name: Optional[str] = None,
description: Optional[str] = None,
parameters_override: Optional[Dict[str, Any]] = None,
tags: Optional[List[str]] = None,
) -> Tool:
name = name or func.__name__
description = description or (func.__doc__.strip() if func.__doc__ else "")
if parameters_override is None:
signature = inspect.signature(func)
type_hints = get_type_hints(func)
args_schema = {
"type": "object",
"properties": {},
"required": [],
}
# The injected-context param (by name OR ActionContext-subclass annotation)
# is filled by the Environment from the hidden ActionContext. Never expose it
# to the LLM, or the LLM would try to generate the secrets that live inside it.
ctx = context_param(func)
ctx_param_name = ctx[0] if ctx else None
for param_name, param in signature.parameters.items():
if param_name == ctx_param_name:
continue
param_type = type_hints.get(param_name, str)
param_schema = {
"type": get_json_type(param_type)
}
args_schema["properties"][param_name] = param_schema
if param.default == inspect.Parameter.empty:
args_schema["required"].append(param_name)
else:
args_schema = parameters_override
return Tool(
name=name,
description=description,
parameters=args_schema,
function=func,
tags=tags or []
)
Un outil qui utilise un paramètre ActionContext omet ce paramètre dans son schéma JSON, de sorte que le LLM ne sait pas qu’il existe et ne peut pas essayer de le générer. Le Environment restitue ensuite ce contexte au moment de l’exécution :
class Environment:
def execute_action(
self,
action: Tool,
args: dict,
action_context: Optional[ActionContext] = None,
) -> Any:
try:
call_args = dict(args)
ctx = context_param(action.function)
if ctx:
ctx_param_name, _ = ctx
call_args[ctx_param_name] = action_context or ActionContext()
return action.function(**call_args)
except Exception as e:
return {"error": str(e)}
C’est là le truc. L’Environnement agit comme la frontière entre le modèle et le monde extérieur. Le modèle fournit les arguments dont il a connaissance, tandis que l’Environnement ajoute les ressources qu’il ne devrait pas connaître. C’est une idée simple qui rend les agents bien plus pratiques.
Problème 3 : sortie structurée et outil de fin intégré
Deux problèmes restent à résoudre. Premièrement, l’agent ne peut actuellement renvoyer que des chaînes de caractères, alors que les workflows réels ont souvent besoin de données structurées. Deuxièmement, l’outil de terminaison devrait être intégré à l’agent, et non quelque chose qu’il faut enregistrer externement à chaque fois.
Tous deux sont résolus ensemble. Pour un format structuré, de nombreuses API de complétion offrent un paramètre response_format qui oblige le LLM à générer du JSON conforme à un schéma. Cela entre en conflit avec une boucle d’appel d’outils. Une fois response_format activé, le modèle écrit généralement sa réponse finale immédiatement au lieu d’appeler des outils, et l’agent s’arrête après la première complétion. Mélanger les deux s’est également révélé gênant et incohérent dans la pratique.
L’alternative consiste à réutiliser les appels d’outils. Le schéma de sortie devient alors les paramètres d’un outil spécial, et une fois le travail terminé, le LLM appelle cet outil avec des arguments correspondant au schéma souhaité. L’agent extrait ces arguments et construit l’objet de sortie final.
Un outil d’aide dans la classe Agent enregistre automatiquement un outil intégré terminate :
OutputSchema = Union[type[str], type[BaseModel]]
class Agent:
...
@classmethod
def _build_terminate_tool(cls, output_schema: OutputSchema) -> Tool:
"""Build the terminate tool that ends the agent loop and carries the final result.
str is normalized to a one-field model {result: str} and unwrapped on return,
so a single code path handles both: the LLM-facing schema always comes from
pydantic's model_json_schema(), and result is the string (for str) or
the schema instance (for a BaseModel).
"""
unwrap_field = None
if output_schema is str:
output_schema = create_model("StrResult", **{cls._TERMINATE_STR_FIELD: (str, ...)})
unwrap_field = cls._TERMINATE_STR_FIELD
def terminate(**kwargs: Any) -> Any:
"""Provide the final result and end the task."""
obj = output_schema(**kwargs)
return getattr(obj, unwrap_field) if unwrap_field else obj
terminate_tool = build_tool(
func=terminate,
name=cls.TERMINATE_TOOL_NAME,
description=f"Provide the final result ({output_schema.__name__}) and end the task.",
parameters_override=output_schema.model_json_schema(),
)
terminate_tool.terminal = True
return terminate_tool
Et voici le cycle de l’agent terminé :
class Agent:
TERMINATE_TOOL_NAME = "terminate"
_TERMINATE_STR_FIELD = "result"
def __init__(
self,
goal: str,
action_registry: ActionRegistry,
output_schema: Optional[OutputSchema] = None,
environment: Optional["Environment"] = None,
action_context: Optional[ActionContext] = None,
):
self.goal = goal
self.action_registry = action_registry
self.output_schema = output_schema
self.result: Any = None
self.environment = environment or Environment()
self.action_context = action_context or ActionContext()
self._validate_context()
# Tools the registry marked terminal (excluding the framework's terminate tool).
terminal_tools = [
t.name for t in action_registry.tools.values()
if t.terminal and t.name != self.TERMINATE_TOOL_NAME
]
# No exit declared at all -> default to a text answer (output_schema=str).
if output_schema is None and not terminal_tools:
output_schema = self.output_schema = str
# Because in my design, Agent can stop the loop after a specific tool call. (to save tokens, LLm do not need to generate terminal tool with the same parameters again)
# This choice make it a little bit complicated when output schema defined in Agent different from the tool's output schema.
if output_schema is not None:
if terminal_tools:
raise ValueError(
f"output_schema={output_schema.__name__} conflicts with terminal "
f"tool(s) {terminal_tools}: their output does not match the schema. "
f"Drop output_schema or unmark those tools as terminal."
)
action_registry.register_tool(self._build_terminate_tool(output_schema))
elif len(terminal_tools) > 1:
raise ValueError(
f"Multiple terminal tools {terminal_tools}; an agent must have exactly "
f"one exit so result has a well-defined type."
)
self.memory = Memory()
self.memory.add(Message(role=Role.SYSTEM, content=self.goal))
def _validate_context(self) -> None:
"""Verify that the agent's action_context satisfies every tool's declared"""
for tool in self.action_registry.tools.values():
ctx = context_param(tool.function)
if ctx is None:
continue # tool declares no context — nothing to satisfy
_, required = ctx
if not isinstance(self.action_context, required):
raise TypeError(
f"Tool {tool.name!r} requires action_context: {required.__name__}, "
f"but this agent was given {type(self.action_context).__name__}. "
f"Pass a context that is-a {required.__name__} (combine schemas via "
f"subclassing if multiple tools need different types), or split the "
f"tools across agents."
)
@classmethod
def _build_terminate_tool(cls, output_schema: OutputSchema) -> Tool:
"""Build the terminate tool that ends the agent loop and carries the final result.
str is normalized to a one-field model {result: str} and unwrapped on return,
so a single code path handles both: the LLM-facing schema always comes from
pydantic's model_json_schema(), and result is the string (for str) or
the schema instance (for a BaseModel).
"""
unwrap_field = None
if output_schema is str:
output_schema = create_model("StrResult", **{cls._TERMINATE_STR_FIELD: (str, ...)})
unwrap_field = cls._TERMINATE_STR_FIELD
def terminate(**kwargs: Any) -> Any:
"""Provide the final result and end the task."""
obj = output_schema(**kwargs)
return getattr(obj, unwrap_field) if unwrap_field else obj
terminate_tool = build_tool(
func=terminate,
name=cls.TERMINATE_TOOL_NAME,
description=f"Provide the final result ({output_schema.__name__}) and end the task.",
parameters_override=output_schema.model_json_schema(),
)
terminate_tool.terminal = True
return terminate_tool
def add_memory(self, message: Message):
self.memory.add(message)
@staticmethod
def _to_content(result: Any) -> str:
"""Render a tool's return value as text for the LLM. """
if isinstance(result, BaseModel):
return result.model_dump_json()
if isinstance(result, (dict, list)):
return json.dumps(result, ensure_ascii=False)
return str(result)
def run(self, user_input: str, max_iterations: int = 15):
self.add_memory(Message(role=Role.USER, content=user_input))
for i in range(max_iterations):
response = generate_response(self.memory, tools=self.action_registry.get_tools_schema())
assistant_msg = response.choices[0].message
# Record the assistant turn (may carry content and/or tool_calls).
self.add_memory(
Message(
role=Role.ASSISTANT,
content=assistant_msg.content,
tool_calls=[tc.model_dump() for tc in assistant_msg.tool_calls]
if assistant_msg.tool_calls else None,
)
)
if assistant_msg.tool_calls:
for tool_call in assistant_msg.tool_calls:
tool_name = tool_call.function.name
tool_args = tool_call.function.arguments
tool = self.action_registry.get_tool(tool_name)
if not tool:
raise ValueError(f"Tool '{tool_name}' not found.")
# Execute the tool in the Environment
result = self.environment.execute_action(
tool, json.loads(tool_args), self.action_context
)
self.add_memory(
Message(
role=Role.TOOL,
tool_call_id=tool_call.id,
name=tool_name,
content=self._to_content(result)
)
)
if tool.terminal:
self.result = result
return self.result
return self.result
Ce que montre l’exercice
Fondamentalement, un agent est un modèle de langage qui s’exécute à l’intérieur d’une boucle. Il lit la situation, choisit une action, obtient le résultat et répète le processus jusqu’à ce que l’objectif soit atteint. Pydantic AI, AutoGen, Microsoft Agent Framework et la plupart des autres solutions sont basés sur ce même schéma.
Étape par étape, une simple boucle d’appel d’outils s’est transformée en quelque chose de plus performant. Un registre d’outils a permis de rendre les actions réutilisables, une couche d’environnement a injecté en toute sécurité des dépendances cachées, et un mécanisme de terminaison structuré a permis aux agents de retourner des sorties typées au lieu de texte brut. Aucune de ces idées n’est difficile en soi, mais ensemble elles constituent la base de la plupart des conceptions d’agents modernes.
Connaître ces éléments de base permet d’évaluer plus facilement les frameworks, de déboguer les comportements anormaux plus aisément et d’adapter ses propres conceptions à ses besoins. Les frameworks accélèrent le développement, mais ce ne sont que des abstractions de concepts qui se révèlent assez simples une fois exposés clairement.