Módulo 3: Agents con create_agent
Estado y Memoria del Agente
Descripción de la cápsula
Los agentes creados con create_agent mantienen estado — un objeto que acumula la conversación, resultados de herramientas, y cualquier dato custom que definas. Entender cómo funciona el estado es clave para construir agentes que recuerden contexto, acumulen información, y tomen decisiones basadas en todo lo que ha ocurrido durante la sesión.
En esta cápsula aprenderás cómo funciona el estado por defecto (la lista de messages que se acumula automáticamente), cómo extenderlo con campos custom usando TypedDict, cómo acceder al estado desde prompts dinámicos y tools, y la diferencia entre state (mutable por ejecución) y config (inmutable). Al terminar, sabrás diseñar el estado de cualquier agente según las necesidades de tu aplicación.
Todo lo que aprendes aquí complementa la cápsula anterior — allí viste cómo los prompts dinámicos leen el estado. Aquí aprenderás a diseñar qué contiene ese estado.
Estado por defecto: messages
Cuando creas un agente con create_agent, el estado por defecto incluye una lista de messages que se acumula automáticamente con cada interacción del loop ReAct:
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
@tool
def get_price(product: str) -> str:
"""Obtiene el precio de un producto."""
prices = {"laptop": "$999", "mouse": "$29", "teclado": "$79"}
return prices.get(product.lower(), f"Producto '{product}' no encontrado")
model = ChatOpenAI(model="gpt-4.1-mini")
agent = create_agent(model, [get_price])
result = agent.invoke({
"messages": [("user", "¿Cuánto cuesta un mouse?")]
})
for msg in result["messages"]:
print(f"{msg.type}: {msg.content[:100] if msg.content else '[tool_call]'}")
# Output esperado:
# human: ¿Cuánto cuesta un mouse?
# ai: [tool_call para get_price]
# tool: $29
# ai: El mouse cuesta $29.
La lista messages contiene todo el historial: mensajes del usuario (HumanMessage), respuestas del modelo (AIMessage con tool calls o texto), y resultados de tools (ToolMessage). Cada ciclo del loop ReAct agrega más mensajes hasta que el agente decide responder sin llamar tools.
Estado custom con state_schema
Por qué extender el estado
El estado por defecto solo tiene messages. Pero en aplicaciones reales necesitas trackear más datos:
- ✅ Fuentes encontradas durante una investigación
- ✅ Contadores (intentos, búsquedas realizadas)
- ✅ Flags de control (investigación completa, usuario verificado)
- ✅ Datos acumulados (URLs visitadas, productos comparados)
Para agregar campos custom, usas state_schema con TypedDict:
from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator
class ResearchState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
sources_found: list[str]
research_complete: bool
@tool
def search_web(query: str) -> str:
"""Busca información en internet."""
return f"Resultado para '{query}': [información encontrada]"
model = ChatOpenAI(model="gpt-4.1-mini")
agent = create_agent(model, [search_web], state_schema=ResearchState)
result = agent.invoke({
"messages": [("user", "Investiga sobre LangChain")],
"sources_found": [],
"research_complete": False,
})
print(result["messages"][-1].content)
# Output esperado: [Respuesta del agente sobre LangChain]
Entendiendo Annotated y operator.add
El campo messages siempre necesita Annotated[list[AnyMessage], operator.add]. Esto le dice a LangGraph cómo combinar los nuevos valores con los existentes:
from typing import TypedDict, Annotated
from langchain_core.messages import AnyMessage
import operator
class MyState(TypedDict):
# operator.add = los nuevos mensajes se AGREGAN a la lista existente
messages: Annotated[list[AnyMessage], operator.add]
# Sin Annotated = el valor se REEMPLAZA completamente
current_topic: str
search_count: int
La regla es simple:
- ✅
Annotated[list, operator.add]→ los valores se acumulan (append) - ✅ Sin
Annotated→ el valor se reemplaza completamente
Para messages, siempre usas operator.add porque quieres que el historial crezca. Para campos como contadores o flags, usas tipos simples sin Annotated porque quieres sobrescribirlos.
Inicializar el estado
Cuando invocas un agente con state_schema custom, pasas los valores iniciales de todos los campos:
from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator
class ShoppingState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
budget: float
items_in_cart: list[str]
@tool
def search_product(name: str) -> str:
"""Busca un producto por nombre."""
catalog = {"laptop": "Laptop Pro — $999.99", "mouse": "Mouse Wireless — $29.99", "monitor": "Monitor 27\" — $349.99"}
return catalog.get(name.lower(), f"Producto '{name}' no encontrado")
model = ChatOpenAI(model="gpt-4.1-mini")
agent = create_agent(
model, [search_product],
state_schema=ShoppingState,
prompt="Eres un asistente de compras. El usuario tiene un presupuesto limitado. Ayúdalo a encontrar productos dentro de su presupuesto. Responde en español.",
)
result = agent.invoke({
"messages": [("user", "Busco un mouse")],
"budget": 100.00,
"items_in_cart": [],
})
print(result["messages"][-1].content)
# Output esperado: [Respuesta con el mouse encontrado y su precio]
Si no pasas un campo que está en state_schema, su valor será None. Es buena práctica siempre pasar todos los campos explícitamente.
Acceder al estado desde prompts dinámicos
En la cápsula anterior viste prompts dinámicos. Ahora que sabes definir estado custom, puedes combinar ambos para crear agentes altamente personalizados:
from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator
class TutorState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
user_name: str
user_level: str
queries_remaining: int
@tool
def search_docs(query: str) -> str:
"""Busca en la documentación técnica."""
return f"Documentación para '{query}': [resultados]"
model = ChatOpenAI(model="gpt-4.1-mini")
def personalized_prompt(state):
"""Genera prompt basado en el estado completo del agente."""
name = state.get("user_name", "usuario")
level = state.get("user_level", "beginner")
remaining = state.get("queries_remaining", 10)
level_instructions = {
"beginner": "Explica conceptos de forma simple con analogías.",
"intermediate": "Da explicaciones técnicas directas con código.",
"advanced": "Sé conciso y técnico. Enfócate en edge cases.",
}
instructions = level_instructions.get(level, level_instructions["beginner"])
prompt = f"Eres un tutor de programación asistiendo a {name}. Nivel: {level}. {instructions} Consultas restantes: {remaining}."
if remaining <= 2:
prompt += " Avisa al usuario que le quedan pocas consultas."
prompt += " Responde en español."
return prompt
agent = create_agent(model, [search_docs], prompt=personalized_prompt, state_schema=TutorState)
result = agent.invoke({
"messages": [("user", "¿Cómo funciona async/await?")],
"user_name": "Laura",
"user_level": "intermediate",
"queries_remaining": 5,
})
print(result["messages"][-1].content)
# Output esperado: [Explicación técnica de async/await adaptada a nivel intermedio]
Acceder al estado desde tools
Las tools pueden recibir información inmutable via config usando RunnableConfig:
from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
from langchain_core.runnables import RunnableConfig
import operator
class AppState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
user_id: str
@tool
def get_user_data(field: str, config: RunnableConfig) -> str:
"""Obtiene datos del usuario actual. field: 'profile', 'orders', 'settings'."""
user_id = config.get("configurable", {}).get("user_id", "unknown")
user_db = {"USR-001": {"profile": "María García — Premium", "orders": "3 pedidos activos", "settings": "Idioma: ES"}}
return user_db.get(user_id, {}).get(field, f"Campo '{field}' no encontrado para {user_id}")
model = ChatOpenAI(model="gpt-4.1-mini")
agent = create_agent(model, [get_user_data], state_schema=AppState, prompt="Eres un asistente de cuenta. Responde en español.")
result = agent.invoke(
{"messages": [("user", "¿Cuáles son mis pedidos?")], "user_id": "USR-001"},
config={"configurable": {"user_id": "USR-001"}},
)
print(result["messages"][-1].content)
# Output esperado: [Respuesta con los 3 pedidos activos del usuario]
La config se pasa como segundo argumento a invoke() y la tool la recibe via el parámetro config: RunnableConfig.
Historial de conversación: memoria de corto plazo
El estado messages funciona como memoria de corto plazo — el agente recuerda todo lo que ha pasado en la sesión actual. Puedes simular conversaciones multi-turno pasando el historial acumulado:
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
@tool
def calculate(expression: str) -> str:
"""Evalúa una expresión matemática."""
try:
return str(eval(expression))
except Exception as e:
return f"Error: {e}"
model = ChatOpenAI(model="gpt-4.1-mini")
agent = create_agent(model, [calculate], prompt="Eres un tutor de matemáticas. Recuerda el contexto de la conversación. Responde en español.")
conversation = []
questions = [
"¿Cuánto es 15 * 23?",
"Ahora divide ese resultado entre 5",
"¿Y si le sumo 100?"
]
for question in questions:
conversation.append(("user", question))
result = agent.invoke({"messages": conversation.copy()})
response = result["messages"][-1]
conversation = [(msg.type if hasattr(msg, 'type') else msg[0],
msg.content if hasattr(msg, 'content') else msg[1])
for msg in result["messages"]]
print(f"User: {question}")
print(f"Agent: {response.content}\n")
# Output esperado:
# User: ¿Cuánto es 15 * 23?
# Agent: 15 × 23 = 345.
#
# User: Ahora divide ese resultado entre 5
# Agent: 345 ÷ 5 = 69.
#
# User: ¿Y si le sumo 100?
# Agent: 69 + 100 = 169.
Cada turno el agente recibe todo el historial — por eso resuelve "ese resultado" sin ambigüedad. Para persistencia entre sesiones, necesitas checkpointing (Módulo 8).
Estado vs Config: qué va en cada uno
| Criterio | State | Config |
|---|---|---|
| Mutabilidad | Cambia durante la ejecución | Fijo durante toda la ejecución |
| Propósito | Datos que evolucionan | Settings que no cambian |
| Ejemplos | messages, contadores, flags | user_id, api_keys, thread_id |
| Quién lo modifica | El agente y las tools | Solo se define al inicio |
| Acceso | state["field"] | config["configurable"]["field"] |
result = agent.invoke(
{
# State: datos que cambian durante la ejecución
"messages": [("user", "Analiza las ventas")],
"queries_executed": [],
"analysis_complete": False,
},
# Config: datos inmutables
config={"configurable": {"database": "production", "user_id": "analyst-42"}},
)
Regla práctica:
- ✅ Si el dato cambia durante la ejecución → state
- ✅ Si el dato es constante durante toda la ejecución → config
Diseñar el estado: ejemplo completo
Cuando diseñes el estado, combina los diferentes tipos de campos que tu agente necesita:
from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator
class ResearchAgentState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
researcher_name: str
research_topic: str
sources: list[str]
key_findings: list[str]
searches_performed: int
max_searches: int
research_complete: bool
@tool
def web_search(query: str) -> str:
"""Busca información en internet sobre un tema."""
return f"Resultados para '{query}': [3 artículos relevantes encontrados]"
@tool
def summarize_findings(findings: str) -> str:
"""Genera un resumen de los hallazgos de investigación."""
return f"Resumen: {findings[:200]}..."
model = ChatOpenAI(model="gpt-4.1")
def research_prompt(state):
name = state.get("researcher_name", "investigador")
topic = state.get("research_topic", "tema no definido")
searches = state.get("searches_performed", 0)
max_s = state.get("max_searches", 5)
if state.get("research_complete", False):
return f"La investigación sobre '{topic}' está completa. Resume los hallazgos. Responde en español."
remaining = max_s - searches
return (
f"Eres un asistente de investigación ayudando a {name}. "
f"Tema: {topic}. Búsquedas: {searches}/{max_s} (quedan {remaining}). "
f"Busca información relevante. Si ya tienes suficiente o no quedan búsquedas, genera el resumen final. Responde en español."
)
agent = create_agent(model, [web_search, summarize_findings], prompt=research_prompt, state_schema=ResearchAgentState)
result = agent.invoke({
"messages": [("user", "Investiga las tendencias actuales en IA generativa")],
"researcher_name": "Carlos",
"research_topic": "IA generativa 2026",
"sources": [],
"key_findings": [],
"searches_performed": 0,
"max_searches": 3,
"research_complete": False,
})
print(result["messages"][-1].content)
# Output esperado: [Respuesta del agente con resultados de investigación]
Checklist: (1) ¿Qué recordar entre turnos? → messages. (2) ¿Acumula info? → list[str]. (3) ¿Contadores? → int. (4) ¿Flags? → bool. (5) ¿Contexto para prompt? → str.
Conexión con el proyecto
En el Agente de Investigación con Tools (proyecto de este módulo):
- Definirás un
ResearchStatecon campos para fuentes, hallazgos y flags de control - El prompt dinámico leerá el estado para adaptar el comportamiento según el progreso
- El agente acumulará fuentes y hallazgos a lo largo de múltiples tool calls
- En la Cápsula 05 (Streaming), verás cómo observar la evolución del estado en tiempo real
Todo lo que aprendes aquí se aplica directamente en la Cápsula 08.
Troubleshooting
Problema 1: "KeyError" al acceder a campos del estado
Causa: El campo no se pasó al llamar invoke(), o no está definido en state_schema.
Solución: Siempre pasa todos los campos del state_schema en invoke(), y usa state.get("field", default) en prompts dinámicos para manejar valores faltantes.
Problema 2: Los mensajes se pierden entre invocaciones
Causa: Cada invoke() es una ejecución independiente. El estado no persiste entre llamadas.
Solución: Acumula los mensajes manualmente entre llamadas:
history = []
# Turno 1
history.append(("user", "¿Qué es Python?"))
result = agent.invoke({"messages": history.copy()})
history = [(m.type, m.content) for m in result["messages"]]
# Turno 2 — incluye historial del turno 1
history.append(("user", "¿Y para qué se usa?"))
result = agent.invoke({"messages": history.copy()})
Para persistencia automática entre sesiones, necesitas checkpointing (Módulo 8).
Problema 3: El state_schema no acepta los campos que paso
Causa: Los tipos en invoke() no coinciden con los definidos en TypedDict.
Solución: Verifica que los tipos coincidan: si defines count: int, pasa count=5 (no count="5").
Problema 4: La lista de messages crece sin límite
Causa: El historial acumula todos los mensajes (incluyendo tool calls intermedios) y puede exceder el context window. Solución:
def trim_messages(messages, max_messages=20):
"""Mantiene solo los últimos N mensajes."""
if len(messages) <= max_messages:
return messages
return messages[-max_messages:]
trimmed = trim_messages(history, max_messages=20)
result = agent.invoke({"messages": trimmed})
Para estrategias avanzadas (summarization, sliding window), consulta el Módulo 8.
Ejercicios
Ejercicio 1: Estado con contador de búsquedas (Fácil)
Define un state_schema con search_count y max_searches. Crea un prompt dinámico que avise al usuario cuando queden menos de 2 búsquedas.
Ver solución
from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator
class SearchState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
search_count: int
max_searches: int
@tool
def search(query: str) -> str:
"""Busca información en internet."""
return f"Resultados para '{query}': [datos encontrados]"
model = ChatOpenAI(model="gpt-4.1-mini")
def search_aware_prompt(state):
count = state.get("search_count", 0)
max_s = state.get("max_searches", 5)
remaining = max_s - count
prompt = f"Eres un asistente de investigación. Búsquedas: {count}/{max_s}. "
if remaining <= 2:
prompt += f"⚠️ Solo quedan {remaining} búsquedas. Prioriza preguntas directas. Avisa al usuario. "
else:
prompt += "Busca libremente para encontrar la mejor información. "
prompt += "Responde en español."
return prompt
agent = create_agent(model, [search], prompt=search_aware_prompt, state_schema=SearchState)
result = agent.invoke({
"messages": [("user", "Busca información sobre LangChain")],
"search_count": 3,
"max_searches": 5,
})
print(result["messages"][-1].content)
# Output esperado: [Respuesta con aviso de que quedan pocas búsquedas]
Explicación: El prompt dinámico lee search_count y max_searches del estado y cambia su comportamiento cuando quedan <=2.
Ejercicio 2: Estado con lista de fuentes acumuladas (Fácil)
Define un estado con sources: list[str]. El prompt dinámico debe listar las fuentes ya consultadas para que el agente no repita búsquedas.
Ver solución
from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator
class SourceState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
sources: list[str]
@tool
def search_article(topic: str) -> str:
"""Busca artículos sobre un tema específico."""
articles = {
"langchain": "Artículo: 'LangChain v1.2 Release Notes' (langchain.com)",
"agents": "Artículo: 'Building AI Agents in 2026' (arxiv.org)",
}
for key, value in articles.items():
if key in topic.lower():
return value
return f"No se encontraron artículos sobre '{topic}'"
model = ChatOpenAI(model="gpt-4.1-mini")
def source_aware_prompt(state):
sources = state.get("sources", [])
prompt = "Eres un investigador que recopila artículos técnicos. "
if sources:
prompt += f"Ya consultaste: [{', '.join(sources)}]. No repitas búsquedas sobre los mismos temas. "
else:
prompt += "Aún no has consultado ninguna fuente. Comienza la investigación. "
prompt += "Responde en español."
return prompt
agent = create_agent(model, [search_article], prompt=source_aware_prompt, state_schema=SourceState)
result = agent.invoke({
"messages": [("user", "Investiga sobre agents en IA")],
"sources": ["langchain.com"],
})
print(result["messages"][-1].content)
# Output esperado: [Respuesta mencionando fuentes nuevas sin repetir langchain.com]
Explicación: El prompt lista las fuentes ya consultadas para evitar redundancia. En una aplicación real, actualizarías sources después de cada búsqueda.
Ejercicio 3: Estado complejo para e-commerce (Medio)
Diseña un state_schema con carrito (list), presupuesto (float), y checkout pendiente (bool). El prompt dinámico debe adaptar recomendaciones al presupuesto restante.
Ver solución
from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
import operator
class EcommerceState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
cart: list[dict]
budget: float
checkout_pending: bool
@tool
def search_products(category: str) -> str:
"""Busca productos por categoría."""
catalog = {
"electronics": "[Auriculares BT $49.99, Cargador USB-C $19.99, Webcam HD $89.99]",
"books": "[Clean Code $35.00, Design Patterns $42.00]",
}
return catalog.get(category.lower(), f"No hay productos en '{category}'")
model = ChatOpenAI(model="gpt-4.1")
def shopping_prompt(state):
cart = state.get("cart", [])
budget = state.get("budget", 0)
total_in_cart = sum(item.get("price", 0) for item in cart)
remaining = budget - total_in_cart
if state.get("checkout_pending", False):
return f"El usuario está listo para pagar. Carrito: {len(cart)} productos por ${total_in_cart:.2f}. Confirma y procede. Responde en español."
prompt = f"Eres un asistente de compras. Presupuesto: ${budget:.2f}. En carrito: {len(cart)} (${total_in_cart:.2f}). Restante: ${remaining:.2f}. "
if remaining < 20:
prompt += "Presupuesto casi agotado — solo sugiere productos baratos o pregunta si quiere checkout. "
prompt += "Responde en español."
return prompt
agent = create_agent(model, [search_products], prompt=shopping_prompt, state_schema=EcommerceState)
result = agent.invoke({
"messages": [("user", "¿Qué productos de electrónica tienen?")],
"cart": [{"name": "Clean Code", "price": 35.00}],
"budget": 100.00,
"checkout_pending": False,
})
print(result["messages"][-1].content)
# Output esperado: [Productos sugeridos dentro del presupuesto restante de $65]
Explicación: El prompt calcula dinámicamente el presupuesto restante y ajusta las recomendaciones según cuánto queda disponible.
Ejercicio 4: Conversación multi-turno con historial (Medio)
Implementa un loop de 3 turnos donde el agente recuerda contexto: pregunta sobre un tema, pide detalle, y pide resumen.
Ver solución
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
@tool
def search_topic(query: str) -> str:
"""Busca información sobre un tema."""
results = {
"langchain": "LangChain es un framework para aplicaciones con LLMs. v1.2+ incluye create_agent.",
"agents": "Los agentes usan el patrón ReAct: razonan y actúan llamando herramientas.",
}
for key, value in results.items():
if key in query.lower():
return value
return f"Información sobre '{query}': [datos generales]"
model = ChatOpenAI(model="gpt-4.1-mini")
agent = create_agent(model, [search_topic], prompt="Eres un tutor de LangChain. Recuerda la conversación. Responde en español.")
conversation_history = []
turns = ["¿Qué es LangChain?", "Cuéntame más sobre los agents", "Resume todo en 3 bullet points"]
for question in turns:
conversation_history.append(("user", question))
result = agent.invoke({"messages": conversation_history.copy()})
response = result["messages"][-1]
conversation_history = []
for msg in result["messages"]:
if hasattr(msg, 'type') and hasattr(msg, 'content'):
conversation_history.append((msg.type, msg.content if msg.content else "[tool_call]"))
print(f"User: {question}")
print(f"Agent: {response.content[:120]}...\n")
# Output esperado:
# User: ¿Qué es LangChain?
# Agent: [Explicación de LangChain]...
# User: Cuéntame más sobre los agents
# Agent: [Detalle sobre agents, construyendo sobre lo anterior]...
# User: Resume todo en 3 bullet points
# Agent: [Resumen de la conversación completa]...
Explicación: El historial se acumula pasando todos los mensajes previos. El agente resuelve referencias como "los agents que mencionaste" gracias al contexto completo.
Ejercicio 5: Distinguir state vs config en un agente real (Difícil)
Crea un agente de análisis donde el state tenga datos mutables (queries ejecutadas, resultados) y el config tenga datos inmutables (database, user_id). La tool lee del config, el prompt del state.
Ver solución
from typing import TypedDict, Annotated
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage
from langchain_core.runnables import RunnableConfig
import operator
class AnalyticsState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
queries_executed: list[str]
results_summary: list[str]
analysis_complete: bool
@tool
def run_query(sql: str, config: RunnableConfig) -> str:
"""Ejecuta una consulta SQL en la base de datos configurada."""
db = config.get("configurable", {}).get("database", "unknown")
mock = {"SELECT COUNT(*) FROM users": f"[{db}] 15,234 usuarios", "SELECT AVG(revenue) FROM sales": f"[{db}] $42.50 promedio"}
for pattern, result in mock.items():
if pattern.lower() in sql.lower():
return result
return f"[{db}] Query: {sql} — 0 resultados"
model = ChatOpenAI(model="gpt-4.1")
def analytics_prompt(state):
queries = state.get("queries_executed", [])
if state.get("analysis_complete", False):
return "Análisis completo. Genera reporte final. Responde en español."
prompt = f"Eres un analista de datos. Queries ejecutadas: {len(queries)}. "
prompt += "Ejecuta consultas para responder la pregunta. Responde en español."
return prompt
agent = create_agent(model, [run_query], prompt=analytics_prompt, state_schema=AnalyticsState)
result = agent.invoke(
{"messages": [("user", "¿Cuántos usuarios tenemos?")], "queries_executed": [], "results_summary": [], "analysis_complete": False},
config={"configurable": {"database": "production_analytics", "user_id": "analyst-maria"}},
)
print(result["messages"][-1].content)
# Output esperado: [Reporte con usuarios totales de production_analytics]
Explicación: state contiene datos mutables (queries, resultados, flag). config contiene datos inmutables (database, user_id). La tool lee del config; el prompt lee del state.
Resumen
En esta cápsula aprendiste:
- El estado por defecto de
create_agentincluye una lista demessagesque se acumula automáticamente - Puedes extender el estado con
state_schemausandoTypedDictpara agregar campos custom Annotated[list, operator.add]hace que los valores se acumulen; sinAnnotated, se reemplazan- Los campos custom típicos son: contadores, listas acumulativas, flags de control, y datos del usuario
- Los prompts dinámicos acceden al estado via
state.get("field") - Las tools acceden a datos inmutables via
config: RunnableConfig - State = datos que cambian durante la ejecución; Config = datos inmutables
- El historial de
messagesfunciona como memoria de corto plazo dentro de la sesión - Para persistencia entre sesiones, necesitas checkpointing (Módulo 8)
Próxima cápsula: Streaming de Agentes — aprenderás a observar el proceso de razonamiento del agente en tiempo real usando agent.stream().
Recursos adicionales
- State Management in LangGraph — Guía conceptual del estado
- How to create agents — Referencia oficial de create_agent
- TypedDict (Python docs) — Documentación oficial de TypedDict
- Annotated Types / Reducers — Cómo funcionan los reducers
- RunnableConfig — Documentación de config
- Messages in LangChain — Tipos de mensajes y acumulación
- Memory Concepts — Short-term vs long-term memory
- Trim Messages — Manejar historiales largos
Módulo 3 — LangChain & LangGraph: From Chains to Agents