构建具备状态、流程可控的生产级 AI 智能体系统

2026 年 7 月 · 阅读时长 15 分钟*

如果 LangChain 的线性流水线已经无法满足你的需求，而你需要真正掌控 AI 智能体如何思考、分支、循环和恢复，那么 LangGraph 正是你一直在等待的框架。本指南涵盖从核心概念到生产实践模式的全部内容，并在全文提供可运行的代码。

什么是 LangGraph，它为何会出现？

核心概念：节点、边与状态

你的第一个 LangGraph 应用

条件边与分支

循环——真正的威力所在

深入理解状态管理

人在回路模式

多智能体架构

持久化与检查点

流式处理与实时输出

生产环境中的 LangGraph

LangGraph 与替代方案对比

第 1 部分——什么是 LangGraph，它为何会出现？ {#part-1}

LangGraph 是一个构建于 LangChain 之上的框架，用于创建以有向图建模、具备状态的多步骤 AI 工作流。它源于一个看似简单的问题：现实世界中的 AI 任务并不是沿直线运行的。

LangChain 的链是线性的——它们按照 A → B → C 的顺序执行，然后终止。这种模型适合演示，但在生产环境中会失效，因为智能体需要重试失败的工具调用、根据 LLM 的决策进行分支、暂停并等待人工审核，还要在持续数小时甚至数天的会话之间维护上下文。

其核心洞见是：智能体本质上就是一个循环——感知世界、作出决策、采取行动、观察结果，然后重复。LangGraph 让这个循环变得显式、可观察且可控制。

LangGraph 解决的五个问题

失败重试：当工具调用返回错误时，智能体可以回到之前的节点并尝试另一种方法，而不是让整条链崩溃。

动态分支：LLM 分类器节点可以将请求路由给账单专家、技术支持流程或通用处理程序——路由在运行时决定，而不是在设计阶段硬编码。

人在回路：执行过程会在预先定义的检查点暂停，只有在人工审核并批准后才会继续——这对于受监管行业和高风险决策至关重要。

长时间运行的会话：状态会以检查点形式保存到数据库，因此智能体在崩溃、重启或暂停数天后，可以准确地从中断位置继续执行。

多智能体协作：可以将多个专业化子智能体组合成一个统一编排的系统，同时在它们之间保持清晰的状态边界。

第 2 部分——核心概念：节点、边与状态 {#part-2}

LangGraph 有三个基本原语。掌握它们之后，你就能一眼看懂本指南中的所有模式。

状态是一个带类型的 Python 字典，由图中的所有节点共享。每个节点都会读取状态、执行工作，然后返回部分更新。Annotated 类型提示控制更新的合并方式——其中最重要的模式是 add_messages，它会将消息追加到消息列表中，而不是替换该列表。

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class AgentState(TypedDict):
    # add_messages = APPEND each update, not replace
    # Without it: each node would overwrite history
    messages: Annotated[list, add_messages]
    task: str
    result: str | None
    attempts: int
    approved: bool

为什么 add_messages 很重要：如果没有它，每个返回 {"messages": [new_msg]} 的节点都会替换全部历史记录。使用 add_messages 后，新消息会被追加进去——无需任何额外的记录管理，会话历史便会自然累积。

节点就是普通的 Python 函数。它们接收当前状态并执行工作，例如调用 LLM、执行工具或运行验证逻辑，然后返回部分状态更新——只需返回发生变化的键。

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-opus-4-6")

def analyst_node(state: AgentState) -> dict:
    """Analyzes the task and produces a structured plan."""
    response = llm.invoke([
        {"role": "system", "content": "You are a senior analyst. Break this task into steps."},
        {"role": "user",   "content": f"Task: {state['task']}"}
    ])
    # Return ONLY the keys that changed — LangGraph merges for you
    return {
        "messages": [response],   # appended via add_messages
        "result": response.content
    }

def validator_node(state: AgentState) -> dict:
    """Quality-checks the result and increments attempt counter."""
    response = llm.invoke([
        {"role": "system", "content": "Score this output 1–10. Reply with only a number."},
        {"role": "user",   "content": state["result"]}
    ])
    score = int(response.content.strip())
    # Only return what changed — don't touch messages or task
    return {
        "attempts": state["attempts"] + 1,
        "approved": score >= 8
    }

边定义控制流。普通边始终按照 A → B 路由。条件边则会调用一个路由函数，由该函数返回一个字符串键，并据此将执行过程引导至多个候选后续节点中的一个。

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# Normal edge — always goes analyst → validator
workflow.add_edge("analyst", "validator")

# Conditional edge — routing function returns next node key
def route_after_validation(state: AgentState) -> str:
    if state["attempts"] >= 3:
        return END               # Hard limit: give up after 3 tries
    if state["approved"]:
        return "writer"          # Quality met, proceed
    return "analyst"             # Low quality, retry

workflow.add_conditional_edges(
    "validator",
    route_after_validation,
    # Map of return values → destination nodes
    {END: END, "writer": "writer", "analyst": "analyst"}
)

第 3 部分——你的第一个 LangGraph 应用 {#part-3}

理论只有在实际构建东西时才能体现其价值。下面是一个完整的 SQL 查询生成智能体——它将自然语言问题转换为 SQL，验证查询的正确性和安全性，并持续迭代，直到达到质量标准或用完三次尝试机会。

图：生成器—验证器—修订循环——持续迭代，直到 SQL 质量评分 ≥ 8，最多尝试 3 次。

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END, add_messages
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage

llm       = ChatAnthropic(model="claude-opus-4-6", temperature=0.2)
judge_llm = ChatAnthropic(model="claude-haiku-4-5-20251001", temperature=0)

# ── State ──────────────────────────────────────────────────────────────
class SQLState(TypedDict):
    messages:      Annotated[list, add_messages]
    question:      str      # natural-language question from user
    schema:        str      # database schema context
    sql_query:     str      # generated SQL
    quality_score: int      # 1–10 from validator
    feedback:      str      # specific improvement hint
    iterations:    int

# ── Nodes ──────────────────────────────────────────────────────────────
def generator_node(state: SQLState) -> dict:
    """Translates a natural-language question into a SQL query."""
    prev_feedback = state.get("feedback", "")
    prompt = f"""You are an expert SQL engineer. Write a single correct SQL query.

Database schema:
{state['schema']}

Question: {state['question']}

{f"Previous attempt failed. Fix this: {prev_feedback}" if prev_feedback else ""}

Return ONLY the SQL query — no explanation, no markdown fences."""

response = llm.invoke([HumanMessage(content=prompt)])
    iteration = state.get("iterations", 0) + 1
    return {
        "sql_query":  response.content.strip(),
        "iterations": iteration,
        "messages":   [AIMessage(content=f"Query attempt {iteration} generated")]
    }

def validator_node(state: SQLState) -> dict:
    """Checks the SQL for correctness, safety, and efficiency."""
    prompt = f"""You are a senior DBA. Review this SQL query against the schema.

Schema:
{state['schema']}

Query:
{state['sql_query']}

Score it 1–10 (10 = production-ready). Check for:
- Correctness (joins, filters match the schema)
- Safety (no DROP/DELETE without WHERE, no unbounded full-table scans)
- Efficiency (proper use of indexes, avoid SELECT *)

Format exactly:
SCORE: <number>
FEEDBACK: <one specific fix if score < 8, else "Looks good">"""

response = judge_llm.invoke([HumanMessage(content=prompt)])
    lines    = response.content.strip().split('\n')
    score    = int(lines[0].replace('SCORE:', '').strip())
    feedback = lines[1].replace('FEEDBACK:', '').strip()
    return {"quality_score": score, "feedback": feedback}

# ── Routing function ───────────────────────────────────────────────────
def should_continue(state: SQLState) -> str:
    if state["quality_score"] >= 8:  return "approved"   # Production-ready
    if state["iterations"]    >= 3:  return "approved"   # Best effort after 3 tries
    return "revise"

── 构建图 ────────────────────────────────────────────────────────

workflow = StateGraph(SQLState)
workflow.add_node("generator", generator_node)
workflow.add_node("validator", validator_node)

workflow.set_entry_point("generator")
workflow.add_edge("generator", "validator")
workflow.add_conditional_edges(
    "validator",
    should_continue,
    {"approved": END, "revise": "generator"}
)

app = workflow.compile()

── 运行 ────────────────────────────────────────────────────────────────

SCHEMA = """
  users(id, name, email, created_at, plan)
  orders(id, user_id, total, status, created_at)
  order_items(id, order_id, product_id, quantity, price)
"""

result = app.invoke({
    "question":      "Find the top 5 users by total spend in the last 30 days",
    "schema":        SCHEMA,
    "sql_query":     "",
    "quality_score": 0,
    "feedback":      "",
    "iterations":    0,
    "messages":      []
})
print(f"Final SQL (score {result['quality_score']}/10):\n{result['sql_query']}")

关键模式：使用快速、廉价的模型（Haiku）作为 DBA 验证器，使用有能力的模型（Opus）作为生成器。验证器的工作是结构化评分——它不需要前沿推理，这通过多次反馈迭代保持成本低。

第 4 部分 — 条件边与分支 {#part-4}

条件边是 LangGraph 实现决策制定的地方。路由函数可以由 LLM 输出、状态值、业务逻辑或任意组合驱动。这是一个客户支持路由器，用于分类意图并分配给正确的处理器。

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END

class SupportState(TypedDict):
    user_message: str
    intent:       str
    response:     str

def classify_intent(state: SupportState) -> dict:
    result = llm.invoke(f"""Classify into ONE category:
    billing | technical | general | escalate

Message: {state['user_message']}
    Reply with only the category word.""")
    return {"intent": result.content.strip().lower()}

# 路由函数——返回值必须与边映射键匹配
def route_by_intent(state: SupportState) -> Literal["billing","technical","general","escalate"]:
    return state["intent"]

def billing_node(state):   return {"response": "Connecting to billing specialist..."}
def technical_node(state): return {"response": "Looking up technical documentation..."}
def escalate_node(state):  return {"response": "Transferring to senior agent."}
def general_node(state):
    answer = llm.invoke(f"Answer: {state['user_message']}").content
    return {"response": answer}

# 构建
workflow = StateGraph(SupportState)
for name, fn in [("classifier", classify_intent), ("billing", billing_node),
                  ("technical", technical_node), ("general", general_node),
                  ("escalate", escalate_node)]:
    workflow.add_node(name, fn)

workflow.set_entry_point("classifier")
workflow.add_conditional_edges(
    "classifier", route_by_intent,
    {"billing":"billing", "technical":"technical",
     "general":"general", "escalate":"escalate"}
)
# 所有分支汇聚到 END
for node in ["billing", "technical", "general", "escalate"]:
    workflow.add_edge(node, END)

support_app = workflow.compile()

常见错误：路由函数必须返回与边映射字典中某个键完全匹配的值。如果 LLM 返回 "billing" 但你的映射有 "Billing"，图会抛出 GraphValueError。路由前始终对 LLM 输出调用 .strip().lower()。

第 5 部分 — 循环与迭代 — 真正的力量 {#part-5}

循环是使 LangGraph 从所有线性替代方案中根本不同之处。一个智能体可以持续工作直到达成目标——搜索直到有足够信息、生成直到质量满足、重试直到工具成功。

模式总是一样的：添加一个条件边，其路由函数可以返回一个已经运行过的节点。图的循环检测器在设计上允许这一点。

from typing import TypedDict

class ResearchState(TypedDict):
    question:       str
    search_results: list[str]
    analysis:       str
    is_sufficient:  bool
    search_count:   int

def search_node(state: ResearchState) -> dict:
    """生成一个针对性的搜索查询并执行。"""
    query = llm.invoke(f"""Generate a specific search query for: {state['question']}
Searches done so far: {state['search_count']}
Last result: {state['search_results'][-1] if state['search_results'] else 'none'}
Make this query different from previous ones.""").content

    result = execute_search(query)   # 替换为真实的搜索 API
    return {
        "search_results": state["search_results"] + [result],
        "search_count":   state["search_count"] + 1
    }

def analyze_node(state: ResearchState) -> dict:
    """判断我们是否有足够的信息来回答。"""
    assessment = llm.invoke(f"""Question: {state['question']}

Research gathered ({len(state['search_results'])} searches):
{chr(10).join(state['search_results'])}

Do you have sufficient information?
Reply: SUFFICIENT or INSUFFICIENT, then your analysis.""").content

    is_sufficient = assessment.startswith("SUFFICIENT")
    analysis = assessment.split('\n', 1)[1] if '\n' in assessment else assessment
    return {"is_sufficient": is_sufficient, "analysis": analysis}

def should_search_more(state: ResearchState) -> str:
    if state["search_count"] >= 5:  return "done"   # 硬上限
    if state["is_sufficient"]:      return "done"
    return "search_more"

# 构建——循环由指向回节点的条件边创建
workflow = StateGraph(ResearchState)
workflow.add_node("search",  search_node)
workflow.add_node("analyze", analyze_node)
workflow.set_entry_point("search")
workflow.add_edge("search", "analyze")
workflow.add_conditional_edges(
    "analyze",
    should_search_more,
    {"done": END, "search_more": "search"}  # ← 这创建了循环
)

始终添加硬退出条件。每个循环都必须有最大迭代计数，无论 LLM 的评估如何都会终止。如果搜索结果始终无关，LLM 可能陷入 INSUFFICIENT 循环。上面 5 次搜索的硬上限是你的熔断器。

第 6 部分 — 状态管理深潜 {#part-6}

每个状态键都有一个 reducer——一个控制节点更新如何合并到现有状态的函数。默认 reducer 是最后写入获胜。理解 reducer 解锁了复杂的并行和协作节点模式。

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
import operator

# ── 模式 1：默认（最后写入获胜）──────────────────────────────
class SimpleState(TypedDict):
    status: str   # 每次更新替换该值
    count:  int   # 每次更新替换该值

# ── 模式 2：仅追加（聊天历史）─────────────────────────────
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]   # 每次更新追加

# ── 模式 3：自定义 reducer ─────────────────────────────────
def merge_results(existing: list, new: list) -> list:
    """按 ID 去重同时保留插入顺序。"""
    seen, merged = set(), []
    for item in existing + new:
        key = item.get("id") if isinstance(item, dict) else item
        if key not in seen:
            seen.add(key)
            merged.append(item)
    return merged

class PipelineState(TypedDict):
    results: Annotated[list, merge_results]   # 智能去重合并
    errors:  Annotated[list, operator.add]    # 简单追加
    status:  str                              # 最后写入获胜

# ── 模式 4：并行写入安全收敛────────────────────────────────────
# 当多个并行节点同时写入同一键时，
# LangGraph 使用所有更新调用 reducer——无竞态条件。
# 这就是使 map-reduce 模式在没有锁的情况下安全的原因。

Reducer = 免费的线程安全。当使用 Send API 扇出到并行节点并且它们都写入同一键时，LangGraph 的 reducer 自动合并结果。为列表使用 operator.add，为聊天历史使用 add_messages，或为自定义逻辑使用你自己的函数。

第 7 部分 — 人机协作模式 {#part-7}

这是 LangGraph 的标志性能力之一。执行在定义的检查点暂停，仅在人类提供输入后才恢复。这不是轮询循环——这是由检查点程序支持的一流中断机制。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END
from langgraph.types import interrupt, Command
from typing import TypedDict

class ApprovalState(TypedDict):
    task:           str
    plan:           str
    human_approved: bool
    final_result:   str

def planner_node(state: ApprovalState) -> dict:
    plan = llm.invoke(f"Create a detailed execution plan for: {state['task']}").content
    return {"plan": plan}

def human_review_node(state: ApprovalState) -> dict:
    # interrupt() 在此处暂停图。
    # 调用者看到 payload；当发送 Command(resume=...) 时恢复执行。
    human_input = interrupt({
        "question": "批准此计划？",
        "plan":     state["plan"]
    })
    return {"human_approved": human_input.get("approved", False)}

def executor_node(state: ApprovalState) -> dict:
    if not state["human_approved"]:
        return {"final_result": "任务已取消。"}
    result = llm.invoke(f"执行此计划:\n{state['plan']}").content
    return {"final_result": result}

# Checkpointer 是必需的 — 它在 interrupt 和 resume 之间存储状态
memory = MemorySaver()

workflow = StateGraph(ApprovalState)
workflow.add_node("planner",      planner_node)
workflow.add_node("human_review", human_review_node)
workflow.add_node("executor",     executor_node)
workflow.set_entry_point("planner")
workflow.add_edge("planner",      "human_review")
workflow.add_edge("human_review", "executor")
workflow.add_edge("executor",     END)

app = workflow.compile(checkpointer=memory)

# ── 步骤 1: 运行到 interrupt ──────────────────────────────────────
thread = {"configurable": {"thread_id": "deploy-001"}}
state  = app.invoke({"task": "Deploy payment service to production"}, thread)
print("计划:", state["plan"])
print("— 等待批准 —")

# ── 步骤 2: 人工审查，然后恢复 ───────────────────────────────
final = app.invoke(Command(resume={"approved": True}), thread)

interrupt_before vs interrupt_after

# interrupt_before: 在节点运行之前暂停 — 你可以先编辑状态
app = workflow.compile(checkpointer=memory, interrupt_before=["dangerous_action"])

# interrupt_after: 在节点运行之后暂停 — 继续前先审查输出
app = workflow.compile(checkpointer=memory, interrupt_after=["code_generator"])

# 动态 interrupt: 仅当节点内的风险逻辑触发时暂停
def smart_executor(state):
    result = execute_action(state)
    if result.risk_score > 0.8:
        decision = interrupt({"reason": "高风险操作", "details": result})
        if not decision["proceed"]:
            return {"status": "cancelled"}
    return {"result": result}

Checkpointer 是强制的。人工在环 (human-in-the-loop) 需要 checkpointer — 图的状态必须被持久化到某处，以便在人工恢复时检索。没有 checkpointer，interrupt() 会抛出运行时错误。开发环境使用 MemorySaver，生产环境使用 PostgresSaver。

Part 8 — 多智能体架构 {#part-8}

LangGraph 的可组合性是其生产级的超能力。每个子智能体都是一个完整编译的 LangGraph 应用。主管 (supervisor) 通过将其作为节点调用来协调它们 — 其内部复杂性对外部图是透明的。

图：主管模式 — 协调器 LLM 决定接下来调用哪个专家，每个专家报告回来。

模式 1 — 主管 / 编排器

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END, add_messages
from langchain_core.messages import AIMessage

class SupervisorState(TypedDict):
    messages:   Annotated[list, add_messages]
    task:       str
    next_agent: str

def supervisor_node(state: SupervisorState) -> dict:
    decision = llm.invoke(f"""你是项目经理。
任务: {state['task']}
历史: {state['messages']}

哪个专家接下来行动？
选项: researcher | writer | reviewer | FINISH
只回复一个单词。""").content.strip()
    return {"next_agent": decision}

def route_to_agent(state: SupervisorState) -> str:
    return END if state["next_agent"] == "FINISH" else state["next_agent"]

# 每个专家本身可以是一个编译的 LangGraph 应用
def researcher_node(state: SupervisorState) -> dict:
    result = research_graph.invoke(state)   # 子图！
    return {"messages": [AIMessage(content=f"研究: {result['analysis']}")]}

def writer_node(state: SupervisorState) -> dict:
    draft = llm.invoke(f"基于以下内容编写: {state['messages'][-1].content}").content
    return {"messages": [AIMessage(content=f"草稿: {draft}")]}

def reviewer_node(state: SupervisorState) -> dict:
    review = llm.invoke(f"审查: {state['messages'][-1].content}").content
    return {"messages": [AIMessage(content=f"审查: {review}")]}

# 构建主管图
g = StateGraph(SupervisorState)
for name, fn in [("supervisor",supervisor_node),("researcher",researcher_node),
                  ("writer",writer_node),("reviewer",reviewer_node)]:
    g.add_node(name, fn)

g.set_entry_point("supervisor")
g.add_conditional_edges("supervisor", route_to_agent)
for agent in ["researcher", "writer", "reviewer"]:
    g.add_edge(agent, "supervisor")    # 始终报告回来

multi_agent_app = g.compile()

模式 2 — 并行智能体 (Map-Reduce)

from langgraph.types import Send
import operator

class MapReduceState(TypedDict):
    documents:    list[str]
    summaries:    Annotated[list, operator.add]  # 并行写入合并至此
    final_report: str

async def summarize_doc(state: dict) -> dict:
    summary = await llm.ainvoke(f"总结: {state['document']}")
    return {"summaries": [summary.content]}

def reduce_node(state: MapReduceState) -> dict:
    combined = "\n\n".join(state["summaries"])
    report   = llm.invoke(f"统一报告:\n{combined}").content
    return {"final_report": report}

# fan_out 返回一个列表