LangGraph[1] ---- 初识 LangGraph:有状态的大模型应用编排

发布时间:2026/7/18 18:03:25
LangGraph[1] ---- 初识 LangGraph:有状态的大模型应用编排 文章目录前言1. 为什么需要 LangGraph2. 核心概念一状态State3. 核心概念二图与节点4. 核心概念三边5. 最小可运行示例6. 小结与预告前言大语言模型已经不只是“一问一答”的玩具了。我们越来越需要它完成多步骤的任务一边搜索资料一边思考、调用工具获取实时数据、在输出前经过人工审核……这些场景中应用不再是单根直线而是一张有分支、有循环、有记忆的网。LangChain 的链式调用Chain把多个步骤串成一条流水线但它难以优雅地处理循环、条件跳转和状态持久化。为此LangChain 团队推出了LangGraph—— 一个基于有向图和全局状态的大模型应用编排框架。今天这篇博客我们就从最核心的三个概念出发带你写出第一个 LangGraph 程序让图“跑”起来。1. 为什么需要 LangGraph假设我们要做一个智能客服用户提问后模型先判断意图。如果是“咨询”需要去知识库搜索再回答。如果是“投诉”需要生成回复草稿必须经人工审核才能发送。对话结束前模型可能需要反复确认信息。如果用传统链式调用你可能会写一堆if-else在链的中间插入人工交互也非常困难。LangGraph 的设计正是为了这些场景把应用的流程定义成一张图节点是具体的处理逻辑边是流转的规则而所有信息都存放在一个贯穿始终的“状态”对象里。2. 核心概念一状态State状态是整个图执行期间共享的数据结构。LangGraph 图在每一步都会接收当前状态节点处理完后返回一个部分状态更新框架再把这些更新合并回全局状态。通常用 Python 的TypedDict定义状态并为字段指定一个合并策略Reducer。最常用的合并策略是给消息列表用的add_messages。fromtypingimportTypedDict,Annotatedfromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportHumanMessage,AIMessageclassState(TypedDict):messages:Annotated[list,add_messages]# 新消息会追加到现有列表末尾counter:int# 普通字段直接覆盖Annotated[list, add_messages]的意思是当节点返回{messages: [new_msg]}时不直接替换整个列表而是把new_msg追加到原来的消息列表中。这非常适合记录聊天历史我们后面会看到它的效果。counter没有用Annotated则默认是直接覆盖节点返回{counter: 2}全局状态里的counter就会直接变成 2。3. 核心概念二图与节点图是一个StateGraph对象节点就是图中的执行单元——其实就是一个普通的 Python 函数。节点函数必须遵守一个约定接收一个参数当前全局状态字典形式。返回一个字典包含想要更新的字段和值。如果某些字段没变可以不写。下面定义一个最简单的节点它模拟一个“智能体”每被调用一次就让counter加 1并生成一条 AI 消息。defagent_node(state:State):countstate.get(counter,0)new_countcount1# 假装这是 LLM 生成的回复replyf收到这是第{new_count}次回复。return{messages:[AIMessage(contentreply)],counter:new_count}然后我们创建一个StateGraph把节点添加进去并告诉图“从哪里开始”。fromlanggraph.graphimportStateGraph,END graphStateGraph(State)# 用我们定义的 State 创建图graph.add_node(agent,agent_node)# 添加名为 agent 的节点graph.set_entry_point(agent)# 设置入口节点现在图上只有一个孤零零的节点还没办法跑——我们需要用边把它连起来。4. 核心概念三边边决定了节点执行完后下一步去哪里。LangGraph 中有两种边普通边无条件地从一个节点走到另一个节点或直接到END。条件边根据节点返回的结果动态选择下一个节点。在我们这个简单例子里希望“agent”节点被执行最多 3 次然后结束。所以要定义一个路由函数根据当前counter的值决定是继续循环回agent节点还是走向结束。defshould_continue(state:State):ifstate[counter]3:returnend# 结束的信号else:returncontinue# 继续循环的信号# 给 agent 节点添加条件边graph.add_conditional_edges(agent,# 从哪个节点出发should_continue,# 路由函数{continue:agent,# 如果路由返回 continue就回到 agent 节点end:END# 如果路由返回 end整个图执行结束})至此我们的图形成了一个循环agent - agent - agent ... - END。只要counter小于 3就会不断回到agent节点。5. 最小可运行示例把所有零件组装起来编译并执行。appgraph.compile()# 准备初始状态一条用户消息和计数器初值 0initial_state{messages:[HumanMessage(content你好请开始工作)],counter:0}# 执行图final_stateapp.invoke(initial_state)# 打印最终的消息列表formsginfinal_state[messages]:print(f[{msg.type.upper()}]{msg.content})输出将会是[HUMAN] 你好请开始工作 [AI] 收到这是第 1 次回复。 [AI] 收到这是第 2 次回复。 [AI] 收到这是第 3 次回复。可以看到每经过一次agent节点消息列表就会自动追加一条 AI 消息而且counter也在同步递增。这就是add_messages和状态合并机制在起作用——节点只需要关心自己产出的新消息旧的历史会被框架自动保留。如果你只放一个节点并用普通边直连END那就得到了一个最朴素的“链”如果再加上条件边和循环就得到了一个具备自主决策能力的Agent。LangGraph 的核心就是这么简单。6. 小结与预告今天我们一起抓住了 LangGraph 的三个核心要素状态State全局共享的数据字典通过 Reducer 控制字段的合并方式。节点Node接受状态、返回部分更新的 Python 函数。边Edge用普通边定义固定流程用条件边实现智能路由和循环。用这“三板斧”你已经能搭建出带记忆、能循环、会分岔的智能流程了。下一篇博客我们会深入持久化与人工介入如何让图在任意步骤暂停、保存、恢复甚至让人在半路改一改状态再继续执行。届时你的 Agent 就真正有了“记忆”和“可干预”的能力。