LangGraph 框架概念与架构

作者：修仙界码农进阶版it_czz
说明：黄色节点（API1/2/3）在同一超步（Superstep）并行执行，执行完毕后统一进入 Merge 节点。
6. 自定义 Reducer

控制状态更新逻辑。

1. from typing import Annotated
3. defmerge_dicts(current:dict, update:dict)->dict:"""深度合并字典"""
4.     result = current.copy()
5.     result.update(update)return result
7. defkeep_last_n(current:list, update:list, n:int=10)->list:"""保留最后 n 个元素"""
8.     combined = current + update
9.     return combined[-n:]classMyState(TypedDict):# 深度合并字典
10.     config: Annotated[dict, merge_dicts]# 保留最后10条消息
11.     messages: Annotated[list,lambda c, u: keep_last_n(c, u,10)]

复制代码

---

与其他框架对比

LangGraph vs LangChain

特性	LangChain	LangGraph
定位	链式调用框架	图式编排框架
循环支持	❌ 不支持	✅ 原生支持
条件分支	⚠️ 有限支持	✅ 灵活支持
状态管理	⚠️ 通过 Memory	✅ 内置状态管理
可控性	⚠️ 较低	✅ 高度可控
适用场景	简单链式任务	复杂 Agent 系统

LangGraph vs AutoGPT

特性	AutoGPT	LangGraph
自主性	✅ 高度自主	⚠️ 可配置
可控性	❌ 难以控制	✅ 完全可控
透明度	❌ 黑盒	✅ 白盒
成本控制	❌ 难以控制	✅ 易于控制
生产就绪	❌ 不适合	✅ 适合

LangGraph vs CrewAI

特性	CrewAI	LangGraph
抽象层次	高（角色/任务）	低（节点/边）
灵活性	⚠️ 受限于框架	✅ 高度灵活
学习曲线	简单	中等
自定义能力	⚠️ 有限	✅ 完全自定义

---

最佳实践

1. 状态设计

✅ 好的实践：

1. classMyState(TypedDict):# 使用明确的类型
2.     user_id:str
3.     messages: Annotated[List[Message], add]
4.     current_step:int# 使用 Optional 表示可能为空
5.     error: Optional[str]
6.     result: Optional[dict]

复制代码

❌ 不好的实践：

1. classMyState(TypedDict):# 类型不明确
2.     data: Any
3.    
4.     # 缺少 Optional
5.     error:str# 可能导致运行时错误

复制代码

2. 节点设计

✅ 好的实践：

1. defmy_node(state: MyState)->dict:"""
2.     清晰的文档字符串
3.     单一职责原则
4.     """try:
5.         result = process(state["input"])return{"output": result}except Exception as e:return{"error":str(e)}

复制代码

❌ 不好的实践：

1. defmy_node(state):# 缺少类型注解# 做太多事情# 没有错误处理# 修改外部状态
2.     global_var = state["input"]return{"output":...}

复制代码

3. 循环控制

✅ 好的实践：

1. classMyState(TypedDict):
2.     iteration:int
3.     max_iterations:int=10
4.     is_complete:booldefshould_continue(state: MyState)->str:if state["is_complete"]:return"end"if state["iteration"]>= state["max_iterations"]:return"end"return"continue"

复制代码

4. 错误处理

✅ 好的实践：

1. defmy_node(state: MyState)->dict:try:
2.         result = risky_operation(state["input"])return{"result": result,"error":None}except Exception as e:
3.         logger.error(f"Node failed: {e}")return{"error":str(e),"result":None}# 在条件边中处理错误defroute_after_node(state: MyState)->str:if state.get("error"):return"error_handler"return"next_node"

复制代码

5. 使用检查点

✅ 好的实践：

1. from langgraph.checkpoint import SqliteSaver
3. # 持久化检查点
4. checkpointer = SqliteSaver.from_conn_string("./checkpoints.db")
5. app = workflow.compile(checkpointer=checkpointer)# 使用有意义的 thread_id
6. config ={"configurable":{"thread_id":f"user-{user_id}-session-{session_id}"}}

复制代码

6. 可观测性

✅ 好的实践：

1. import logging
3. logger = logging.getLogger(__name__)defmy_node(state: MyState)->dict:
4.     logger.info(f"Node started with state: {state.keys()}")
5.    
6.     result = process(state["input"])
7.    
8.     logger.info(f"Node completed, output keys: {result.keys()}")return result

复制代码

7. 测试

✅ 好的实践：

1. import pytest
3. deftest_my_node():# 测试单个节点
4.     state ={"input":"test"}
5.     result = my_node(state)assert result["output"]== expected
7. deftest_graph_flow():# 测试完整流程
8.     app = build_graph()
9.     final_state = app.invoke({"input":"test"})assert final_state["result"]== expected

复制代码

---

总结

LangGraph 的核心价值

循环和分支：支持复杂的控制流状态管理：简化多步骤应用的状态维护可控性：提供细粒度的控制能力可观测性：完整的执行追踪持久化：支持暂停、恢复、回溯

适用场景

✅ 多步骤推理任务✅ 需要反馈循环的 Agent✅ 人机协作流程✅ 复杂的业务工作流✅ 需要状态持久化的长任务

原文地址：https://blog.csdn.net/it_czz/article/details/154738059