Маршрутизація, розповсюдження, 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, дивіться bounded agentic loops for LLM tool use.
Шаблон 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: мозок, руки, маршрутизатор — Дізнайтеся, як реалізувати цикл розумових дій ReAct (reason-act-observe) у вигляді підграфа LangGraph, з примусовим відображенням інформації, обмеженнями на кількість ітерацій та паралельними методами збору даних.
- Створення AI-агента з нуля: шаблони, ReAct та LangGraph — Ознайомтесь із основними концепціями AI-агентів — плануванням, використанням інструментів, відображенням інформації та шаблоном ReAct — а також із тим, як LangChain та LangGraph допомагають у ручній розробці такого агента.