Маршрутизация, рассеивание, ReAct, критика и одобрение: пять шаблонов LangGraph
Изучите пять шаблонов рабочих процессов с использованием агентов в LangGraph: от маршрутизаторов и циклов ReAct до ворот оценки и утверждения человеком, а также необходимые ограничения для каждого из них в производственных условиях.
Длинные запросы редко помогают устранить проблемы с ненадежными функциями ИИ. Когда системе необходимо искать информацию, писать код, выполнять проверки соответствия или редактировать текст для клиентов, один недетерминистичный вызов модели слишком нестабилен. Структурированный подход помогает: модель принимает решения там, где требуется оценка ситуации, в то время как код контролирует логику выполнения, циклы и момент остановки. Ниже приведены пять таких шаблонов, каждый из которых сопровождается рабочим примером на LangGraph на Python, а также указаны моменты, которые необходимо устранить перед внедрением в производство.
Почему графы подходят для рабочих процессов агентов
Традиционные программы выполняются по прямой линии. Агентам же нужны циклы, условные ветвления и постоянное хранение состояния: если сгенерированный код не проходит тест, системе необходимо зафиксировать ошибку, вернуться назад и попробовать снова.
LangGraph моделирует это как ориентированный граф:
- Узлы — это функции на Python, выполняющие одну задачу, такую как запрос к SQL-базе или вызов модели.
Чтобы узнать больше об этих примитивах, ознакомьтесь с LangGraph на практике: состояние, узлы и рёбра.
Шаблон 1: маршрутизатор
Маршрутизатор представляет собой классификатор на точке входа. Вместо того чтобы отправлять всё к большой и дорогой модели, легковесный компонент направляет каждый запрос к специализированной модели, подграфу или локальному инструменту.
┌───> [Specialized Coding Agent] ───> [END]
[START] ──> [Router]
└───> [General Knowledge Agent] ───> [END]
Используйте его для снижения задержек и затрат или для соответствия намеренийм специализированным инструментам.
Небольшая модель при температуре 0 присваивает запросу метку coding или general, а функция add_conditional_edges соотносит эту метку с узлом-обрабатчиком. Если модель возвращает что-то неожиданное, функция route_decision переходит на использование метки general.
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']}")
Имена моделей были актуальны при написании примера; замените их на текущие названия вашего поставщика. Структурированный вывод более надежен, чем обработка одного слова.
Шаблон 2: оркестратор и рабочие процессы
Для задач, слишком обширных для одного запроса, оркестратор разделяет цель на независимые подзадачи, рабочие процессы их выполняют, а синтезатор объединяет результаты.
┌───> [Worker A: Section 1] ───┐
[START] ──> [Orchestrator] ├───> [Worker B: Section 2] ───┼───> [Synthesizer] ───> [END]
└───> [Worker C: Section 3] ───┘
Этот подход подходит для длинных текстов, таких как отчеты, и исследований с использованием нескольких источников.
Оркестратор запрашивает JSON-список из двух подтем, узел workers пишет абзац для каждой из них, а синтезатор объединяет их.
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"])
Два важных момента. Этот узел-рабочий обрабатывает задачи последовательно, поэтому ничего не выполняется параллельно; API Send LangGraph может направлять по одному рабочему на задачу, собирая результаты с помощью редюсера состояния. Кроме того, модели иногда заключают JSON в рамки Markdown, поэтому необходимо проверять план с использованием структурированного вывода, а не полагаться на json.loads для обработки сырого текста.
Шаблон 3: ReAct — рассуждения и действия в цикле
ReAct чередует рассуждения с действиями: модель оценивает ситуацию, вызывает инструмент, такой как поиск или запрос к базе данных, наблюдает за результатом и прекращает работу, как только может дать ответ.
┌────────────────────────┐
▼ │
[START] ──> [Reasoner (Thought)] ───> (Should Call Tool?) ───> [Tool Executor (Act)]
│
└─ (Has Final Answer) ──> [END]
Этот подход подходит для агентов, занимающихся исследованиями, поддержкой и отладкой, в случаях, когда необходимые данные невозможно предсказать.
Механизм рассуждений запрашивает ACTION: call_stock_api или FINAL: ..., включая последнее наблюдение. Инструмент возвращает симулированные котировки, а возврат к reasoner закрывает цикл. Параметр should_continue ограничивает количество итераций тремя.
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']}")
Если лимит достигнут до получения ответа FINAL:, переменная final_answer так и не устанавливается, и последняя команда print вызывает ошибку KeyError; поэтому необходимо обработать этот случай. В реальных системах обычно используется прямой вызов нативных инструментов вместо строковых маркеров. Аналог для TypeScript смотрите по ссылке ограниченные агентные циклы для использования инструментов LLM.
Шаблон 4: оценщик и оптимизатор
Модель, оценивающая собственную работу, склонна просто утверждать её без критики, поэтому оптимизатор генерирует и корректирует результаты, в то время как отдельный, более строгий оценщик проводит критику.
┌───> [Optimizer (Generate/Refine)] ───> [Evaluator (Critique)]
│ │
└──────────────── (If Rejected) ────────────────┼───> [Approved] ───> [END
Этот подход подходит для составления текстов, генерации кода, перевода и ситуаций, требующих строгих стандартов качества или соблюдения регламентов.
Более дешёвая модель при температуре 0.7 пишет слоган, учитывая обратную связь полученную на последующих этапах. Оценщик при температуре 0 проверяет, содержит ли текст слова future или smart, и отвечает в фиксированном формате ACCEPTED / FEEDBACK; модуль routing_gate возвращает отклонённые результаты.
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']}")
Этот механизм также утверждает результат после трёх итераций независимо от качества, поэтому результат может остаться непринятым; необходимо сохранять флаг accepted вместе с результатом. Подобные правила на основе ключевых слов дешевле и надёжнее реализуются в коде.
Шаблон 5: участие человека
Для рискованных операций, таких как создание таблиц, траты средств или отправка писем клиентам, использование точек контроля позволяет графу остановиться перед чувствительным узлом, сохранить состояние и дождаться одобрения.
[START] ──> [Stager] ──> ⛔ (State Saved to DB / Graph Pauses)
│
[DevOps Manager Clicks "Approve"]
│
▼
[Executor (Run Production Deploy)] ──> [END]
Используйте его для миграций, развертывания, платежей или массовой рассылки писем.
stager готовит команду, а executor выполняет её. Компиляция с использованием точки контроля и параметра interrupt_before=[„executor“] прерывает процесс после стадии подготовки. Запуски идентифицируются по thread_id; функция get_state отображает сохранённые значения и ожидающий шаг (‚executor‘,); функция update_state фиксирует одобрение, после чего процесс возобновляется с помощью invoke(None, config).
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']}")
Три предупреждения. MemorySaver работает исключительно в памяти; для пауз, сохраняющихся после перезагрузки, требуется чекпоинтер с поддержкой базы данных. executor никогда не проверяет поле approved, поэтому необходимо добавить проверку или условие, прерывающее выполнение при отклонении. В новых версиях LangGraph также предоставляется функция interrupt(), поэтому ознакомьтесь с актуальной документацией для определения рекомендуемого подхода.
Выбор паттерна
Надежность достигается за счет соответствия структуры задаче, а не за счет использования более крупных моделей или длинных запросов:
- Router: множество типов запросов с разными требованиями к затратам или навыкам.
- Orchestrator и workers: крупная задача, разделяемая на независимые части.
- ReAct: необходимая информация может быть получена только во время выполнения.
Шаблоны составляются композитно: маршрутизатор может направить задачу агенту ReAct, чье окончательное действие требует одобрения. Всегда соблюдайте три основных принципа: ограничивайте каждый цикл, проверяйте результаты работы модели, от которых зависит код, и фиксируйте, был ли результат одобрен или исчерпаны попытки. Таким образом, модели не обязательно нужно действовать правильно с первого раза, поскольку рабочий процесс позволяет ей маршрутизировать задачи, тестировать их, пытаться снова и передавать решение людям.
Связанные материалы
- Создание исследовательского агента ReAct в LangGraph: механизмы Brain, Hands и Router — Узнайте, как реализовать цикл размышления-действия-наблюдения ReAct в виде подграфа LangGraph с принудительным отражением, ограничениями на итерации и параллельным методом сбора информации.
- Создание ИИ-агента с нуля: шаблоны, подход ReAct и LangGraph — Ознакомьтесь с основными концепциями ИИ-агентов — планированием, использованием инструментов, отражением действий и шаблоном ReAct — а также с тем, как LangChain и LangGraph помогают вручную создать такой агент.