Łączenie ścieżek, rozprzestrzenianie się, ReAct, krytyka i zatwierdzenie: pięć wzorców LangGraph
Przyswoj pięć wzorców pracy agentowej w LangGraph – od routerów i pętli ReAct po bramy ewaluacyjne oraz zatwierdzenie przez człowieka – wraz z zasadami bezpieczeństwa niezbędnymi dla każdego z nich w środowisku produkcyjnym.
Dłuższe polecenia rzadko naprawiają niewiarygodną funkcję sztucznej inteligencji. Gdy system musi przeglądać dane, pisać kod, wykonywać kontrole zgodności lub edytować tekst przeznaczony dla klientów, jedna nieokreślona próba uruchomienia modelu jest zbyt nietrwała. Struktura pomaga: model podejmuje decyzje tam, gdzie potrzebna jest ocena, podczas gdy kod kontroluje ścieżki przepływu, pętle i zatrzymywanie procesu. Poniżej przedstawiono pięć takich wzorców, z przykładami działających rozwiązań LangGraph w języku Python oraz uwagami dotyczącymi korekt przed wprowadzeniem do produkcji.
Dlaczego grafy pasują do procesów agentów
Konwencjonalne programy działają w linii prostej. Agenci potrzebują pętli, warunkowych gałęzi oraz trwałego stanu: jeśli wygenerowany kod nie przejdzie testu, system musi zarejestrować błąd, wrócić i spróbować ponownie.
LangGraph przedstawia to jako graf skierowany:
- Węzły to funkcje w Pythonie, które wykonują jedną czynność, taką jak zapytanie SQL lub wywołanie modelu.
Aby lepiej poznać te elementy, zapoznaj się z LangGraph w praktyce: stan, węzły i krawędzie.
Wzorzec 1: router
Router to klasyfikator znajdujący się w punkcie wejścia. Zamiast wysyłać wszystko do dużego, drogiego modelu, lekki element kieruje każdą prośbę do specjalistycznego modelu, podgrafu lub lokalnego narzędzia.
┌───> [Specialized Coding Agent] ───> [END]
[START] ──> [Router]
└───> [General Knowledge Agent] ───> [END]
Można go użyć, aby zmniejszyć opóźnienia i koszty lub dopasować intencje do specjalistycznych narzędzi.
Niewielki model przy temperaturze 0 oznacza zapytanie jako coding lub general, a funkcja add_conditional_edges mapuje to oznaczenie na węzeł obsługujący. Jeśli model zwróci coś nieoczekiwanego, route_decision przechodzi na wartość 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']}")
Nazwy modeli były aktualne w momencie napisania przykładu; należy je zastąpić aktualnymi nazwami dostawcy. Strukturyzowany wynik jest bardziej odporny niż analiza pojedynczego słowa.
Wzorzec 2: orkiestrator i roboty
W przypadku zadań zbyt szerokich na jedno polecenie, orkiestrator dzieli cel na niezależne podzadania, roboty je wykonują, a syntezator łączy uzyskane wyniki.
┌───> [Worker A: Section 1] ───┐
[START] ──> [Orchestrator] ├───> [Worker B: Section 2] ───┼───> [Synthesizer] ───> [END]
└───> [Worker C: Section 3] ───┘
Podoba się to do treści dłuższych, takich jak raporty, oraz badań opartych na wielu źródłach.
Orkiestrator żąda listy JSON zawierającej dwa podtematy, węzeł workers pisze akapit dla każdego z nich, a syntezator łączy je.
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"])
Dwie uwagi. Ten węzeł pracowników przetwarza zadania sekwencyjnie, więc nic nie działa równolegle; API Send LangGraph może przydzielić jeden pracownika na każde zadanie, zbierając wyniki za pomocą reduktora stanu. Ponadto modele czasami otaczają JSON ramkami Markdown, więc należy zweryfikować plan przy użyciu strukturyzowanego wyjścia, zamiast polegać na json.loads w przypadku surowego tekstu.
Wzorzec 3: ReAct – rozumowanie i działanie w pętli
ReAct naprzemiennie przeprowadza rozumowanie i podejmuje działania: model ocenia sytuację, wywołuje narzędzie takie jak wyszukiwanie lub zapytanie do bazy danych, obserwuje wynik i przestaje, gdy może udzielić odpowiedzi.
┌────────────────────────┐
▼ │
[START] ──> [Reasoner (Thought)] ───> (Should Call Tool?) ───> [Tool Executor (Act)]
│
└─ (Has Final Answer) ──> [END]
Podoba się to agentom do badań, wsparcia i debugowania, gdzie potrzebne dane nie mogą być przewidziane.
System rozumowania żąda ACTION: call_stock_api lub FINAL: ..., włączając ostatnią obserwację. Narzędzie zwraca symulowany koszt, a powrót do reasoner zamyka pętlę. Wartość should_continue ogranicza liczbę iteracji do trzech.
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']}")
Jeśli limit zostanie osiągnięty przed uzyskaniem odpowiedzi FINAL:, final_answer nigdy nie zostanie ustawiony, a ostatnie wywołanie print powoduje błąd KeyError, więc należy obsłużyć ten przypadek. Prawdziwe systemy zazwyczaj używają bezpośredniego wywoływania narzędzi zamiast znaczników tekstowych. Odnośnik do odpowiednika w TypeScript znajduje się pod adresem bounded agentic loops for LLM tool use.
Wzorzec 4: ewaluator i optymalizator
Model oceniający własną pracę ma tendencję do jej bezkrytycznego zatwierdzania, dlatego optymalizator tworzy i modyfikuje teksty, podczas gdy oddzielny, bardziej rygorystyczny oceniający dokonuje krytyki.
┌───> [Optimizer (Generate/Refine)] ───> [Evaluator (Critique)]
│ │
└──────────────── (If Rejected) ────────────────┼───> [Approved] ───> [END
Podoba się to do tworzenia projektów, generowania kodu, tłumaczeń oraz w przypadku ścisłych zasad jakości lub regulacyjnych. Tańszy model przy temperaturze 0,7 pisze hasło, uwzględniając informacje zwrotne z późniejszych iteracji. Oceniający przy temperaturze 0 sprawdza, czy tekst zawiera słowa future lub smart, i odpowiada w ustalonym formacie ACCEPTED / FEEDBACK; routing_gate odsyła teksty odrzucone.
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']}")
Brama zatwierdza wynik również po trzech iteracjach, więc rezultat może zostać odrzucony; należy zachować flagę accepted razem z wynikiem. Taka reguła oparta na słowach kluczowych jest tańsza i bardziej niezawodna, jeśli zostanie wdrożona w kodzie.
Wzorzec 5: człowiek w pętli
W przypadku ryzykownych operacji, takich jak usuwanie tabel, wydawanie pieniędzy lub wysyłanie e-maili do klientów, mechanizm checkpointingu pozwala grafowi zatrzymać się przed wrażliwym węzłem, zachować stan i czekać na zatwierdzenie.
[START] ──> [Stager] ──> ⛔ (State Saved to DB / Graph Pauses)
│
[DevOps Manager Clicks "Approve"]
│
▼
[Executor (Run Production Deploy)] ──> [END]
Wykorzystaj go do migracji, wdrażania, płatności lub masowej wysyłki e-maili.
stager przygotowuje polecenie, a executor je wykonuje. Kompilacja z checkpointerem i ustawieniem interrupt_before=[„executor“] zatrzymuje proces po etapie przygotowawczym. Wykonania są identyfikowane za pomocą thread_id, więc get_state pokazuje zapisane wartości oraz oczekujący krok ('executor',); update_state rejestruje zatwierdzenie, a invoke(None, config) wznowia działanie.
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']}")
Trzy zastrzeżenia. MemorySaver działa wyłącznie w pamięci; przerwy, które przetrwają restarty, wymagają wsparcia bazą danych w postaci punktu kontrolnego. executor nigdy nie sprawdza pola approved, więc należy dodać taką weryfikację lub warunek, który zakończy wykonywanie w przypadku odrzucenia. Nowsze wersje LangGraph oferują również funkcję interrupt(), dlatego sprawdź aktualną dokumentację w celu znalezienia zalecanej metody.
Wybór wzorca
Niezawodność wynika z dopasowania struktury do problemu, a nie z większych modeli czy dłuższych poleceń:
- Router: wiele typów zapytań o różnych wymaganiach pod względem kosztu lub umiejętności.
- Orchestrator i pracownicy: duża zadanie podzielone na niezależne części.
- ReAct: wymagane informacje można uzyskać jedynie w czasie wykonywania.
Wzory są komponowane: router może kierować zadanie do agenta ReAct, którego ostateczne działanie wymaga zatwierdzenia. We wszystkich miejscach należy zachować trzy zasady bezpieczeństwa: ograniczyć czas każdego cyklu, zweryfikować wynik modelu, od którego zależy kod, oraz odnotować, czy rezultat został zatwierdzony, czy też wyczerpano próby. Dzięki temu model nie musi być poprawny od razu, ponieważ proces pracy umożliwia mu kierowanie zadaniami, ich testowanie, ponawianie prób oraz przekazywanie ich ludziom do rozpatrzenia.
Literatura pokrewna
- Budowanie agenta badawczego ReAct w LangGraph: Brain, Hands, Router — Dowiedz się, jak zaimplementować pętlę ReAct (reason-act-observe) jako podgraf LangGraph, z wymuszoną refleksją, budżetami iteracji oraz równoległymi metodami badawczymi typu scatter-gather.
- Budowanie agenta AI od zera: wzorce, ReAct i LangGraph — Poznaj podstawowe koncepcje leżące u podstaw agentów AI — planowanie, używanie narzędzi, refleksja oraz wzorzec ReAct — oraz to, jak LangChain i LangGraph wpisują się w proces ręcznego tworzenia takiego agenta.