AI智能体开发实战：用LangGraph构建多步骤任务编排系统

为什么选择LangGraph构建AI智能体

人工智能应用正从单轮问答向多步骤任务编排演进。传统的大模型调用模式受限于上下文窗口和单次推理能力，面对需要规划、工具调用、状态回溯的复杂场景时表现不佳。LangGraph作为LangChain生态的扩展库，专门解决AI智能体的多步骤编排问题，支持循环图、条件分支和持久化状态，适合构建生产级Agent系统。

核心优势在于其图结构设计。开发者用节点（Node）定义每个处理步骤，用边（Edge）控制流转逻辑，条件边（Conditional Edge）实现动态路由。这种设计比线性Pipeline灵活得多，能表达回退、重试、人工审批等真实业务流程。

环境准备与项目初始化

开发环境使用Python 3.11+，安装依赖：

pip install langgraph langchain-openai langchain-community

项目目录结构参考：

agent_project/

├── graph.py # 主图定义

├── nodes/

│ ├── planner.py # 规划节点

│ ├── executor.py # 执行节点

│ └── reviewer.py # 审核节点

└── state.py # 状态定义

定义智能体状态模型

LangGraph使用TypedDict定义状态，所有节点共享同一份状态对象：

from typing import TypedDict, Annotated, Sequence

from langgraph.graph.message import add_messages

import operator

class AgentState(TypedDict):

messages: Annotated[Sequence, add_messages]

plan: list[str]

current_step: int

retry_count: Annotated[int, operator.add]

max_retries: int

status: str # planning / executing / reviewing / done / failed

每个字段的Annotated注解控制合并策略。add_messages让消息列表自动追加而非覆盖，operator.add让retry_count累加。这比手动管理状态变量可靠得多。

构建核心节点函数

规划节点接收用户输入，生成任务执行计划：

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def planner_node(state: AgentState) -> dict:

last_msg = state["messages"][-1].content

prompt = f"将以下任务拆解为可执行步骤，每步一行：{last_msg}"

response = llm.invoke(prompt)

steps = [s.strip() for s in response.content.strip().split("\n") if s.strip()]

return {"plan": steps, "current_step": 0, "status": "executing"}

执行节点逐步执行计划中的任务，支持工具调用：

def executor_node(state: AgentState) -> dict:

idx = state["current_step"]

if idx >= len(state["plan"]):

return {"status": "reviewing"}

step = state["plan"][idx]

result = llm.invoke(f"执行步骤：{step}")

return {

"messages": [{"role": "assistant", "content": f"步骤{idx+1}完成: {result.content[:200]}"}],

"current_step": idx + 1

}

条件路由与循环控制

LangGraph的关键能力是条件边，根据状态决定下一步走向：

from langgraph.graph import StateGraph, END

def route_after_execute(state: AgentState) -> str:

if state["status"] == "reviewing":

return "reviewer"

if state["retry_count"] >= state["max_retries"]:

return END

return "executor" # 继续执行下一步

审核节点判断执行结果是否达标，不达标则回退重试：

def reviewer_node(state: AgentState) -> dict:

last_output = state["messages"][-1].content

check = llm.invoke(f"评估结果是否满足要求：{last_output}")

if "不满足" in check.content:

return {"retry_count": 1, "status": "planning"}

return {"status": "done"}

这样当审核不通过时，状态切换回planning，图自动路由到规划节点重新制定计划。

组装完整图并运行

将节点和边组合为可执行的StateGraph：

graph = StateGraph(AgentState)

graph.add_node("planner", planner_node)

graph.add_node("executor", executor_node)

graph.add_node("reviewer", reviewer_node)

graph.set_entry_point("planner")

graph.add_conditional_edges("executor", route_after_execute)

graph.add_edge("planner", "executor")

graph.add_conditional_edges("reviewer", lambda s: "planner" if s["status"] == "planning" else END)

app = graph.compile()

运行时传入初始状态：

result = app.invoke({

"messages": [{"role": "user", "content": "帮我调研Kubernetes HPA策略并整理成文档"}],

"plan": [],

"current_step": 0,

"retry_count": 0,

"max_retries": 2,

"status": "planning"

})

生产环境的关键注意事项

持久化方面，LangGraph支持SQLite和PostgreSQL后端，重启后不丢失状态。配置checkpointer：

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("./agent_state.db")

app = graph.compile(checkpointer=checkpointer)

工具调用安全边界要严格限制。为每个工具函数添加输入校验和超时控制，防止Agent执行越权操作。典型做法是把工具函数限定在沙箱容器中运行，网络访问仅放行白名单域名。

成本控制方面，生产级Agent应设置token上限和步数上限。在状态中增加budget字段，每次LLM调用前检查剩余预算，超限直接终止并返回中间结果。这在处理长链任务时尤其重要——一个失控的循环可能快速消耗大量token。

监控层面，在每个节点的入口和出口添加日志埋点，记录输入摘要、执行耗时、token消耗。配合LangSmith能可视化完整的图执行路径，快速定位性能瓶颈或逻辑死循环。