Proyecto Iniciante: Agente de Q&A con Uso de Herramientas y Enrutamiento
Construye un agente de Q&A completo usando LangGraph con uso de herramientas, enrutamiento condicional y gestión de estado.
Proyecto Iniciante: Agente de Q&A con Uso de Herramientas y Enrutamiento
En este proyecto final, construirás un agente de Q&A completo que puede responder preguntas usando herramientas y enrutamiento inteligente. Esto reúne todo lo que has aprendido: estado, nodos, aristas, LLMs, herramientas y ramificación condicional.
Visión General del Proyecto
El agente:
- Clasificará la pregunta del usuario en una categoría
- Enrutará al manipulador apropiado basándose en la categoría
- Ejecutará herramientas o generará respuestas según sea necesario
- Repetirá si se llamaron herramientas, para procesar resultados
- Devolverá una respuesta final
Paso 1: Configuración e Importaciones
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolExecutor
from typing_extensions import TypedDict, List, Annotated
from typing import Any
from operator import add
import jsonPaso 2: Definir Herramientas
@tool
def web_search(query: str) -> str:
"""Buscar en la web información actual. Usar para noticias, hechos y conocimiento general."""
return f"## Resultados Web para '{query}'\n" \
f"1. LangGraph es un framework para construir agentes de IA con estado.\n" \
f"2. Fue creado por LangChain Inc.\n" \
f"3. La versión 0.2 fue lanzada en 2024."
@tool
def calculator(expression: str) -> str:
"""Evaluar una expresión matemática. Usar sintaxis Python: +, -, *, /, **, //, %."""
try:
safe_dict = {"__builtins__": {}}
result = eval(expression, safe_dict)
return str(result)
except Exception as e:
return f"Error de cálculo: {e}"
@tool
def get_timezone(city: str) -> str:
"""Obtener la zona horaria para una ciudad determinada."""
timezones = {
"new york": "America/New_York (UTC-5)",
"london": "Europe/London (UTC+0)",
"tokyo": "Asia/Tokyo (UTC+9)",
"paris": "Europe/Paris (UTC+1)",
"sydney": "Australia/Sydney (UTC+11)"
}
return timezones.get(city.lower(), f"Zona horaria no encontrada para {city}")Paso 3: Inicializar LLM y Enlaces de Herramientas
llm = ChatOpenAI(model="gpt-4o-mini")
all_tools = [web_search, calculator, get_timezone]
llm_with_tools = llm.bind_tools(all_tools)
tool_executor = ToolExecutor(all_tools)Paso 4: Definir Estado
class AgentState(TypedDict):
messages: Annotated[List[Any], add]
question: str
category: str
tool_results: List[str]
final_answer: str
error: strPaso 5: Definir Nodos
def classify_question(state: AgentState) -> dict:
prompt = ChatPromptTemplate.from_messages([
("system", "Clasifica la pregunta en exactamente una categoría:\n"
"- 'web_search': preguntas sobre noticias, hechos, personas, conceptos\n"
"- 'calculator': problemas matemáticos, cálculos, ecuaciones\n"
"- 'chat': conversación general, opiniones, explicaciones\n\n"
"Responde SOLO con el nombre de la categoría."),
("human", "{question}")
])
chain = prompt | llm | StrOutputParser()
category = chain.invoke({"question": state["question"]}).strip().lower()
if category not in ["web_search", "calculator", "chat"]:
category = "chat"
return {"category": category, "messages": [HumanMessage(state["question"])]}
def web_search_node(state: AgentState) -> dict:
try:
result = tool_executor.invoke({
"name": "web_search",
"args": {"query": state["question"]},
"id": "search_1",
"type": "tool_call"
})
return {
"tool_results": state["tool_results"] + [str(result)],
"messages": [ToolMessage(content=str(result), tool_call_id="search_1")]
}
except Exception as e:
return {"error": f"Búsqueda web falló: {e}"}
def calculator_node(state: AgentState) -> dict:
try:
result = tool_executor.invoke({
"name": "calculator",
"args": {"expression": state["question"]},
"id": "calc_1",
"type": "tool_call"
})
return {
"tool_results": state["tool_results"] + [str(result)],
"messages": [ToolMessage(content=str(result), tool_call_id="calc_1")]
}
except Exception as e:
return {"error": f"Calculadora falló: {e}"}
def agent_node(state: AgentState) -> dict:
context = "\n".join(state["tool_results"]) if state["tool_results"] else "No se necesitaron herramientas."
prompt = ChatPromptTemplate.from_messages([
("system", "Eres un asistente de Q&A útil. Responde a la pregunta del usuario "
"basándote en el contexto disponible. Sé conciso y preciso.\n\n"
"Contexto:\n{context}"),
("human", "{question}")
])
chain = prompt | llm | StrOutputParser()
answer = chain.invoke({"context": context, "question": state["question"]})
return {"final_answer": answer, "messages": [AIMessage(content=answer)]}Paso 6: Construir el Grafo
builder = StateGraph(AgentState)
builder.add_node("classify", classify_question)
builder.add_node("web_search", web_search_node)
builder.add_node("calculator", calculator_node)
builder.add_node("agent", agent_node)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", lambda s: s["category"], {
"web_search": "web_search",
"calculator": "calculator",
"chat": "agent"
})
builder.add_edge("web_search", "agent")
builder.add_edge("calculator", "agent")
builder.add_edge("agent", END)
app = builder.compile()Paso 7: Ejecutar el Agente
def ask_question(question: str) -> str:
result = app.invoke({
"messages": [],
"question": question,
"category": "",
"tool_results": [],
"final_answer": "",
"error": ""
})
return result["final_answer"]
print(ask_question("¿Qué es LangGraph?"))
print(ask_question("¿Cuánto es 245 * 37?"))
print(ask_question("¿Qué zona horaria tiene Tokio?"))
print(ask_question("¿Cuál es el sentido de la vida?"))
print(ask_question("Resuelve: (15 + 7) * 2"))Tu agente ahora enruta preguntas inteligentemente a la herramienta correcta y sintetiza una respuesta final. Este es el patrón central detrás de todos los agentes LangGraph.
Preguntas de Práctica
¿Cuál es el primer nodo ejecutado en el proyecto del agente de Q&A?
¿Qué sucede después de que un nodo de herramienta (web_search o calculator) se ejecuta?
¿Cuál es la categoría de fallback por defecto en el nodo de clasificación?
¿Qué reductor de estado se usa para el campo messages?
¿Por qué deberías establecer recursion_limit al invocar el grafo?
¿Qué tipo de arista determina qué nodo de herramienta se ejecuta?
¿Cómo accede el nodo agente a las salidas de las herramientas?
¿A qué categoría se enruta una pregunta matemática como '¿Cuánto es 2+2?'?
¿Cuál es el propósito de envolver app.invoke() en try/except?
¿Qué hace ToolExecutor en este proyecto?
Puntos Clave
- El agente de Q&A usa clasificación → enrutamiento → ejecución de herramientas → síntesis
- El estado fluye a través de cada nodo, acumulando resultados a lo largo del camino
- El reductor
adden messages anexa al historial de la conversación - Las aristas condicionales permiten enrutamiento inteligente basado en la categoría
- El nodo agente sintetiza resultados de herramientas en una respuesta final coherente
- Usa siempre try/except y recursion_limit para manejo robusto de errores
- Esta arquitectura es la base para todos los agentes LangGraph de producción