Маршрутызацыя, розповсюджэнне, 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"])
Есць два застерэжнення. Што стосуецца вузла worker, ён адбывае задачі па порядку, таму нічога не выконваецца паралельна; 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 як падграф LangGraph, з прымусовым аналізам, лімітамі ітерацый і паралельным даследжэнням у формате «рассеянне-збіранне».
- Стварэнне агента AI з нуля: шаблоны, ReAct і LangGraph — Пазнайце основныя концэпцыі, якія лежачы ў основе агентаў AI — планаванне, викорыстоўванне інструментаў, аналіз і шаблон ReAct — а таксама як LangChain і LangGraph дапамагаюць у ручнай стварэнні такога агента.