LangGraph实战：从零构建一个具备状态记忆的智能对话机器人

1. 为什么我们需要一个“有记忆”的机器人你有没有遇到过这样的场景你问一个AI助手“我昨天提到的那个项目进展怎么样了”它却一脸茫然地回答“抱歉我不记得我们之前的对话了。”或者你和它聊了十分钟电影突然问它“那你觉得我刚才推荐的那部电影男主角的演技怎么样”它可能完全接不上话茬。这就是传统对话机器人最大的痛点它们没有真正的“记忆”。每一次对话对它们来说都是一次全新的开始。这种“健忘症”让深度、连贯的交流变得几乎不可能。我们人类聊天之所以能层层递进、引经据典全靠大脑里那个强大的“状态记忆”系统。在AI开发领域我们过去常用的是“链”Chain式结构。你可以把它想象成一条生产线原料用户输入从一端进去经过几个固定的工序比如理解问题 - 搜索知识 - 生成回答成品AI回复就从另一端出来了。这条生产线是直的不能回头也不能根据中间产品的情况临时改变工序。当对话稍微复杂一点需要根据上一轮的回答来决定下一轮做什么时这种“直男”思维就彻底不够用了。后来有了“代理执行器”AgentExecutor它聪明了一点像个有经验的工头能根据情况决定下一步调用哪个工具比如是去查数据库还是去计算。但它管理复杂对话流程的能力依然有限状态也就是对话的历史和上下文需要开发者自己小心翼翼地手递手传递很容易出错或丢失。所以当我们需要构建一个真正智能的、能进行多轮复杂对话的机器人时我们需要一个更强大的“大脑架构”。这就是LangGraph登场的时候。它不再是一条线或一个点而是一张图。在这张图里对话可以灵活地“流动”可以分支可以循环而所有的对话历史、用户信息、中间结果都被妥善地保存在一个叫“状态”的容器里跟随流程一路传递。今天我就带你从零开始用LangGraph亲手搭建一个属于你自己的、具备持久状态记忆的智能对话伙伴。2. 理解LangGraph的核心状态、节点与边在动手写代码之前我们必须把LangGraph最核心的三个概念吃透。这就像学武功要先扎马步基础不牢后面全是坑。我自己刚开始学的时候就是因为没彻底搞懂它们之间的关系调试代码调得焦头烂额。2.1 状态机器人的“记忆面包”在LangGraph里State是整个应用的基石和中央存储器。它就是一个Python字典或者一个Pydantic模型里面装着对话进行所需的一切信息。你可以把它想象成机器人的“记忆面包”或者一个随身携带的“小本本”。这个“小本本”里通常会记些什么呢对话历史用户和AI说过的每一句话这是实现上下文理解的关键。用户信息比如用户的ID、偏好设置让对话更个性化。中间结果比如从数据库检索到的资料、经过分析后的关键信息等。控制标志比如记录对话进行了几轮、当前处于什么流程阶段。在代码里我们可以用一个非常简单的字典来定义状态但更规范、更强大的方式是使用Pydantic模型。这能确保数据的类型安全自动获得代码提示减少低级错误。例如我们可以定义一个最简单的对话状态from typing import List, Annotated from typing_extensions import TypedDict from langgraph.graph import add_messages class State(TypedDict): # 这个字段专门用来存储对话消息历史 messages: Annotated[List[str], add_messages] # 可以添加其他字段比如用户ID user_id: str # 或者一个标志记录是否已经问候过用户 has_greeted: bool这里的add_messages是一个“归约器”它能自动帮我们把新的消息追加到已有的消息列表后面而不用我们手动去写state[‘messages’].append(new_message)非常方便。状态是随着图的执行而不断演化的每个节点都可以读取它、修改它。2.2 节点机器人的“技能模块”Node就是图中的一个个“站点”或“处理单元”。在LangGraph中一个节点本质上就是一个Python函数。这个函数接收当前的状态作为输入干一些活比如调用大模型、查询数据库、进行逻辑判断然后返回一个更新后的状态。每个节点都应该职责单一就像一个功能模块。例如一个节点专门负责分析用户意图。一个节点负责根据意图去检索知识库。一个节点负责调用天气查询工具。一个节点负责组织语言生成最终回复。这种设计让我们的代码变得非常清晰和模块化。以后想加一个新功能很简单再写一个节点函数把它“挂”到图里合适的位置就行了。下面是一个节点函数的例子它调用大模型来生成回复from langchain_openai import ChatOpenAI # 初始化大模型 llm ChatOpenAI(model“gpt-4”) def chatbot_node(state: State): 节点函数调用AI模型生成回复 # 1. 从状态中获取完整的对话历史 conversation_history state[“messages”] # 2. 将历史对话作为上下文调用大模型 ai_response llm.invoke(conversation_history) # 3. 将AI的回复也加入到状态的消息历史中 new_messages state[“messages”] [ai_response] # 4. 返回更新后的状态这里更新了messages字段 return {“messages”: new_messages}你看这个函数只做一件事根据聊天历史让AI接着说下去。它接收一个状态返回一个新的状态。这就是节点的全部秘密。2.3 边决定对话的“流向”Edge定义了节点之间的连接关系决定了对话的“流向”。这是LangGraph比传统链式结构强大的关键所在。边分为两种普通边无条件地从一个节点指向另一个节点。对话走到这个节点后必须沿着这条边去下一个节点。条件边根据当前状态里的某些条件动态决定下一步该去哪个节点。这实现了真正的“分支”逻辑。条件边是让对话“活”起来的神奇开关。比如用户问“今天天气怎么样”经过“意图识别”节点后状态里被标记为intent: “query_weather”。那么条件边就会检查这个标记把对话流导向“查询天气工具”节点而不是去“生成普通回复”节点。我们可以用一个路由函数来实现条件边def route_after_intent(state: State): 条件边路由函数根据识别出的意图决定下一步 # 从状态中取出上一节点意图识别节点设置的结果 current_intent state.get(“current_intent”) if current_intent “greeting”: # 如果是问候直接去生成回复节点 return “generate_response” elif current_intent “query_weather”: # 如果是查询天气需要先去调用天气工具 return “call_weather_tool” elif current_intent “need_clarification”: # 如果问题模糊需要进入澄清节点反问用户 return “clarify_question” else: # 默认情况去生成回复 return “generate_response”通过状态、节点、边的组合我们就能绘制出一张完整的对话“地图”。机器人会带着它的“记忆小本本”状态按照地图的指示在不同的技能点节点之间穿梭并根据路标条件边做出选择最终完成一次复杂的对话交互。3. 实战第一步搭建你的第一个对话图理论讲得差不多了现在我们打开代码编辑器开始真正搭建。我保证从这里开始每一步你都能跟着做出来。我们先从一个最简单的、但已具备核心记忆功能的对话机器人开始。3.1 环境准备与依赖安装首先确保你的Python环境是3.8或以上版本。我强烈建议使用虚拟环境这样项目之间的依赖不会打架。打开你的终端执行以下命令# 创建并激活一个虚拟环境以venv为例 python -m venv langgraph-env # 在Windows上激活 langgraph-env\Scripts\activate # 在Mac/Linux上激活 source langgraph-env/bin/activate # 安装核心库 pip install langgraph langchain-openai这里我们安装了两个核心包langgraph是今天的主角langchain-openai则提供了方便调用OpenAI系列模型如GPT-4的接口。如果你使用其他模型比如国内的通义千问、文心一言可能需要安装对应的LangChain集成包或直接用其官方SDK。安装完成后在项目目录下创建一个新的Python文件比如my_chatbot.py。3.2 构建一个基础的单轮对话图我们从最基础的开始构建一个能记住对话历史的简单问答图。这个图只有一个核心节点调用大模型生成回复。但关键在于我们传给这个节点的“状态”里包含了之前所有的对话消息。from typing import List, Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI # 1. 定义状态结构核心是messages用于存储所有对话 class ConversationState(TypedDict): messages: Annotated[List, add_messages] # 使用add_messages自动处理消息列表 # 2. 初始化大语言模型这里以OpenAI为例你需要替换为自己的API密钥和base_url llm ChatOpenAI( model“gpt-3.5-turbo”, api_key“your-api-key-here”, # 请替换成你的真实API Key base_url“https://api.openai.com/v1” # 如果你用第三方代理可能需要修改 ) # 3. 定义唯一的节点函数生成回复 def call_model(state: ConversationState): 节点将整个对话历史发给模型让它接着回复 # 直接调用模型传入当前全部消息历史 response llm.invoke(state[“messages”]) # 关键返回一个字典更新状态中的messages字段。 # add_messages归约器会自动将新的response追加到消息列表末尾。 return {“messages”: [response]} # 4. 开始构建图 builder StateGraph(ConversationState) # 5. 将我们的节点函数添加到图中并给它起个名字叫“assistant” builder.add_node(“assistant”, call_model) # 6. 设置图的入口和出口 builder.add_edge(START, “assistant”) # 从开始直接进入助理节点 builder.add_edge(“assistant”, END) # 从助理节点直接结束 # 7. 编译图得到一个可执行的对象 graph builder.compile() # 8. 来和我们的机器人聊聊天 # 初始化状态第一条消息是用户说的 initial_state {“messages”: [{“role”: “user”, “content”: “你好请介绍下你自己。”}]} final_state graph.invoke(initial_state) # 打印出最终的对话历史 for msg in final_state[“messages”]: print(f“{msg.role}: {msg.content}”)运行这段代码你会看到AI的自我介绍。但它的强大之处在于多轮对话。让我们修改一下调用部分模拟一个连续对话# 模拟多轮对话 current_state {“messages”: [{“role”: “user”, “content”: “我叫小明。”}]} print(“第一轮“) result1 graph.invoke(current_state) for msg in result1[“messages”][-1:]: # 只打印最新回复 print(f“AI: {msg.content}”) # 关键将第一轮的结果状态作为第二轮输入的基础 current_state result1 # 手动添加用户第二句话在实际应用中这里是新的用户输入 current_state[“messages”].append({“role”: “user”, “content”: “你还记得我叫什么名字吗”}) print(“\n第二轮“) result2 graph.invoke(current_state) for msg in result2[“messages”][-1:]: print(f“AI: {msg.content}”)你会发现在第二轮对话中AI能够回答“你叫小明”。因为它收到的状态current_state里包含了第一轮对话的全部历史用户说“我叫小明”和AI的回复。状态在这里完美地充当了记忆载体。3.3 可视化你的对话流程图LangGraph一个非常酷的功能是能自动生成流程图让你直观地看到自己设计的对话路径。在上面的代码后添加# 生成并保存流程图 try: # 获取图的Mermaid格式定义 mermaid_code graph.get_graph().draw_mermaid() print(“Mermaid代码已生成可复制到支持Mermaid的编辑器如GitHub Markdown, Typora中查看。”) # 也可以尝试生成图片需要本地安装Graphviz # png_data graph.get_graph().draw_mermaid_png() # with open(“my_chatbot_graph.png”, “wb”) as f: # f.write(png_data) # print(“流程图已保存为 my_chatbot_graph.png”) except Exception as e: print(f“生成流程图时出错可能缺少依赖: {e}”) print(“你可以通过 pip install pygraphviz 尝试安装依赖或者直接使用Mermaid代码在线查看。”)虽然生成图片可能需要一些额外的系统依赖但获取到的Mermaid代码可以很容易地在很多Markdown编辑器或在线工具里渲染成结构图。看到那张只有一个节点的简单图你会对“图”的概念有更具体的认识。接下来我们要让它复杂起来。4. 升级机器人让对话拥有“决策”能力只有一个节点的机器人只是个复读机。真正的智能体现在“选择”上。现在我们给机器人增加一个“大脑前哨”——意图识别节点并根据识别结果通过条件边动态决定对话的流向。4.1 设计一个多节点对话图我们的目标是构建一个能处理两种意图的机器人普通聊天直接调用大模型回复。查询特定信息比如问“今天的日期”则先调用一个获取日期的工具再将结果交给大模型组织成友好回复。首先定义更丰富的状态并创建工具函数from datetime import datetime from typing import Literal # 升级版状态定义 class SmartBotState(TypedDict): messages: Annotated[List, add_messages] current_intent: str # 新增记录当前识别出的意图 tool_result: str # 新增存储工具调用的结果 # 定义一个简单的“获取当前日期”的工具函数 def get_current_date(): 工具函数返回当前日期字符串 return datetime.now().strftime(“%Y年%m月%d日”) # 初始化模型 llm ChatOpenAI(model“gpt-3.5-turbo”, api_key“your-key”)4.2 创建多个节点并连接现在创建三个节点意图识别节点分析用户最新消息的意图。工具调用节点如果意图是查询日期就执行工具。回复生成节点综合对话历史和工具结果生成最终回复。# 节点1意图识别 def intent_classifier(state: SmartBotState): 节点判断用户意图 last_message state[“messages”][-1].content # 简单的关键词匹配在实际项目中可以用更复杂的NLU模型 if “日期” in last_message or “今天几号” in last_message or “date” in last_message.lower(): intent “get_date” else: intent “general_chat” # 将识别出的意图更新到状态中 return {“current_intent”: intent} # 节点2调用工具 def call_tool(state: SmartBotState): 节点执行工具调用 if state[“current_intent”] “get_date”: result get_current_date() else: result “未使用工具” # 将工具执行结果更新到状态中 return {“tool_result”: result} # 节点3生成回复 def response_generator(state: SmartBotState): 节点生成最终回复给用户 messages state[“messages”] tool_result state.get(“tool_result”, “”) # 根据是否有工具结果构造不同的提示词给模型 if tool_result and tool_result ! “未使用工具”: # 将工具结果作为一条系统消息插入到历史中帮助模型理解上下文 system_msg {“role”: “system”, “content”: f“用户想查询信息。你已获取到以下结果{tool_result}。请根据此信息友好地回答用户。”} full_history [system_msg] messages else: full_history messages response llm.invoke(full_history) return {“messages”: [response]}4.3 使用条件边实现动态路由这是最精彩的部分。我们将创建一张图让对话流在“识别意图”后能自动选择是直接去“生成回复”还是先去“调用工具”。from langgraph.graph import StateGraph, START, END # 创建图构建器 builder StateGraph(SmartBotState) # 添加三个节点 builder.add_node(“classify_intent”, intent_classifier) builder.add_node(“use_tool”, call_tool) builder.add_node(“generate_response”, response_generator) # 设置起始边从START到“意图识别”节点 builder.add_edge(START, “classify_intent”) # 关键定义条件边路由函数 def decide_next_step(state: SmartBotState) - str: 根据意图决定下一个节点是‘调用工具’还是‘生成回复’ if state[“current_intent”] “get_date”: return “use_tool” # 需要查日期先去工具节点 else: return “generate_response” # 普通聊天直接去生成回复 # 添加条件边从“意图识别”节点出发根据路由函数结果决定去向 builder.add_conditional_edges( “classify_intent”, # 源节点 decide_next_step, # 路由函数 # 指定路由函数可能返回的所有目标节点供框架验证 {“use_tool”: “use_tool”, “generate_response”: “generate_response”} ) # 设置固定边从“工具调用”节点出来后必须去“生成回复”节点 builder.add_edge(“use_tool”, “generate_response”) # 设置固定边从“生成回复”节点出来后结束本次对话轮次 builder.add_edge(“generate_response”, END) # 编译图 smart_graph builder.compile()4.4 测试你的智能机器人现在让我们来测试这个拥有决策能力的机器人print(“ 测试场景1普通聊天 “) state1 {“messages”: [{“role”: “user”, “content”: “你好今天天气不错。”}]} result1 smart_graph.invoke(state1) print(f“用户: {state1[‘messages’][0][‘content’]}”) print(f“AI: {result1[‘messages’][-1].content}”) print(“\n 测试场景2查询日期 “) state2 {“messages”: [{“role”: “user”, “content”: “今天是几号”}]} result2 smart_graph.invoke(state2) print(f“用户: {state2[‘messages’][0][‘content’]}”) print(f“AI: {result2[‘messages’][-1].content}”) # 查看状态里记录的工具结果 print(f“内部工具调用结果{result2.get(‘tool_result’)}”)运行代码你会看到当你说“天气不错”时机器人直接进入闲聊模式。当你问“今天几号”时机器人会先调用get_current_date工具获取日期然后将这个日期信息作为上下文再让大模型生成一个像“今天是2023年10月27日”这样的自然回复。这就是状态记忆和条件路由的威力机器人不仅能记住对话历史还能根据对话内容状态中的意图动态改变自己的行为路径。你可以通过添加更多的意图分类和工具节点轻松地将这个机器人扩展成一个功能丰富的个人助理。5. 高级技巧与避坑指南通过前面的实战你已经掌握了LangGraph的核心。但在实际开发更复杂的应用时还会遇到一些典型问题。下面分享几个我踩过坑后总结的高级技巧和注意事项。5.1 管理多轮对话与状态持久化在实际应用中比如一个Web聊天机器人对话是跨HTTP请求持续的。你不能每次用户发消息都从头开始invoke一个全新的图。你需要持久化状态。LangGraph的状态本质上是字典你可以轻松地将其序列化如转成JSON存储到数据库、Redis或文件里。下次用户发起请求时再反序列化加载回来继续执行。import json # 假设一次对话结束我们得到最终状态 final_state # 1. 序列化状态注意消息对象可能需要特殊处理这里用其dict形式 state_to_save { “messages”: [msg.dict() for msg in final_state[“messages”]], “current_intent”: final_state[“current_intent”], “tool_result”: final_state[“tool_result”] } # 保存到数据库或会话中 # db.save(user_id, json.dumps(state_to_save)) # 2. 下次对话时从存储中加载 # loaded_data json.loads(db.load(user_id)) # 重建状态注意消息对象可能需要从dict还原 # restored_state {“messages”: [HumanMessage/AIMessage(**msg) for msg in loaded_data[“messages”]], …} # new_input {“messages”: restored_state[“messages”] [new_user_message]} # new_result graph.invoke(new_input)关键点add_messages归约器处理的消息对象如HumanMessage,AIMessage有特定的结构。直接序列化整个状态字典可能会遇到问题。更稳健的做法是只序列化必要的业务数据在恢复时重新构造消息列表。5.2 处理循环与中断避免对话“鬼打墙”在图结构中很容易不小心设计出循环导致对话在一个圈里无限运行。比如“识别意图 - 需要澄清 - 反问用户 - 等待输入 - 识别意图…”。LangGraph提供了两种机制来控制循环设置最大循环次数在编译图时通过checkpointer参数可以间接控制但更直接的方法是在你的条件边逻辑里加入计数器。使用中断机制在图定义中可以设置一个特定的节点或条件当满足时例如用户说“结束对话”直接路由到END节点。一个简单的循环计数示例class StateWithCounter(TypedDict): messages: Annotated[List, add_messages] loop_count: int # 新增循环计数器 def node_that_may_loop(state: StateWithCounter): # 业务逻辑... new_loop_count state.get(“loop_count”, 0) 1 return {“loop_count”: new_loop_count} def conditional_edge_with_limit(state: StateWithCounter): if state.get(“loop_count”, 0) 5: # 限制最多循环5次 return “force_end_node” elif some_condition: return “next_node” else: return “node_that_may_loop”5.3 调试技巧如何看清对话的每一步当你的图变得复杂调试就成了挑战。graph.stream()方法是你最好的朋友。它不会一次性返回最终结果而是返回一个生成器每次产出经过一个节点后的中间状态。# 使用stream来跟踪执行过程 test_input {“messages”: [{“role”: “user”, “content”: “今天日期是多少”}]} print(“开始流式执行追踪“) for step, output in enumerate(smart_graph.stream(test_input)): node_name list(output.keys())[0] # 输出字典的key是节点名 node_state output[node_name] # value是该节点执行后的状态 print(f”— 步骤 {step1}: 经过节点 ‘{node_name}’ —“) print(f“ 当前意图: {node_state.get(‘current_intent’, ‘N/A’)}“) print(f“ 工具结果: {node_state.get(‘tool_result’, ‘N/A’)}“) if node_state[“messages”]: last_msg node_state[“messages”][-1] if hasattr(last_msg, ‘content’): print(f“ 最新消息: {last_msg.content[:50]}…“) # 打印前50个字符 print()通过stream你可以像看日志一样清晰地看到状态是如何随着对话流经每个节点而一步步变化的。这对于定位是哪个节点的逻辑出了问题或者条件边的判断是否准确有巨大的帮助。构建基于LangGraph的对话机器人就像在绘制一张智能对话的“地图”。这张地图让AI的思考过程变得可见、可控、可扩展。从今天起忘掉那条笔直的“生产线”开始用“图”的思维来设计你的对话系统吧。当你看到自己设计的机器人能根据上下文自如地切换话题、调用工具、进行多轮深度交流时那种成就感绝对是单纯的API调用无法比拟的。