LangGraph 函数字典从入门到入魂一网打尽所有常用 API引言本文是 LangGraph 的“新华字典”收录了所有你可能会用到的函数、类和方法。每个函数都配有 代码示例 和 真实使用场景让你不仅知道“怎么用”更知道“什么时候用”。风格CSDN 老司机带路幽默不废话干货拉满 写在前面LangGraph 的函数体系LangGraph 的函数可以分为几大家族图构建家族StateGraph、add_node、add_edge、add_conditional_edges状态管理家族Annotated、add_messages、自定义 Reducer执行家族invoke、stream、ainvoke、astream持久化家族MemorySaver、SqliteSaver、Checkpointer高级特性家族interrupt_before/after、get_state、update_state下面我们一个一个盘第一章图构建函数 —— 搭骨架的“钢筋水泥”StateGraph —— 一切的起点函数签名classStateGraph(StateType):def__init__(self,state_schema:Type[StateType])作用创建一个状态图对象是所有 LangGraph 应用的“地基”。参数state_schema是你定义的状态结构TypedDict 或 Pydantic Model。代码示例fromtypingimportTypedDictfromlanggraph.graphimportStateGraphclassMyState(TypedDict):counter:intmessages:listgraphStateGraph(MyState)# 开工使用场景每次你要构建一个新的 Agent 或工作流第一行代码就是它。调侃它就像你买了个毛坯房后面的add_node就是在砌墙、装门。add_node —— 给图加“动作演员”函数签名graph.add_node(name:str,action:Callable[[State],dict])作用向图中添加一个节点。节点就是一个普通的 Python 函数接收当前状态返回要更新的状态字段。参数name节点的名字后面跳转时要用action节点函数输入 State输出 dict部分状态更新代码示例defincrement_counter(state:MyState):return{counter:state[counter]1}graph.add_node(increment,increment_counter)使用场景每个“动作”都是一个节点。比如“调用 LLM”、“执行工具”、“人工审核”都可以拆成独立的节点。调侃节点就像公司里的员工各干各的活互不干扰。add_edge —— 连接节点规划“剧情走向”函数签名graph.add_edge(start:str,end:str)作用添加一条从 start 节点到 end 节点的固定边。执行完 start 后一定走 end。参数start源节点名end目标节点名可以是END代码示例fromlanggraph.graphimportEND graph.add_edge(increment,print_result)# 加完数就去打印graph.add_edge(print_result,END)# 打印完就结束使用场景当流程是确定的、没有分支的时候用固定边。比如“先 A 后 B 再结束”。调侃这就是你每天上班的固定路线家 → 地铁 → 公司 → 家。add_conditional_edges —— 给剧情加“选择题”函数签名graph.add_conditional_edges(source:str,path:Callable[[State],str],path_map:Optional[dict]None)作用根据当前状态动态决定下一个节点。参数source源节点path一个函数输入 State返回一个字符串节点名path_map可选把返回值映射到实际节点名代码示例defshould_continue(state:MyState):ifstate[counter]10:returnendelse:returnincrementgraph.add_conditional_edges(sourceincrement,pathshould_continue,path_map{end:END,increment:increment}# 可选)使用场景经典场景是 ReAct Agent 的判断——“如果 AI 想用工具就走工具节点否则结束”。调侃这就是你妈问你“考了多少分”——100 分奖励玩具60 分奖励巴掌。set_entry_point —— 指定“第一幕从哪开始”函数签名graph.set_entry_point(node_name:str)作用设置图的入口节点。参数node_name是你想最先执行的节点名。代码示例graph.set_entry_point(increment)# 从 increment 开始使用场景每个图必须有且只有一个入口点不然图不知道从哪开始跑。调侃这就像电影的开场画面没它观众一脸懵逼。**compile**—— 把“图纸”变成“可执行程序”函数签名appgraph.compile(checkpointer:Optional[Checkpointer]None,interrupt_before:Optional[list]None)作用编译图返回一个可运行的 Runnable 对象。编译后才能调用invoke或stream。参数checkpointer持久化检查点后面细讲interrupt_before/after在哪些节点前后中断人机协同代码示例使用场景图搭完后最后一步就是编译。不编译就像没装引擎的汽车。调侃编译后的app就是你的“最终产品”可以上线卖了。第二章状态管理函数 —— 让数据“活”起来**Annotated** 自定义 Reducer —— 控制状态如何合并函数签名fromtypingimportAnnotatedclassMyState(TypedDict):messages:Annotated[list,add_messages]# 内置 reducercounter:Annotated[int,lambdaleft,right:leftright]# 自定义作用给状态字段加一个“合并规则”。当多个节点返回同一个字段时决定是覆盖还是累加。关键点add_messages内置 reducer自动把新消息追加到列表末尾自定义 reducer你可以写任意合并逻辑代码示例fromlanggraph.graph.messageimportadd_messagesclassChatState(TypedDict):messages:Annotated[list,add_messages]# 追加消息total_tokens:Annotated[int,lambdax,y:xy]# 累加 token使用场景messages必须用追加否则会覆盖历史对话计数器、总 token 数等用累加用户配置用覆盖只保留最新调侃没有 Reducer 的状态就像没有规矩的会议室谁都能把白板擦掉。第三章执行函数 —— 让图“跑起来”**invoke**—— 同步执行拿最终结果函数签名resultapp.invoke(input:dict,config:Optional[dict]None)作用同步执行整个图返回最终状态。参数input初始状态字典config配置如{configurable: {thread_id: xxx}}代码示例resultapp.invoke({messages:[HumanMessage(content你好)]})print(result[messages][-1].content)使用场景当你不关心中间步骤只要最终答案时用。调侃这就像吃火锅你只管最后捞肉中间涮的过程不关心。**stream**—— 流式执行看中间步骤函数签名forchunkinapp.stream(input:dict,config:Optional[dict]None,stream_mode:strvalues):print(chunk)作用逐步执行每完成一个节点就 yield 一个结果。stream_mode 参数valuesyield 当前完整状态updates只 yield 本次节点返回的更新差异代码示例forstepinapp.stream({counter:0},stream_modeupdates):print(step)# 输出{increment: {counter: 1}} 等使用场景调试想看每个节点的输出UI 实时更新比如 Agent 思考过程逐条显示长任务提前给用户反馈调侃这就像直播带货每一步都给你看不像 invoke 是录播。**ainvoke**/**astream**—— 异步版本函数签名作用与同步版相同但支持async/await适合 FastAPI、异步 Web 应用。使用场景你的 Web 服务用了asyncio或者想在事件循环里调用。第四章持久化函数 —— 让 Agent 拥有“记忆”**MemorySaver**/**SqliteSaver**—— 保存状态函数签名fromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.checkpoint.sqliteimportSqliteSaver memoryMemorySaver()appgraph.compile(checkpointermemory)作用自动保存每个节点执行后的状态支持通过thread_id恢复会话。区别MemorySaver存内存程序重启就丢SqliteSaver存 SQLite 文件永久保存代码示例# 使用 SQLite 持久化withSqliteSaver.from_conn_string(checkpoints.db)assaver:appgraph.compile(checkpointersaver)config{configurable:{thread_id:user_123}}# 第一轮对话app.invoke({messages:[HumanMessage(content我叫小明)]},config)# 第二轮对话记住我叫小明app.invoke({messages:[HumanMessage(content我叫什么)]},config)# 输出你叫小明使用场景多轮对话机器人长任务如数据清洗中断后恢复调试回溯到某个节点状态调侃没有 checkpointer 的 Agent 是金鱼7 秒记忆。加上它就是大象。**interrupt_before**/**interrupt_after**—— 人机协同函数签名appgraph.compile(checkpointersaver,interrupt_before[tools],# 进入 tools 节点前暂停interrupt_after[send_email]# 执行完 send_email 后暂停)作用在指定节点前后自动中断等待人工干预后继续。恢复执行# 检查状态stateapp.get_state(config)# 人工确认后继续app.invoke(None,config)使用场景敏感操作审批发送邮件、转账、删除数据人工纠错AI 生成的代码需要人审核人机协作的客服系统调侃这就像给 AI 配了个“领导审批”环节防止它胡来。**get_state**/**update_state**—— 时间旅行函数签名# 获取当前状态stateapp.get_state(config)# 获取历史快照historylist(app.get_state_history(config))# 回滚到某个历史状态app.update_state(config,history[-2].values)作用get_state获取指定 thread 的当前状态get_state_history获取所有历史快照时间旅行update_state手动更新状态可用于修正错误使用场景调试回退到出错前修改输入重试人工修正AI 判断错了人手动改状态后继续审计追溯 Agent 的每一步决策调侃这是 LangGraph 的“后悔药”吃了能回到过去。第五章可视化与辅助函数 —— 让你“看到”图**get_graph**—— 画图函数签名# 生成 Mermaid 格式graph_representationapp.get_graph().draw_mermaid()# 保存为 PNG需要安装 pygraphvizwithopen(graph.png,wb)asf:f.write(app.get_graph().draw_mermaid_png())作用生成图的可视化表示支持 Mermaid 和 PNG。使用场景给老板/客户展示你的 Agent 架构调试检查节点连接是否正确文档自动生成流程图调侃这比手画流程图快多了还能随代码自动更新。**START**/**END**—— 特殊节点函数签名fromlanggraph.graphimportSTART,END graph.add_edge(START,first_node)graph.add_edge(last_node,END)作用START虚拟的入口点用来简化边的添加END虚拟的结束点表示图执行完毕代码示例# 这样写更简洁graph.add_edge(START,node_a)graph.add_edge(node_a,END)使用场景几乎所有图都会用到尤其是用add_edge替代set_entry_point时 。 速查表一图看懂所有函数函数作用一句话总结StateGraph创建图对象买个毛坯房add_node加节点砌墙add_edge加固定边铺直路add_conditional_edges加条件边铺岔路set_entry_point设入口定大门compile编译图装引擎invoke同步执行一次跑完stream流式执行一步步跑MemorySaver内存持久化临时记忆SqliteSaverSQLite持久化永久记忆interrupt_before/after中断点设红绿灯get_state获取状态看当前update_state更新状态改历史get_graph可视化画图纸 附赠官方资源导航API 参考文档https://langchain-ai.github.io/langgraph/reference/GitHub 示例库https://github.com/langchain-ai/langgraph/tree/main/examplesLangGraph Studiohttps://langchain-ai.github.io/langgraph/studio/可视化调试工具 写在最后这篇文章就像一本字典不需要一次性读完。当你写代码时忘了某个函数怎么用回来翻翻就行。使用建议第一次接触看“图构建家族”跑通第一个例子做多轮对话看“持久化家族”做人机协同看interrupt_before和update_state调试复杂 Agent看get_graph和get_state_history最后送大家一句话LangGraph 的 API 不难难的是把业务流程抽象成图。但只要画得出流程图就能用 LangGraph 写出来。