Construyendo un agente de IA desde cero en Python: El bucle del juego paso a paso
Crea un agente para llamar a herramientas que funcione con Python puro, utilizando Objetivos, Acciones, Memoria y Entorno, y luego agrega un registro de herramientas, inyección de dependencias oculta y salida tipada.
Los frameworks para agentes como Pydantic AI, AutoGen y Microsoft Agent Framework son útiles para poner en marcha un prototipo rápidamente. Sin embargo, después de usar varios de ellos, suelen surgir algunos problemas.
- El ecosistema sigue en desarrollo. Cada framework cubre algunas de las funcionalidades que se necesitan y omite otras, mientras que otro framework cuenta con aquellas que falta en el primero. Eso es normal en un campo aún joven, y con el tiempo la mayoría podrán manejar los casos de uso más comunes.
- La abstracción se acumula. Los frameworks buscan servir a una amplia gama de diseños de agentes, por lo que añaden capas que son difíciles de seguir y aún más complicadas de depurar. LangChain es el ejemplo típico: cuando el sistema funciona mal, estas capas dificultan la identificación de la causa.
Lo alentador es que un agente no tiene por qué ser complicado. Para muchas tareas basta con una implementación sencilla. Crear uno desde cero te da control total sobre lo que ingresa al modelo, cómo se toman las decisiones y cómo se gestionan los costos; además, conocer los principios hace que cualquier framework posterior sea más fácil de confiar.
La idea central: el bucle del agente
Comienza por simplificar la idea. Un agente de IA es un modelo de lenguaje grande que puede tomar acciones y utilizar sus resultados para decidir qué hacer a continuación. En un intercambio tradicional con un LLM, envías una solicitud y recibes una respuesta. En cambio, un agente funciona en bucle: examina el estado actual, elige una acción, la lleva a cabo, recibe retroalimentación y vuelve a repetir el proceso hasta que se completa la tarea.
GAME: un diseño de cuatro partes
Para construir ese bucle utilizaremos la estructura GAME, que divide al agente en Objetivos, Acciones, Memoria y Entorno. Esto proporciona un modelo mental compacto y mantiene el código modular y fácil de extender.
Objetivo
El objetivo indica qué está intentando lograr el agente. Contiene la descripción de la tarea y cualquier instrucción que guíe su comportamiento, y se convierte en el prompt del sistema.
class Goal(BaseModel):
name: str
description: str
Acción (herramienta)
La base de código aquí la denomina Herramienta, pero ambos términos significan lo mismo. Una acción es algo que el agente puede realizar: llamar a una API, buscar documentos, consultar una base de datos o hacer un cálculo.
class Tool(BaseModel):
name: str
description: str
parameters: Dict[str, Any]
function: Callable
tags: Optional[List[str]] = None
terminal: bool = False
Las API actuales de LLM, incluidas las de OpenAI, aceptan definiciones de herramientas junto con el prompt. Dichas definiciones deben redactarse en formato JSON Schema, indicando el nombre de cada herramienta, sus parámetros y la entrada esperada.
Como primera herramienta, tomemos una función que suma dos números:
def add(a: float, b: float) -> float:
return a + b
Luego descríbalo en un formato que el modelo pueda leer:
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,
)
Cuando se construye la solicitud, estos objetos de acción se convierten en definiciones de herramientas y se envían al modelo. El modelo nunca ejecuta la función en sí; solo decide qué herramienta llamar y genera los argumentos necesarios para ella.
Supongamos que el usuario pregunta: "¿Cuánto es 10 + 20?" El modelo podría responder algo como esto:
{
"name": "add",
"arguments": {
"a": 10,
"b": 20
}
}
Elegir la acción y proporcionar sus argumentos es toda la tarea del modelo.
Envío de acciones al modelo
Este ayudante pasa el historial de la conversación y las acciones disponibles al 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": []
}
Luego, el agente ejecuta la función de coincidencia en Python y pasa el resultado al siguiente paso del bucle.
Memoria
La memoria almacena lo que se recopiló en iteraciones anteriores, de modo que el agente puede tener en cuenta las observaciones pasadas al elegir su próximo paso.
Por ahora, su tamaño es mínimo: los nuevos mensajes se añaden simplemente a una lista. Las ventanas de contexto de los LLM son finitas, por lo que un sistema real necesita algo más inteligente para recortar o resumir la información, algo que este ejemplo omite.
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 modelo Message representa cada interacción en el bucle. Las observaciones, los resultados de las herramientas, la entrada del usuario y las respuestas del asistente se convierten todos en mensajes. Su estructura sigue de cerca el formato de mensaje de las API actuales de LLM, por lo que crear prompts a partir del historial es sencillo y la memoria resulta fácil de ampliar.
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
Entorno
Puede preguntarse por qué existe un entorno, si la acción podría ejecutarse directamente dentro de su propia clase.
class Action:
...
def execute_action(self, **args):
return function(**args)
Eso está bien para ejemplos sencillos. Se vuelve limitante cuando se necesita inyección de dependencias, contexto oculto, conexiones a bases de datos, sesiones de usuario u otros recursos externos, y las secciones posteriores muestran cómo sucede eso.
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)}
Construyendo el bucle del agente
Una vez definidas las cuatro partes de GAME, se puede escribir el bucle principal. Este reúne una instrucción, solicita una respuesta al LLM, ejecuta las acciones seleccionadas, recopila las observaciones y repite el proceso hasta que se cumpla una condición de parada.
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"
}
]
Algunas notas sobre los detalles. El formato del mensaje podría organizarse de manera diferente al rol "tool" utilizado aquí, pero la estructura requeriría pequeños ajustes. La mayoría de las APIs de LLM siguen una convención según la cual el resultado de una herramienta debe estar vinculado a la llamada a dicha herramienta que la precedió, normalmente a través de un tool_id; por lo tanto, el mismo identificador aparece dos veces en el historial: una vez en la solicitud y otra vez en el resultado correspondiente.
Los modelos modernos también se entrenan intensamente con flujos de llamadas a herramientas. Reutilizar los roles y patrones de mensajes que vieron durante el entrenamiento se alinea con sus expectativas y, por lo general, permite llamadas a herramientas más estables y un mejor rendimiento general.
La herramienta especial finish también merece ser mencionada. Le brinda al modelo una forma explícita de indicar que la tarea está completada, para que no se quede atascado llamando herramientas indefinidamente. Se registra como cualquier otra herramienta, en cada agente, lo cual es un aspecto del diseño que se mejorará pronto.
Dónde falla la primera versión
Este agente funciona y muestra todo el ciclo sin nada oculto detrás de un framework, lo cual es ideal para aprender. Más allá de los ejemplos sencillos iniciales, rápidamente surgen tres limitaciones.
- Las definiciones de herramientas no son escalables. Cada acción requiere un JSON Schema escrito a mano que debe mantenerse sincronizado con su función en Python. A medida que aumentan las herramientas, esto se vuelve repetitivo y propenso a errores; además, un pequeño cambio en la firma puede hacer que el esquema quede obsoleto, lo que hace que el modelo genere llamadas inválidas.
finish detiene el bucle y devuelve un resultado, pero cada agente debe registrarla, y solo puede devolver un valor simple. Los agentes más capaces buscarán reglas de detención más complejas y una salida estructurada.Cada uno de estos aspectos se aborda por turnos, comenzando con la forma en que se gestionan y registran las herramientas.
Problema 1: un registro de herramientas
Comencemos con el más problemático: mantener alineados manualmente el esquema y la función. Las APIs de completado requieren una lista de esquemas JSON que describan las herramientas disponibles.
En la primera versión, cada herramienta venía como dos elementos escritos por separado: una función de Python que realizaba el trabajo y un JSON Schema que indicaba al LLM cómo llamarla:
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 solución es generar el esquema a partir de la función. La función se convierte en la única fuente de verdad, y el esquema se deriva automáticamente cuando se registra la herramienta.
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)
A continuación, este generador debería ejecutarse automáticamente cada vez que se defina una función de herramienta, por lo que un decorador se encarga de esa tarea:
@tool()
def add(a: float, b: float) -> float:
"""Adds two floats."""
return a + b
El decorador es deliberadamente sencillo. Construye la Tool, la adjunta al objeto de función y devuelve la función sin modificaciones:
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)
Poniéndolo a prueba:
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
}
Cuando hay muchas herramientas y varios agentes, y estos últimos comparten herramientas, es útil contar con un lugar central para gestionarlas. Un registro de acciones desempeña ese papel, ya que registra qué herramientas puede utilizar un agente determinado.
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()
]
El registro de una función consiste entonces en una sola llamada:
registry.register(add)
El registro lee los metadatos que @tool adjuntó y los convierte en una instancia de Tool. Para una función que no estuvo decorada, recurre automáticamente a la llamada a build_tool(). Ese es un valor por defecto conveniente, y quienes desean un control total aún pueden crear objetos Tool a mano.
También note que terminal se establece durante el registro y no en la herramienta:
registry.register(tool, terminal=True)
El hecho de que una herramienta termine el bucle es una propiedad del agente, no de la función. La misma herramienta puede ser terminal para un agente y ordinaria para otro. Cuando se solicita un valor terminal diferente, el registro copia la Tool en lugar de modificar la original, de modo que la configuración no se filtra entre registros y una definición puede reutilizarse con seguridad entre agentes.
Problema 2: inyección de dependencias
Las herramientas funcionan actualmente, con una restricción importante: solo pueden acceder a los argumentos generados por el LLM.
Las herramientas prácticas suelen necesitar recursos que nunca deberían mostrarse al modelo, como conexiones a bases de datos, clientes de API, objetos de configuración, sesiones de usuario y estado de la aplicación. Algunas también necesitan acceder a la memoria del agente. Imagine una herramienta que guarda artículos de investigación en una base de datos:
@tool()
def save_paper(
title: str,
content: str,
database_url: str,
api_key: str,
):
...
Para resumirlo mejor, una clase contiene el contexto de la acción:
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 el contexto fuera simplemente otro argumento de la herramienta, el LLM tendría que inventar esos valores, lo cual no tiene sentido, ya que se supone que deben ser inyectados. Aquí es donde Environment ayuda. Los parámetros de contexto están ocultos al modelo y se completan manualmente en el momento de la ejecución.
Dado que algunas herramientas necesitan contexto y otras no, el primer paso es una función que verifica, a partir de las anotaciones de tipo, si una función acepta un ActionContext o una subclase de él:
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 []
)
Una herramienta que utiliza un parámetro ActionContext omite dicho parámetro en su esquema JSON, por lo que el LLM no sabe que existe y no puede intentar generarlo. Entonces, Environment vuelve a incluir el contexto en el momento de la ejecución:
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)}
Ese es el truco. El Entorno actúa como la frontera entre el modelo y el mundo exterior. El modelo proporciona los argumentos de los que tiene conocimiento, mientras que el Entorno agrega los recursos sobre los cuales no debería tener información. Es una idea sencilla que hace que los agentes sean mucho más prácticos.
Problema 3: salida estructurada y una herramienta de finalización integrada
Quedan dos cuestiones por resolver. Primero, el agente actualmente solo puede devolver cadenas de texto, aunque los flujos de trabajo reales suelen necesitar datos estructurados. Segundo, la herramienta de finalización debería estar integrada en el agente, y no ser algo que se registre externamente cada vez.
Se resuelven ambos juntos. Para un formato estructurado, muchas API de completamiento ofrecen un parámetro response_format que obliga al LLM a emitir JSON que se ajuste a un esquema. Esto entra en conflicto con un bucle de llamadas a herramientas. Una vez que se activa response_format, el modelo suele escribir su respuesta final de inmediato en lugar de llamar a herramientas, y el agente se detiene después de la primera completación. Mezclar los dos métodos también resulta incómodo e inconsistente en la práctica.
La alternativa es reutilizar las llamadas a herramientas. El esquema de salida se convierte en los parámetros de una herramienta especial, y cuando el trabajo está terminado, el LLM llama a esa herramienta con argumentos que coinciden con el esquema deseado. El agente extrae esos argumentos y construye el objeto de salida final.
Un ayudante en la clase Agent registra automáticamente una herramienta terminate incorporada:
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
Y aquí está el bucle del agente terminado:
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
Lo que muestra el ejercicio
En esencia, un agente es un modelo de lenguaje que funciona dentro de un bucle. Lee la situación, elige una acción, obtiene el resultado y repite el proceso hasta alcanzar el objetivo. Pydantic AI, AutoGen, Microsoft Agent Framework y la mayoría de las demás herramientas se basan en este mismo patrón.
Paso a paso, un simple bucle de llamadas a herramientas se transformó en algo más capaz. Un registro de herramientas permitió que las acciones fueran reutilizables, una capa de entorno inyectó dependencias ocultas de forma segura, y un mecanismo estructurado de terminación permitió a los agentes devolver resultados con formato definido en lugar de texto simple. Ninguna de estas ideas es difícil por sí sola, pero juntas constituyen la base del diseño moderno de agentes.
Conocer estos conceptos básicos facilita evaluar los frameworks, depurar comportamientos inusuales y adaptar los propios diseños a las necesidades. Los frameworks aceleran el desarrollo, pero son abstracciones de conceptos que resultan ser bastante simples una vez expuestos claramente.