LangGraph 与 Agent 自主规划：从目标分解到执行闭环

发表于2026-02-13|更新于2026-04-29|技术教程

|浏览量:

Agent 不是工具的组合，而是规划能力的体现。当 LLM 从”回答问题”进化到”完成任务”，核心挑战不再是模型能力本身，而是如何让 Agent 具备自主规划、执行和反思的完整闭环。LangGraph 正是为此而生——它用状态机的思维方式，为复杂 Agent 系统提供了可编排、可观测、可调试的基础设施。

从链式调用到状态图

早期的 LLM 应用大多是线性的：输入 → 模型 → 输出。LangChain 的 Chain 概念将这种线性流程代码化，但面对需要多步推理、工具调用、条件分支的复杂任务时，线性结构显得力不从心。

ReAct 模式（Reasoning + Acting）是一个转折点。它让模型在”思考”和”行动”之间循环，直到完成任务。但 ReAct 的循环逻辑是内嵌在提示词中的，开发者很难精确控制循环的边界、状态的流转，以及异常时的回退策略。

LangGraph 的核心洞察是：将 Agent 的运行时建模为状态图（StateGraph）。每个节点是一个计算单元（可以是 LLM 调用、工具执行或人工介入），边代表状态转移，而图的状态（State）则是贯穿整个执行过程的共享上下文。

from langgraph.graph import START, StateGraph, END
from typing_extensions import TypedDict

class AgentState(TypedDict):
    messages: list
    next_step: str
    iteration_count: int

def planner(state: AgentState):
    # 规划节点：分解目标为子任务
    return {"next_step": "execute"}

def executor(state: AgentState):
    # 执行节点：调用工具或生成内容
    return {"iteration_count": state["iteration_count"] + 1}

def should_continue(state: AgentState):
    # 条件边：决定是否继续或结束
    if state["iteration_count"] > 5:
        return END
    return "planner"

graph = StateGraph(AgentState)
graph.add_node("planner", planner)
graph.add_node("executor", executor)
graph.add_edge(START, "planner")
graph.add_edge("planner", "executor")
graph.add_conditional_edges("executor", should_continue)

app = graph.compile()

这段代码展示了 LangGraph 的基本模式：状态定义、节点实现、边连接。关键在于 add_conditional_edges——它让状态转移逻辑变得显式且可编程，而不是隐藏在模型的生成内容中。

自主规划的三层架构

基于 LangGraph 的实践，可以将 Agent 的自主规划能力分解为三个层次：

graph TD
    A[用户目标输入] --> B[任务分解层 Planning]
    B --> C[执行与观察层 Execution & Observation]
    C --> D{验证结果?}
    D -->|满足标准| E[进入下一任务]
    D -->|不满足| F[反思与调整层 Reflection]
    F --> G[修订计划]
    G --> C
    E --> H{所有任务完成?}
    H -->|否| C
    H -->|是| I[输出最终结果]
    
    style B fill:#e1f5ff
    style C fill:#fff4e1
    style F fill:#ffe1e1

1. 任务分解层（Planning）

规划的本质是将模糊的目标转化为可执行的子任务序列。在 LangGraph 中，这通常通过一个专门的规划节点实现：

def planning_node(state: AgentState):
    prompt = """基于以下目标，制定执行计划：
    目标：{goal}
    
    要求：
    1. 将目标分解为 3-5 个可验证的子任务
    2. 每个子任务需明确完成标准
    3. 标注子任务间的依赖关系
    
    输出格式：JSON 数组，包含 name, description, success_criteria, dependencies"""
    
    response = llm.invoke(prompt.format(goal=state["goal"]))
    plan = parse_json(response)
    return {"plan": plan, "current_task_idx": 0}

规划的质量直接决定 Agent 的执行效率。实践中发现，让模型显式输出”成功标准”能显著提升后续验证节点的准确性。同时，记录子任务间的依赖关系，可以在执行阶段实现并行化优化。

2. 执行与观察层（Execution & Observation）

执行节点负责完成当前子任务。这可能是调用外部工具、生成代码、检索信息，或是向用户请求输入。执行完成后，观察节点（Observer）收集执行结果并评估进展。

def execute_task(state: AgentState):
    current_task = state["plan"][state["current_task_idx"]]
    
    if current_task["type"] == "search":
        result = search_tool.invoke(current_task["query"])
    elif current_task["type"] == "code":
        result = code_executor.invoke(current_task["code"])
    else:
        result = llm.invoke(current_task["prompt"])
    
    return {"execution_result": result, "task_status": "completed"}

def observe_and_evaluate(state: AgentState):
    result = state["execution_result"]
    criteria = state["plan"][state["current_task_idx"]]["success_criteria"]
    
    evaluation_prompt = f"""评估以下执行结果是否满足成功标准：
    
    执行结果：{result}
    成功标准：{criteria}
    
    判断是否满足标准，如果不满足，说明原因并建议修正方案。"""
    
    evaluation = llm.invoke(evaluation_prompt)
    return {"evaluation": evaluation, "needs_retry": "不满足" in evaluation}

观察节点的设计体现了”显式验证”的原则——不要假设执行一定成功，而是通过独立的评估步骤来确认。这种设计虽然增加了计算开销，但能显著降低错误累积的风险。

3. 反思与调整层（Reflection & Adaptation）

当执行偏离预期时，Agent 需要具备反思能力。LangGraph 的循环结构天然支持这种反思模式：

def reflection_router(state: AgentState):
    if state.get("needs_retry") and state["retry_count"] < 3:
        return "revise_plan"
    elif state["current_task_idx"] < len(state["plan"]) - 1:
        return "next_task"
    else:
        return END

def revise_plan(state: AgentState):
    reflection_prompt = f"""原计划执行遇到问题：
    
    当前计划：{state["plan"]}
    失败任务：{state["plan"][state["current_task_idx"]]}
    失败原因：{state["evaluation"]}
    
    请修订计划，可以：
    1. 将失败任务拆分为更细的子任务
    2. 调整任务顺序或依赖关系
    3. 添加前置验证步骤
    
    输出修订后的完整计划。"""
    
    revised_plan = llm.invoke(reflection_prompt)
    return {"plan": parse_json(revised_plan), "retry_count": state["retry_count"] + 1}

反思不是简单的重试，而是基于失败原因进行策略调整。这种机制让 Agent 具备了类似人类的问题解决能力：当一条路走不通时，不是盲目重复，而是分析原因并尝试新的方法。

多 Agent 协作的三种模式

复杂任务往往需要多个 Agent 协作完成。LangGraph 的状态图模型为_multi-agent_架构提供了清晰的实现路径。

graph LR
    subgraph 模式一：共享状态协作
    A1[输入] --> B1[Coder Agent]
    B1 --> C1[共享状态]
    C1 --> D1[Reviewer Agent]
    D1 --> C1
    C1 --> E1[输出]
    end
    
    subgraph 模式二：监督者路由
    A2[输入] --> B2[Supervisor]
    B2 -->|学术查询| C2[Research Agent]
    B2 -->|新闻查询| D2[News Agent]
    B2 -->|代码查询| E2[Code Agent]
    C2 --> F2[Synthesizer]
    D2 --> F2
    E2 --> F2
    F2 --> G2[输出]
    end
    
    subgraph 模式三：分层团队
    A3[用户问题] --> B3[顶层路由]
    B3 --> C3[技术支持团队]
    B3 --> D3[销售团队]
    B3 --> E3[投诉处理团队]
    C3 --> F3[子图内部协作]
    D3 --> G3[子图内部协作]
    E3 --> H3[子图内部协作]
    end

模式一：共享状态协作

所有 Agent 共享同一个状态对象，每个 Agent 的产出对其他 Agent 可见。这种模式适合需要紧密协作的场景，比如代码审查：Coder Agent 生成代码，Reviewer Agent 基于同一份代码提出修改建议。

def create_collaboration_graph():
    graph = StateGraph(SharedState)
    
    # 两个 Agent 共享同一个消息历史
    graph.add_node("coder", coder_agent)
    graph.add_node("reviewer", reviewer_agent)
    
    def router(state):
        if "APPROVED" in state["messages"][-1].content:
            return END
        return "coder" if state["last_speaker"] == "reviewer" else "reviewer"
    
    graph.add_conditional_edges("coder", router)
    graph.add_conditional_edges("reviewer", router)
    
    return graph.compile()

模式二：监督者路由

一个 Supervisor Agent 负责任务分发，多个 Worker Agent 各自处理独立的状态。这种模式适合任务可以明确拆分的场景，比如研究助手：Supervisor 将查询路由到学术搜索 Agent、新闻搜索 Agent 或代码搜索 Agent。

def supervisor_router(state: SupervisorState):
    task_type = classify_task(state["query"])
    routing_map = {
        "academic": "research_agent",
        "news": "news_agent", 
        "code": "code_agent"
    }
    return routing_map.get(task_type, "general_agent")

graph.add_conditional_edges("supervisor", supervisor_router)
graph.add_edge("research_agent", "synthesizer")
graph.add_edge("news_agent", "synthesizer")
graph.add_edge("code_agent", "synthesizer")

模式三：分层团队

多层嵌套的状态图，每个子图本身是一个完整的 Agent 团队。这种模式适合复杂的企业级应用，比如自动化客服：顶层路由将问题分发给技术支持团队、销售团队或投诉处理团队，每个团队内部有自己的协作流程。

# 子图：技术支持团队
tech_support_graph = create_tech_support_team()

# 主图：客服中心  
main_graph = StateGraph(CustomerState)
main_graph.add_node("routing", triage_agent)
main_graph.add_node("tech_team", tech_support_graph)  # 子图作为节点
main_graph.add_node("sales_team", sales_support_graph)

Human-in-the-loop：必要的边界

完全自主的 Agent 是危险的。LangGraph 通过 interrupt 机制在关键节点插入人工确认：

graph TD
    A[开始执行任务] --> B{风险检测}
    B -->|低风险| C[自动执行]
    B -->|高风险| D[中断等待人工]
    D --> E[展示操作详情]
    E --> F{人工决策}
    F -->|批准| G[继续执行]
    F -->|拒绝| H[取消操作]
    F -->|修改| I[调整参数后执行]
    G --> J[记录执行结果]
    H --> J
    I --> J
    C --> J
    J --> K[继续后续流程]
    
    style D fill:#fff3cd
    style F fill:#d4edda

from langgraph.types import interrupt

def sensitive_action_node(state: AgentState):
    # 中断执行，等待人工确认
    user_decision = interrupt({
        "action": "delete_file",
        "target": state["file_path"],
        "reason": state["deletion_reason"]
    })
    
    if user_decision["approved"]:
        execute_deletion(state["file_path"])
    else:
        return {"status": "cancelled", "reason": user_decision["feedback"]}

人工介入不是对 Agent 能力的不信任，而是对风险的合理管控。在实践中，以下场景建议保留人工节点：

破坏性操作：删除数据、修改配置、发送对外邮件
高成本操作：调用付费 API、启动长时间计算任务
合规敏感：访问隐私数据、生成对外发布的正式内容
异常路径：当 Agent 进入未预料到的分支时

持久化与断点恢复

LangGraph 内置了 Checkpoint 机制，让长时间运行的 Agent 具备容错能力：

from langgraph.checkpoint.sqlite import SqliteSaver

# 配置检查点持久化
checkpointer = SqliteSaver.from_conn_string(":memory:")
app = graph.compile(checkpointer=checkpointer)

# 运行时可随时中断和恢复
config = {"configurable": {"thread_id": "task-123"}}
for event in app.stream(initial_state, config):
    print(event)
    # 如果进程崩溃，可以从最后一个检查点恢复