目录
LangGraph对象：图（节点、边、状态）的定义与使用全解析
在AI Agent开发中，LangGraph凭借其状态机驱动的流程控制能力，成为构建复杂工作流的核心工具。本文将通过深度解析图对象的三大核心要素（状态、节点、边），结合代码案例，带你掌握LangGraph的底层设计逻辑与实战技巧。
LangGraph图对象的哲学内核
LangGraph的图结构本质上是对现实业务流程的抽象建模：
状态（State）：存储工作流全生命周期的数据，如同快递单贯穿所有中转站
节点（Node）：执行具体操作的函数，对应流程中的决策点或动作点
边（Edge）：定义节点间的流转规则，包含固定路径与条件分支
关键特性：支持循环执行（如机器人避障导航）、条件分支（如审批流金额判断）、状态持久化（中途暂停/恢复）
核心组件定义与代码实现
1. 状态定义：结构化数据容器
代码块
Python
自动换行
复制代码
from typing import TypedDict, Annotated, Sequence
from langgraph.graph import MessagesState
classOrderProcessingState(TypedDict):
# 用户原始请求
user_request: str
# 库存查询结果
inventory_check: Annotated[dict, "库存系统返回数据"]
# 支付状态
payment_status: Annotated[bool, "支付是否成功"]
# 最终订单号
order_id: Annotated[str, "生成的唯一订单ID"]
# 继承MessagesState支持对话历史
messages: Annotated[Sequence, MessagesState]复制成功
最佳实践：使用TypedDict+Annotated明确字段语义，继承MessagesState可兼容对话类场景
2. 节点定义：执行单元设计
代码块
Python
自动换行
复制代码
from langgraph.graph import StateGraph
from langchain_openai import ChatOpenAI
# 初始化LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
definventory_check_node(state: OrderProcessingState) -> dict:
"""模拟库存查询节点"""
# 实际场景应调用真实库存API
mock_response = {"product_id": "P12345", "stock": 10}
return {"inventory_check": mock_response}
defpayment_process_node(state: OrderProcessingState) -> dict:
"""模拟支付处理节点"""
# 根据库存判断支付是否成功
if state["inventory_check"]["stock"] > 0:
return {"payment_status": True, "order_id": "O987654321"}
return {"payment_status": False}复制成功
节点设计原则：
输入参数始终是完整状态对象
返回值仅需包含需要更新的字段（自动合并到全局状态）
可混合使用LLM调用与业务逻辑
代码块
Python
自动换行
复制代码
# 创建状态图
workflow = StateGraph(OrderProcessingState)
# 添加节点
workflow.add_node("check_inventory", inventory_check_node)
workflow.add_node("process_payment", payment_process_node)
# 添加边
workflow.add_edge(START, "check_inventory")
workflow.add_edge("check_inventory", "process_payment")
# 添加条件边
workflow.add_conditional_edges(
"process_payment",
lambda state: "success"if state["payment_status"] else"failure",
{
"success": END,
"failure": "check_inventory"# 支付失败则重新查询库存
}
)
# 编译图
app = workflow.compile()复制成功
条件边高级用法：
使用lambda表达式动态判断分支
可实现多层嵌套条件（如审批流中的金额分级）
支持循环边（如机器人导航中的避障重试）
代码块
Python
自动换行
复制代码
[START]
↓
[库存检查] → [支付处理] → [成功结束]
↑__________ [支付失败后重新检查库存]复制成功
2. 代码执行示例
代码块
Python
自动换行
复制代码
# 初始状态
initial_state = {
"user_request": "购买商品P12345",
"messages": [{"role": "user", "content": "我要下单"}]
}
# 执行工作流
result = app.invoke(initial_state)
# 输出结果
print("最终订单状态:", result)
"""
{
'user_request': '购买商品P12345',
'inventory_check': {'product_id': 'P12345', 'stock': 10},
'payment_status': True,
'order_id': 'O987654321',
'messages': [...]
}
"""复制成功
进阶技巧与行业实践
1. 状态持久化实现
代码块
Python
自动换行
复制代码
from langgraph.checkpoint.memory import MemorySaver
# 创建带持久化的图实例
app = workflow.compile(checkpointer=MemorySaver())
# 恢复中断的流程
thread_config = {"configurable": {"thread_id": "user_123"}}
result = app.invoke(initial_state, thread_config)复制成功
2. 多智能体协作场景
代码块
Python
自动换行
复制代码
# 定义客服与质检两个角色
workflow.add_node("customer_service", cs_agent_node)
workflow.add_node("quality_check", qc_agent_node)
# 设置角色间对话边
workflow.add_edge("customer_service", "quality_check")
workflow.add_conditional_edges(
"quality_check",
lambda s: "escalate"if s["needs_human"] else"auto_resolve"
)复制成功
典型应用场景对比
从流程思维到智能体思维
LangGraph通过显式定义状态流转，将复杂的AI工作流转化为可组合、可调试的图结构。掌握其核心组件的定义与使用，是构建高可用AI Agent系统的关键。建议开发者从简单场景切入，逐步探索多智能体协作、动态流程优化等高级模式。
cv44533408
分享至
投诉或建议
