尧图网站建设 
传统LangChain的痛点流程不可见下一步走哪条路藏在循环 / 模型决策里画不出确定的流程图没法给人 review、没法单测某个分支不能暂停流程一旦开始只能跑到结束没法在 “建工单” 前停下来等主管审批状态不持久进程崩了 / 重启了跑到一半的状态全丢无法从断点续跑不能回退 / 审计看不到每一步的中间数据出错后无法退回到上一步换个走法重跑对于以上都是一个黑盒只有输入和输出中间的所有过程都不可见LangGraph是什么LangGraph 是一个 Python 库它让你把 Agent 流程声明成一张有向图图里的方框是处理步骤分类、知识 Agent、工单 Agent…箭头是步骤之间的走向框架负责按箭头驱动这些方框、统一保管中间数据每走完一个方框就存一次快照从而支持暂停、恢复、回退LangGraph中的状态和节点** State状态图运行期间所有节点共享的一份 “黑板数据”**用 TypedDict 声明有哪些字段。它就是你手写版TaskContext的框架托管版Node节点图上的一个方框本质是一个普通函数签名固定 —— 吃进当前 state返回一个 dict。它对应你手写版里的一个 Agent 类Agent.handle(ctx)函数内部可以调 LLM、调工具、查库LangGraph 不关心就是状态进入节点在节点里面经过一系列操作后出来状态就变了变状态进入节点后的变化节点只交改动不交全文supervisor 节点跑完只返回 {“logs”: […], “next”: “ticket”}而不是把整份 state 抄一遍交回去。为什么因为框架需要知道你到底改了啥然后你不能整这个state[“logs”].append(“处理完成”)不能直接操作state只能返回state然后框架自己操作合并state中的每一个字段得声明如何合并普通字段比如 next: str后来的覆盖前面的。主管写了 next“ticket”那 next 就是 “ticket”简单粗暴list operator.add拼接。节点1往 logs 里加了 [‘请求开始’]节点2加了 [‘主管判定 - 工单’]合起来就是三段完整日志。这就是为什么 logs 能拼成那三段list add_messages对话消息专用。不光能追加还能替换——如果两条消息 id 相同都是 m1合并时不是变成两条而是后来那条把前面的更新掉LangGraph中的边Edge边是连接方框的箭头决定 “一个节点执行完后下一步去哪”两种边固定边add_edge(a, b)a 执行完无条件去 b。START图入口、END图出口是两个特殊节点标图的首尾条件边add_conditional_edges(...)a 执行完后框架调用一个路由函数按它返回的字符串运行时决定去哪。这就是你手写版classify()agents.get(next)的框架化真实的APIg.add_conditional_edges(supervisor,# 从哪个节点出发route_by_field,# 路由函数吃 state、返回字符串传函数本身不加括号{knowledge:knowledge,ticket:ticket,chitchat:chitchat},)# path_map返回值 - 目标节点名也可映射到 ENDcompile/invoke/stream——画图纸 → 出成品 → 跑起来graphStateGraph(State)graph.add_node(llm,call_model)graph.add_edge(START,llm)graph.add_edge(llm,END)graph StateGraph(State)初始化图纸graph.add_node(“llm”, call_model)加节点前面是节点的名字后面是节点执行的函数compile把图纸变成能跑的东西appgraph.compile()compile() 之后才拿到一个可执行对象(app)。它会做这些事:校验图合不合法(有没有悬空节点、环写不合法之类)算出执行顺序、确定入口出口生成一个能真正跑的 runtimeinvoke(初始state) — 一口气跑到底final_stateapp.invoke({question:什么是LangGraph?})同步阻塞,从 START 跑到 END,中间过程你看不见,只拿到最终那份完整的 state。适合:流程已经调通了,只要结果stream(初始state) — 每跑完一个节点就吐一口forchunkinapp.stream({question:什么是LangGraph?}):print(chunk)输出是{llm: {answer: ...}} {tool: {observation: ...}} {llm: {answer: ...最终答案}}每执行完一个节点,yield 一次 {节点名: 该节点产出的片段}recursion_limit图框架的 maxHops为环是显式允许的框架必须有总闸recursion_limit递归步数上限默认 25经过一个节点算一步超限抛GraphRecursionError强制终止检查点CheckpointerCheckpointer检查点器是框架的状态持久化组件图每执行完一个节点它就自动把当前完整 state 存成一份快照checkpoint快照checkpoint)某一步执行完时 state 的完整存档附带 “下一步要执行哪个节点”Checkpointer决定快照存到哪。三种实现对照你熟的存储thread_id快照按会话隔离开了 checkpointer 之后,每次 app.invoke() 结束,框架会把当前状态(所有 messages、变量)存成一份快照。存到哪?总得有个 key:config{configurable:{thread_id:thread-1}}app.invoke(输入,config)框架拿到 thread-1,就去存储里找名字叫 thread-1 的上一份快照,加载进来 → 在这个基础上追加你这次的新输入 → 跑完再存回去有了快照的能力1内置记忆# 第一轮config1{configurable:{thread_id:thread-1}}app.invoke({messages:[{role:user,content:你好}]},config1)# → checkpointer 存快照: messages [user:你好, ai:你好!]# → 共 2 条# 第二轮app.invoke({messages:[{role:user,content:我刚才说啥?}]},config1)# → 框架自动: 读取 thread-1 的旧快照 → 拿到旧的 2 条 → 追加新的 2 条# → 存新快照: messages [user:你好, ai:你好!, user:我刚才说啥?, ai:你说的是你好]# → 共 4 条唯一要做的:两轮传同一个 thread_id。剩下全部框架搞定 —— 读旧状态、追加新消息、存新快照能力2读档重跑LangGraph 的图执行是一步一步走节点的,每走一步,checkpointer 就存一份快照假设你的图是:入口 → agent → 结束,你 invoke 了两次(两轮对话):快照1: 执行完入口节点后的状态 ← 有存档 快照2: 执行完agent节点后的状态 ← 有存档 快照3: 执行完结束节点后的状态 ← 有存档 第二次 invoke: 快照4: 执行完入口节点后的状态 快照5: 执行完agent节点后的状态 快照6: 执行完结束节点后的状态用途 1 — 审计:看看第 3 步的时候 messages 是什么状态,第 4 步又变成了什么,每一步的输入输出都能追溯用途2 —从旧快照分叉重跑# 假设快照2 是agent 执行完但还没到结束的状态old_confighistory[2].config# 拿到那个时刻的 config# 从那个时刻重新跑,可以传不同的输入app.invoke({messages:[{role:user,content:换个说法回答}]},# 不同的输入!old_config# 从旧存档出发)能力3人工审批先写一个需要人工审批的点fromlanggraph.typesimportinterrupt,Commanddefticket_node(state):# 抛出问题,图当场暂停answerinterrupt({question:工单Agent准备建工单,是否批准?})# answer 就是外面 resume 进来的值ifanswerapproved:ticket_idcreate_ticket(...)return{messages:[f工单{ticket_id}已创建]}else:return{messages:[工单已取消]}调用 invoke,图跑到 ticket 节点就停了config{configurable:{thread_id:thread-1}}app.invoke(input,config)# 图执行过程:# 入口 → agent → ticket → 调用 interrupt() → 停!暂停在这!## 你可以检查:app.get_state(config).next# → (ticket,) 说明停在 ticket 节点# interrupt 传出的信息 {question: 工单Agent准备建工单,是否批准?}主管审批回复执行# 主管点了批准app.invoke(Command(resumeapproved),config)# 框架做的事:# 1. 加载 thread-1 的最新快照(ticket 节点的中断状态)# 2. 从 ticket 节点重新开始执行# 3. interrupt() 返回 approved ← 就是你 resume 进去的值# 4. ticket_node 继续往下跑 → 建单成功 → CF20260923001LangGraph的全景串联compile(checkpointer...) 把图纸变成可执行 app ┌──────────────────────────────────────────────────────────────────────┐ │ │ START ─固定边─▶ supervisor ─条件边(路由函数 path_map)─▶ knowledge ─固定边─▶ END │ ticket └──(循环条件边指回前面节点) chitchat │ StateTypedDict 黑板框架托管的 TaskContext │ 字段 reducer普通覆盖 / Annotated[list, operator.add]追加 │ / Annotated[list, add_messages]消息追加同id更新 │ Checkpointer每个节点执行完自动存快照按 thread_id 隔离 │ → 断点续跑 / get_state_history 时间旅行 / interrupt 人工审批 │ └──────────────────────────────────────────────────────────────────────┘LangGraph 没发明新业务概念主管 / 分类 / 路由 / 防循环它把这些套路框架化并额外补齐了状态持久化、回放和人工节点显式图编排和模型自治编排路径能预先枚举、要审计 / 审批 / 失败可恢复工单、退款、审批、合规流程→显式图路径无法预先枚举、开放式探索研究助手、通用任务 Agent→模型自治实际生产常是混合显式主管图把流程主干钉住某个节点内部再跑 ReAct模型自治编排ReAct就是给模型一个问题模型自己判断调用什么工具灵活性高但是流程不可见还有可能死循环控制权是模型显示图排列LangGrapg模型的执行流程都是人事先规划好的框架只是照着图一步步执行灵活度低但是流程可见还能回溯、暂停防止死循环控制权是画的图 LangGraph概念
传统LangChain的痛点流程不可见下一步走哪条路藏在循环 / 模型决策里画不出确定的流程图没法给人 review、没法单测某个分支不能暂停流程一旦开始只能跑到结束没法在 “建工单” 前停下来等主管审批状态不持久进程崩了 / 重启了跑到一半的状态全丢无法从断点续跑不能回退 / 审计看不到每一步的中间数据出错后无法退回到上一步换个走法重跑对于以上都是一个黑盒只有输入和输出中间的所有过程都不可见LangGraph是什么LangGraph 是一个 Python 库它让你把 Agent 流程声明成一张有向图图里的方框是处理步骤分类、知识 Agent、工单 Agent…箭头是步骤之间的走向框架负责按箭头驱动这些方框、统一保管中间数据每走完一个方框就存一次快照从而支持暂停、恢复、回退LangGraph中的状态和节点** State状态图运行期间所有节点共享的一份 “黑板数据”**用 TypedDict 声明有哪些字段。它就是你手写版TaskContext的框架托管版Node节点图上的一个方框本质是一个普通函数签名固定 —— 吃进当前 state返回一个 dict。它对应你手写版里的一个 Agent 类Agent.handle(ctx)函数内部可以调 LLM、调工具、查库LangGraph 不关心就是状态进入节点在节点里面经过一系列操作后出来状态就变了变状态进入节点后的变化节点只交改动不交全文supervisor 节点跑完只返回 {“logs”: […], “next”: “ticket”}而不是把整份 state 抄一遍交回去。为什么因为框架需要知道你到底改了啥然后你不能整这个state[“logs”].append(“处理完成”)不能直接操作state只能返回state然后框架自己操作合并state中的每一个字段得声明如何合并普通字段比如 next: str后来的覆盖前面的。主管写了 next“ticket”那 next 就是 “ticket”简单粗暴list operator.add拼接。节点1往 logs 里加了 [‘请求开始’]节点2加了 [‘主管判定 - 工单’]合起来就是三段完整日志。这就是为什么 logs 能拼成那三段list add_messages对话消息专用。不光能追加还能替换——如果两条消息 id 相同都是 m1合并时不是变成两条而是后来那条把前面的更新掉LangGraph中的边Edge边是连接方框的箭头决定 “一个节点执行完后下一步去哪”两种边固定边add_edge(a, b)a 执行完无条件去 b。START图入口、END图出口是两个特殊节点标图的首尾条件边add_conditional_edges(...)a 执行完后框架调用一个路由函数按它返回的字符串运行时决定去哪。这就是你手写版classify()agents.get(next)的框架化真实的APIg.add_conditional_edges(supervisor,# 从哪个节点出发route_by_field,# 路由函数吃 state、返回字符串传函数本身不加括号{knowledge:knowledge,ticket:ticket,chitchat:chitchat},)# path_map返回值 - 目标节点名也可映射到 ENDcompile/invoke/stream——画图纸 → 出成品 → 跑起来graphStateGraph(State)graph.add_node(llm,call_model)graph.add_edge(START,llm)graph.add_edge(llm,END)graph StateGraph(State)初始化图纸graph.add_node(“llm”, call_model)加节点前面是节点的名字后面是节点执行的函数compile把图纸变成能跑的东西appgraph.compile()compile() 之后才拿到一个可执行对象(app)。它会做这些事:校验图合不合法(有没有悬空节点、环写不合法之类)算出执行顺序、确定入口出口生成一个能真正跑的 runtimeinvoke(初始state) — 一口气跑到底final_stateapp.invoke({question:什么是LangGraph?})同步阻塞,从 START 跑到 END,中间过程你看不见,只拿到最终那份完整的 state。适合:流程已经调通了,只要结果stream(初始state) — 每跑完一个节点就吐一口forchunkinapp.stream({question:什么是LangGraph?}):print(chunk)输出是{llm: {answer: ...}} {tool: {observation: ...}} {llm: {answer: ...最终答案}}每执行完一个节点,yield 一次 {节点名: 该节点产出的片段}recursion_limit图框架的 maxHops为环是显式允许的框架必须有总闸recursion_limit递归步数上限默认 25经过一个节点算一步超限抛GraphRecursionError强制终止检查点CheckpointerCheckpointer检查点器是框架的状态持久化组件图每执行完一个节点它就自动把当前完整 state 存成一份快照checkpoint快照checkpoint)某一步执行完时 state 的完整存档附带 “下一步要执行哪个节点”Checkpointer决定快照存到哪。三种实现对照你熟的存储thread_id快照按会话隔离开了 checkpointer 之后,每次 app.invoke() 结束,框架会把当前状态(所有 messages、变量)存成一份快照。存到哪?总得有个 key:config{configurable:{thread_id:thread-1}}app.invoke(输入,config)框架拿到 thread-1,就去存储里找名字叫 thread-1 的上一份快照,加载进来 → 在这个基础上追加你这次的新输入 → 跑完再存回去有了快照的能力1内置记忆# 第一轮config1{configurable:{thread_id:thread-1}}app.invoke({messages:[{role:user,content:你好}]},config1)# → checkpointer 存快照: messages [user:你好, ai:你好!]# → 共 2 条# 第二轮app.invoke({messages:[{role:user,content:我刚才说啥?}]},config1)# → 框架自动: 读取 thread-1 的旧快照 → 拿到旧的 2 条 → 追加新的 2 条# → 存新快照: messages [user:你好, ai:你好!, user:我刚才说啥?, ai:你说的是你好]# → 共 4 条唯一要做的:两轮传同一个 thread_id。剩下全部框架搞定 —— 读旧状态、追加新消息、存新快照能力2读档重跑LangGraph 的图执行是一步一步走节点的,每走一步,checkpointer 就存一份快照假设你的图是:入口 → agent → 结束,你 invoke 了两次(两轮对话):快照1: 执行完入口节点后的状态 ← 有存档 快照2: 执行完agent节点后的状态 ← 有存档 快照3: 执行完结束节点后的状态 ← 有存档 第二次 invoke: 快照4: 执行完入口节点后的状态 快照5: 执行完agent节点后的状态 快照6: 执行完结束节点后的状态用途 1 — 审计:看看第 3 步的时候 messages 是什么状态,第 4 步又变成了什么,每一步的输入输出都能追溯用途2 —从旧快照分叉重跑# 假设快照2 是agent 执行完但还没到结束的状态old_confighistory[2].config# 拿到那个时刻的 config# 从那个时刻重新跑,可以传不同的输入app.invoke({messages:[{role:user,content:换个说法回答}]},# 不同的输入!old_config# 从旧存档出发)能力3人工审批先写一个需要人工审批的点fromlanggraph.typesimportinterrupt,Commanddefticket_node(state):# 抛出问题,图当场暂停answerinterrupt({question:工单Agent准备建工单,是否批准?})# answer 就是外面 resume 进来的值ifanswerapproved:ticket_idcreate_ticket(...)return{messages:[f工单{ticket_id}已创建]}else:return{messages:[工单已取消]}调用 invoke,图跑到 ticket 节点就停了config{configurable:{thread_id:thread-1}}app.invoke(input,config)# 图执行过程:# 入口 → agent → ticket → 调用 interrupt() → 停!暂停在这!## 你可以检查:app.get_state(config).next# → (ticket,) 说明停在 ticket 节点# interrupt 传出的信息 {question: 工单Agent准备建工单,是否批准?}主管审批回复执行# 主管点了批准app.invoke(Command(resumeapproved),config)# 框架做的事:# 1. 加载 thread-1 的最新快照(ticket 节点的中断状态)# 2. 从 ticket 节点重新开始执行# 3. interrupt() 返回 approved ← 就是你 resume 进去的值# 4. ticket_node 继续往下跑 → 建单成功 → CF20260923001LangGraph的全景串联compile(checkpointer...) 把图纸变成可执行 app ┌──────────────────────────────────────────────────────────────────────┐ │ │ START ─固定边─▶ supervisor ─条件边(路由函数 path_map)─▶ knowledge ─固定边─▶ END │ ticket └──(循环条件边指回前面节点) chitchat │ StateTypedDict 黑板框架托管的 TaskContext │ 字段 reducer普通覆盖 / Annotated[list, operator.add]追加 │ / Annotated[list, add_messages]消息追加同id更新 │ Checkpointer每个节点执行完自动存快照按 thread_id 隔离 │ → 断点续跑 / get_state_history 时间旅行 / interrupt 人工审批 │ └──────────────────────────────────────────────────────────────────────┘LangGraph 没发明新业务概念主管 / 分类 / 路由 / 防循环它把这些套路框架化并额外补齐了状态持久化、回放和人工节点显式图编排和模型自治编排路径能预先枚举、要审计 / 审批 / 失败可恢复工单、退款、审批、合规流程→显式图路径无法预先枚举、开放式探索研究助手、通用任务 Agent→模型自治实际生产常是混合显式主管图把流程主干钉住某个节点内部再跑 ReAct模型自治编排ReAct就是给模型一个问题模型自己判断调用什么工具灵活性高但是流程不可见还有可能死循环控制权是模型显示图排列LangGrapg模型的执行流程都是人事先规划好的框架只是照着图一步步执行灵活度低但是流程可见还能回溯、暂停防止死循环控制权是画的图