Enrutamiento, difusión, ReAct, crítica y aprobación: cinco patrones de LangGraph
Aprenda cinco patrones de flujo de trabajo basados en agentes en LangGraph, desde routers y bucles ReAct hasta puertas de evaluación y aprobación humana, junto con las medidas de seguridad que cada uno requiere en entornos de producción.
Los prompts más extensos rara vez solucionan un problema en una función de IA poco fiable. Cuando un sistema debe navegar, escribir código, realizar verificaciones de cumplimiento o editar texto dirigido a los clientes, una sola llamada al modelo no determinista resulta demasiado frágil. La estructura ayuda: el modelo toma decisiones donde se necesita juicio, mientras que el código controla la ruta a seguir, los bucles y el cese de las operaciones. A continuación se presentan cinco patrones de este tipo, cada uno con un ejemplo ejecutable de LangGraph en Python y las consideraciones a resolver antes de su uso en producción.
Por qué un grafo es adecuado para los flujos de trabajo de agentes
Los programas convencionales se ejecutan en línea recta. Los agentes necesitan bucles, ramas condicionales y estado persistente: si el código generado falla una prueba, el sistema debe capturar el error, volver atrás e intentarlo de nuevo.
LangGraph modela esto como un grafo dirigido:
- Nodos son funciones en Python que realizan una sola tarea, como una consulta SQL o una llamada al modelo.
Para conocer más a fondo estas primitivas, consulte LangGraph en la práctica: estado, nodos y aristas.
Patrón 1: el enrutador
Un enrutador es un clasificador en el punto de entrada. En lugar de enviar todo a un modelo grande y costoso, un paso ligero dirige cada solicitud a un modelo especializado, subgráfico o herramienta local.
┌───> [Specialized Coding Agent] ───> [END]
[START] ──> [Router]
└───> [General Knowledge Agent] ───> [END]
Úselo para reducir la latencia y el costo, o para asociar intenciones con herramientas especializadas.
Un modelo pequeño a temperatura 0 etiqueta la consulta como coding o general, y add_conditional_edges asigna esa etiqueta a un nodo de procesamiento. route_decision recurre a general si el modelo devuelve algo inesperado.
from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
# 1. Define the shared state
class RouterState(TypedDict):
query: str
route: str
response: str
# Use a fast, cost-effective model for classification
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 2. Define the Nodes
def classify_query(state: RouterState):
prompt = f"""Classify the following user query into one of two categories: 'coding' or 'general'.
Respond with exactly one word, either 'coding' or 'general'.
Query: {state['query']}"""
response = model.invoke([HumanMessage(content=prompt)])
classification = response.content.strip().lower()
return {"route": classification}
def handle_coding(state: RouterState):
return {"response": "Executing advanced syntax processing and code compilation logic..."}
def handle_general(state: RouterState):
return {"response": "Processing casual conversation or general knowledge search..."}
# 3. Define Conditional Routing Logic
def route_decision(state: RouterState) -> Literal["coding", "general"]:
return state["route"] if state["route"] in ["coding", "general"] else "general"
# 4. Construct the Graph
workflow = StateGraph(RouterState)
workflow.add_node("classifier", classify_query)
workflow.add_node("coding_agent", handle_coding)
workflow.add_node("general_agent", handle_general)
workflow.add_edge(START, "classifier")
workflow.add_conditional_edges("classifier", route_decision, {
"coding": "coding_agent",
"general": "general_agent"
})
workflow.add_edge("coding_agent", END)
workflow.add_edge("general_agent", END)
# Compile and Run
app = workflow.compile()
result = app.invoke({"query": "How do I implement a binary search tree in Python?"})
print(f"Route Taken: {result['route']}\nResponse: {result['response']}")
Los nombres de los modelos eran los actuales cuando se escribió el ejemplo; sustitúyalos por los actuales de su proveedor. La salida estructurada es más robusta que el análisis de una sola palabra.
Patrón 2: orquestador y trabajadores
Para tareas demasiado amplias para una sola instrucción, un orquestador divide el objetivo en subtareas independientes, los trabajadores las completan y un sintetizador combina los resultados.
┌───> [Worker A: Section 1] ───┐
[START] ──> [Orchestrator] ├───> [Worker B: Section 2] ───┼───> [Synthesizer] ───> [END]
└───> [Worker C: Section 3] ───┘
Es adecuado para contenido extenso como informes e investigaciones con múltiples fuentes.
El orquestador solicita una lista JSON con dos subtemas; el nodo workers escribe un párrafo para cada uno, y el sintetizador los combina.
import json
from typing import TypedDict, List
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
class OrchestratorState(TypedDict):
topic: str
tasks: List[str]
worker_outputs: List[str]
final_report: str
model = ChatOpenAI(model="gpt-4o", temperature=0.2)
def orchestrator_plan(state: OrchestratorState):
prompt = f"Create a JSON list of exactly two sub-topics needed to write a comprehensive guide about: {state['topic']}. Return ONLY a valid JSON list of strings."
response = model.invoke([HumanMessage(content=prompt)])
tasks = json.loads(response.content.strip())
return {"tasks": tasks, "worker_outputs": []}
def worker_execute(state: OrchestratorState):
outputs = []
for task in state["tasks"]:
prompt = f"Write a brief, highly technical paragraph explaining: {task}"
response = model.invoke([HumanMessage(content=prompt)])
outputs.append(response.content)
return {"worker_outputs": outputs}
def synthesize_report(state: OrchestratorState):
combined_context = "\n\n".join(state["worker_outputs"])
prompt = f"Combine the following sections into a cohesive newsletter update regarding {state['topic']}:\n\n{combined_context}"
response = model.invoke([HumanMessage(content=prompt)])
return {"final_report": response.content}
# Graph Construction
orchestrator_flow = StateGraph(OrchestratorState)
orchestrator_flow.add_node("orchestrator", orchestrator_plan)
orchestrator_flow.add_node("workers", worker_execute)
orchestrator_flow.add_node("synthesizer", synthesize_report)
orchestrator_flow.add_edge(START, "orchestrator")
orchestrator_flow.add_edge("orchestrator", "workers")
orchestrator_flow.add_edge("workers", "synthesizer")
orchestrator_flow.add_edge("synthesizer", END)
app = orchestrator_flow.compile()
output = app.invoke({"topic": "Quantum Computing Security Implications"})
print(output["final_report"])
Hay dos consideraciones importantes. Este nodo worker recorre las tareas de forma secuencial, por lo que nada se ejecuta en paralelo; la API Send de LangGraph puede asignar un worker por tarea, recopilando los resultados a través de un reductor de estado. Además, a veces los modelos envuelven el JSON entre marcos Markdown, por lo que se debe validar el plan con salida estructurada en lugar de confiar en json.loads sobre texto sin procesar.
Patrón 3: ReAct, razonamiento y acción en un bucle
ReAct alterna el razonamiento con las acciones: el modelo evalúa la situación, llama a una herramienta como una búsqueda o consulta a una base de datos, observa el resultado y se detiene una vez que puede responder.
┌────────────────────────┐
▼ │
[START] ──> [Reasoner (Thought)] ───> (Should Call Tool?) ───> [Tool Executor (Act)]
│
└─ (Has Final Answer) ──> [END]
Es adecuado para agentes de investigación, soporte y depuración, donde los datos necesarios no pueden predecirse.
El razonador solicita ACTION: call_stock_api o FINAL: ..., incluyendo la última observación. La herramienta devuelve una cotización simulada, y el retorno al reasoner cierra el bucle. should_continue limita el número de iteraciones a tres.
from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
class ReActState(TypedDict):
user_input: str
agent_thought: str
tool_output: str
final_answer: str
loop_count: int
model = ChatOpenAI(model="gpt-4o", temperature=0)
def reason(state: ReActState):
loop_count = state.get("loop_count", 0) + 1
tool_context = f"\nTool Observation: {state.get('tool_output', '')}" if loop_count > 1 else ""
prompt = f"""You are a ReAct agent. Your goal is to find the current stock price of AAPL.
Current Loop: {loop_count} {tool_context}
Decide your next step. You must respond in one of two ways:
1. If you need data, say: 'ACTION: call_stock_api'
2. If you have the data, provide the answer starting with: 'FINAL: [your answer]'
User Request: {state['user_input']}"""
response = model.invoke([HumanMessage(content=prompt)]).content.strip()
if "FINAL:" in response:
return {"final_answer": response.replace("FINAL:", "").strip(), "loop_count": loop_count, "agent_thought": "done"}
else:
return {"agent_thought": "call_tool", "loop_count": loop_count}
def call_tool(state: ReActState):
print("-> System: Executing external stock database API call...")
mock_api_result = "$185.40 USD (Up 1.2% today)"
return {"tool_output": mock_api_result}
def should_continue(state: ReActState) -> Literal["call_tool", "end"]:
# Hard loop-break guardrail to prevent infinite execution loops
if state["agent_thought"] == "call_tool" and state["loop_count"] < 3:
return "call_tool"
return "end"
react_flow = StateGraph(ReActState)
react_flow.add_node("reasoner", reason)
react_flow.add_node("tool_executor", call_tool)
react_flow.add_edge(START, "reasoner")
react_flow.add_conditional_edges("reasoner", should_continue, {
"call_tool": "tool_executor",
"end": END
})
react_flow.add_edge("tool_executor", "reasoner")
app = react_flow.compile()
result = app.invoke({"user_input": "What is the market status of Apple right now?", "loop_count": 0})
print(f"\nFinal Agent Resolution:\n{result['final_answer']}")
Si se alcanza el límite antes de obtener una respuesta FINAL:, final_answer nunca se establece y el último print genera un KeyError; por lo tanto, hay que manejar ese caso. Los sistemas reales suelen utilizar llamadas a herramientas nativas en lugar de marcadores de cadena. Para el equivalente en TypeScript, consulte bucles de agente limitados para el uso de herramientas LLM.
Patrón 4: evaluador y optimizador
Un modelo que evalúa su propio trabajo tiende a aprobarlo sin cuestionamientos, por lo que un optimizador lo genera y revisa mientras que un evaluador separado y más estricto realiza la crítica.
┌───> [Optimizer (Generate/Refine)] ───> [Evaluator (Critique)]
│ │
└──────────────── (If Rejected) ────────────────┼───> [Approved] ───> [END
Es adecuado para la redacción, generación de código, traducción y reglas estrictas de calidad o normativas.
Un modelo más económico con temperatura 0.7 escribe un eslogan, incorporando los comentarios de las iteraciones posteriores. El evaluador con temperatura 0 verifica que contenga future o smart y responde en un formato fijo de ACCEPTED / FEEDBACK; routing_gate envía las rechazos de vuelta.
from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
class EvaluationState(TypedDict):
task: str
draft: str
feedback: str
accepted: bool
iterations: int
generator_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
evaluator_llm = ChatOpenAI(model="gpt-4o", temperature=0)
def generate_draft(state: EvaluationState):
iterations = state.get("iterations", 0) + 1
feedback_context = f"\nPrevious Feedback to incorporate: {state.get('feedback', '')}" if iterations > 1 else ""
prompt = f"""Write a catchy, 3-sentence marketing slogan for: '{state['task']}'.
{feedback_context}
Provide ONLY the slogan."""
response = generator_llm.invoke([HumanMessage(content=prompt)]).content.strip()
return {"draft": response, "iterations": iterations}
def evaluate_draft(state: EvaluationState):
prompt = f"""Review the following marketing slogan for the product '{state['task']}':
Slogan: "{state['draft']}"
CRITERIA: The slogan must include the exact word 'future' or 'smart'.
Respond in EXACTLY the following format:
ACCEPTED: True or False
FEEDBACK: [If rejected, explain what needs fixing. If accepted, leave blank.]"""
response = evaluator_llm.invoke([HumanMessage(content=prompt)]).content.strip()
accepted = "ACCEPTED: True" in response
feedback = response.split("FEEDBACK:")[-1].strip() if not accepted else ""
return {"accepted": accepted, "feedback": feedback}
def routing_gate(state: EvaluationState) -> Literal["refine", "approve"]:
if state["accepted"] or state["iterations"] >= 3:
return "approve"
return "refine"
eval_flow = StateGraph(EvaluationState)
eval_flow.add_node("generator", generate_draft)
eval_flow.add_node("evaluator", evaluate_draft)
eval_flow.add_edge(START, "generator")
eval_flow.add_edge("generator", "evaluator")
eval_flow.add_conditional_edges("evaluator", routing_gate, {
"refine": "generator",
"approve": END
})
app = eval_flow.compile()
result = app.invoke({"task": "Eco-friendly Electric Skateboards", "iterations": 0})
print(f"Final Slogan: {result['draft']}\nTotal Iterations: {result['iterations']}")
La puerta de control también aprueba después de tres iteraciones sin importar nada, por lo que el resultado podría no ser aceptado; mantenga la bandera accepted junto con el resultado. Una regla de palabra clave como esta es más económica y fiable si se verifica en el código.
Patrón 5: intervención humana
En operaciones riesgosas como la eliminación de tablas, el gasto de dinero o el envío de correos electrónicos a clientes, la función de puntos de control permite que el grafo se detenga antes de llegar a un nodo sensible, conserve su estado y espere la aprobación.
[START] ──> [Stager] ──> ⛔ (State Saved to DB / Graph Pauses)
│
[DevOps Manager Clicks "Approve"]
│
▼
[Executor (Run Production Deploy)] ──> [END]
Úsela para migraciones, despliegues, pagos o envíos masivos de correos electrónicos.
stager prepara una orden y executor la ejecuta. Al compilar con un punto de control y interrupt_before=["executor"], la ejecución se detiene después de la fase de preparación. Las ejecuciones se identifican mediante thread_id; por lo tanto, get_state muestra los valores guardados y el paso pendiente ('executor',); update_state registra la aprobación y invoke(None, config) reanuda la ejecución.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
class DeploymentState(TypedDict):
command: str
approved: bool
execution_log: str
# 1. Initialize thread checkpoint memory saver
memory = MemorySaver()
def stage_deployment(state: DeploymentState):
print("-> System: Staging server deployment commands...")
return {"command": "sudo systemctl restart production_api"}
def execute_deployment(state: DeploymentState):
print("-> System: Execution approved. Running command on production servers...")
return {"execution_log": f"Successfully executed: {state['command']}"}
hitl_flow = StateGraph(DeploymentState)
hitl_flow.add_node("stager", stage_deployment)
hitl_flow.add_node("executor", execute_deployment)
hitl_flow.add_edge(START, "stager")
hitl_flow.add_edge("stager", "executor")
hitl_flow.add_edge("executor", END)
# CRITICAL: Define the interrupt checkpoint before the executor node runs
app = hitl_flow.compile(checkpointer=memory, interrupt_before=["executor"])
# --- SIMULATING THE ACTIVE DEPLOYMENT WORKFLOW ---
config = {"configurable": {"thread_id": "prod_deploy_001"}}
# 1. Kick off the graph execution
initial_state = app.invoke({"command": "", "approved": False}, config)
# Verify the graph successfully halted its progress
print(f"\n[Current Graph State]: {app.get_state(config).values}")
print(f"[Next Pending Steps]: {app.get_state(config).next}") # Next step will say: ('executor',)
print("\n--- Halting Execution. Waiting for DevOps Manager Review... ---\n")
# 2. Simulate Human Reviewing the State and Updating with Approval
app.update_state(config, {"approved": True}, as_node="stager")
# 3. Resume execution thread seamlessly from the exact checkpoint
final_output = app.invoke(None, config)
print(f"[Final System Output]: {final_output['execution_log']}")
Tres precauciones. MemorySaver funciona únicamente en memoria; las pausas que persisten tras un reinicio requieren un punto de control respaldado por una base de datos. executor nunca verifica el estado approved, por lo que hay que añadir una verificación o una condición que detenga la ejecución en caso de rechazo. Las versiones más recientes de LangGraph también ofrecen una función interrupt(); consulte la documentación actual para conocer el enfoque recomendado.
Elegir un patrón
La fiabilidad proviene de adaptar la estructura al problema, no de modelos más grandes o prompts más extensos:
- Router: muchos tipos de solicitudes con diferentes necesidades de costo o habilidades.
- Orquestador y trabajadores: una tarea grande que se divide en partes independientes.
- ReAct: la información necesaria solo se puede obtener en tiempo de ejecución.
Los patrones se componen: un enrutador puede dirigir la tarea a un agente ReAct cuya acción final requiere aprobación. Mantenga tres medidas de control en todo momento: limite cada bucle, valide la salida del modelo de la que depende el código y registre si un resultado fue aprobado o se agotaron los intentos. De esta manera, el modelo no necesita ser correcto desde la primera vez, ya que el flujo de trabajo le permite enrutar, probar, intentar nuevamente y delegar en personas.
Lecturas relacionadas
- Construyendo un agente de investigación ReAct en LangGraph: Cerebro, Manos, Router — Aprenda cómo implementar el bucle de razonamiento-acción-observación ReAct como un sub-gráfico de LangGraph, con reflexión forzada, presupuestos de iteración e investigación paralela de tipo scatter-gather.
- Construyendo un agente de IA desde cero: Patrones, ReAct y LangGraph — Conozca los conceptos fundamentales detrás de los agentes de IA: planificación, uso de herramientas, reflexión y el patrón ReAct—y cómo LangChain y LangGraph se integran para crear uno manualmente.