AI创想

标题: LangGraph 实战：从基础聊天机器人到智能交互系统的进阶之路 [打印本页]
作者: 米落枫    时间: 7 小时前
标题: LangGraph 实战：从基础聊天机器人到智能交互系统的进阶之路
作者：佑瞻
在开发智能聊天机器人时，我们常常面临这样的挑战：如何让机器人不仅能回答预设范围内的问题，还能动态调用外部工具获取最新信息，甚至记住多轮对话的上下文？今天我们就来聊聊 LangGraph 这个强大的工具，如何帮助我们构建从基础到智能的交互系统，一步步实现更自然、更智能的对话体验。
一、构建第一个基础聊天机器人

1. 准备工作与核心概念

开发前我们需要明确几个关键概念：LangGraph 基于状态图（StateGraph）构建应用，将整个交互流程视为一个 "状态机"，通过节点（Nodes）定义功能单元，用边（Edges）描述状态转换逻辑。这种设计让我们可以清晰地管理对话流程和数据流转。
首先安装必要的依赖：
python
  1. pip install -U langgraph langsmith
复制代码
2. 创建状态图框架

我们从定义状态开始，每个 StateGraph 都需要明确状态结构。以下是一个基础聊天机器人的状态定义：
python
  1. from typing import Annotated
  2. from typing_extensions import TypedDict
  3. from langgraph.graph import StateGraph, START
  4. from langgraph.graph.message import add_messages
  6. class State(TypedDict):
  7.     # 使用add_messages确保消息列表是追加而非覆盖
  8.     messages: Annotated[list, add_messages]
  10. graph_builder = StateGraph(State)
复制代码
3. 添加聊天节点与 LLM 集成

接下来集成 LLM 模型，这里以 OpenAI 为例：
python
  1. import os
  2. from langchain.chat_models import init_chat_model
  4. os.environ["OPENAI_API_KEY"] = "sk-..."  # 填入你的API密钥
  5. llm = init_chat_model("openai:gpt-4.1")
  7. # 定义聊天节点函数
  8. def chatbot(state: State):
  9.     return {"messages": [llm.invoke(state["messages"])]}
  11. # 将节点添加到状态图
  12. graph_builder.add_node("chatbot", chatbot)
复制代码
4. 定义流程并编译

设置入口点并编译状态图：
python
  1. # 添加从起点到聊天节点的边
  2. graph_builder.add_edge(START, "chatbot")
  3. # 编译得到可执行的图
  4. graph = graph_builder.compile()
复制代码
5. 运行与交互

最后实现一个简单的交互循环：
python
  1. from IPython.display import Image, display
  3. def stream_graph_updates(user_input: str):
  4.     for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
  5.         for value in event.values():
  6.             print("Assistant:", value["messages"][-1].content)
  8. # 交互循环
  9. while True:
  10.     try:
  11.         user_input = input("User: ")
  12.         if user_input.lower() in ["quit", "exit", "q"]:
  13.             print("Goodbye!")
  14.             break
  15.         stream_graph_updates(user_input)
  16.     except:
  17.         # 异常处理，默认提问
  18.         user_input = "What do you know about LangGraph?"
  19.         print("User: " + user_input)
  20.         stream_graph_updates(user_input)
  21.         break
复制代码
运行后，当用户提问 "What do you know about LangGraph?"，机器人会回答："LangGraph is a library designed to help build stateful multi-agent applications using language models..."
二、为机器人添加工具调用能力

基础聊天机器人只能回答 LLM 训练数据内的问题，为了处理更广泛的查询，我们需要集成工具调用能力。这里以 Tavily 搜索工具为例：
1. 安装与配置搜索工具

python
  1. pip install -U langchain-tavily
复制代码
配置 API 密钥：
python
  1. import os
  2. os.environ["TAVILY_API_KEY"] = "你的Tavily API密钥"
复制代码
2. 定义与测试工具

python
  1. from langchain_tavily import TavilySearch
  3. # 初始化搜索工具，最多返回2条结果
  4. tool = TavilySearch(max_results=2)
  5. tools = [tool]
  7. # 测试工具调用
  8. tool.invoke("What's a 'node' in LangGraph?")
复制代码
工具会返回包含标题、URL 和内容摘要的搜索结果，这些信息可以帮助机器人回答更专业的问题。
3. 升级状态图以支持工具调用

修改 LLM 以绑定工具，并添加工具节点：
python
  1. # 绑定工具到LLM
  2. llm_with_tools = llm.bind_tools(tools)
  4. # 定义工具执行节点
  5. class BasicToolNode:
  6.     def __init__(self, tools: list) -> None:
  7.         self.tools_by_name = {tool.name: tool for tool in tools}
  8.    
  9.     def __call__(self, inputs: dict):
  10.         if messages := inputs.get("messages", []):
  11.             message = messages[-1]
  12.         else:
  13.             raise ValueError("No message found in input")
  14.         
  15.         outputs = []
  16.         for tool_call in message.tool_calls:
  17.             tool_result = self.tools_by_name[tool_call["name"]].invoke(tool_call["args"])
  18.             outputs.append(
  19.                 ToolMessage(
  20.                     content=json.dumps(tool_result),
  21.                     name=tool_call["name"],
  22.                     tool_call_id=tool_call["id"],
  23.                 )
  24.             )
  25.         return {"messages": outputs}
  27. tool_node = BasicToolNode(tools=[tool])
  28. graph_builder.add_node("tools", tool_node)
复制代码
4. 添加条件路由

定义路由函数判断是否需要调用工具：
python
  1. from langgraph.graph import END
  3. def route_tools(state: State):
  4.     """根据最新消息判断是否需要调用工具"""
  5.     if isinstance(state, list):
  6.         ai_message = state[-1]
  7.     elif messages := state.get("messages", []):
  8.         ai_message = messages[-1]
  9.     else:
  10.         raise ValueError(f"No messages found in input state to tool_edge: {state}")
  11.    
  12.     if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
  13.         return "tools"
  14.     return END
  16. # 添加条件边
  17. graph_builder.add_conditional_edges(
  18.     "chatbot",
  19.     route_tools,
  20.     {"tools": "tools", END: END},
  21. )
  22. graph_builder.add_edge("tools", "chatbot")  # 工具调用后返回聊天节点
复制代码
5. 使用预构建组件简化开发

LangGraph 提供了预构建组件来简化开发：
python
  1. from langgraph.prebuilt import ToolNode, tools_condition
  3. # 替换为预构建工具节点
  4. tool_node = ToolNode(tools=[tool])
  5. graph_builder.add_node("tools", tool_node)
  7. # 使用预构建条件判断
  8. graph_builder.add_conditional_edges(
  9.     "chatbot",
  10.     tools_condition,
  11. )
复制代码
现在当用户提问 "最新的人工智能发展趋势" 时，机器人会先调用搜索工具，再根据结果生成回答：
plaintext
  1. Assistant: [{'text': "To provide you with accurate and up-to-date information, I'll need to search for the latest trends. Let me do that for you.", 'type': 'text'}, {'id': 'toolu_01Q588CszHaSvvP2MxRq9zRD', 'input': {'query': '最新的人工智能发展趋势'}, 'name': 'tavily_search_results_json', 'type': 'tool_use'}]
  2. Assistant: [{"url": "https://www.example.com/ai-trends-2024", "content": "2024年人工智能的主要趋势包括多模态模型的普及、边缘计算的应用..."}, {"url": "https://arxiv.org/abs/2401.00123", "content": "最新研究表明，神经符号系统的结合将成为未来发展方向..."}]
  3. Assistant: 根据搜索结果，2024年人工智能的主要发展趋势包括：1. 多模态模型的广泛应用...
复制代码
三、为机器人添加对话记忆

当前机器人还无法记住多轮对话的上下文，我们可以通过检查点机制来实现记忆功能：
1. 实现记忆功能

python
  1. from langgraph.checkpoint.memory import MemorySaver
  3. # 创建内存检查点
  4. memory = MemorySaver()
  6. # 编译时传入检查点
  7. graph = graph_builder.compile(checkpointer=memory)
复制代码
2. 多轮对话测试

python
  1. # 第一轮对话
  2. config = {"configurable": {"thread_id": "1"}}
  3. user_input = "Hi there! My name is Will."
  4. events = graph.stream(
  5.     {"messages": [{"role": "user", "content": user_input}]},
  6.     config,
  7.     stream_mode="values",
  8. )
  9. # 机器人回应：Hello Will! It's nice to meet you. How can I assist you today?
  11. # 第二轮对话
  12. user_input = "Remember my name?"
  13. events = graph.stream(
  14.     {"messages": [{"role": "user", "content": user_input}]},
  15.     config,
  16.     stream_mode="values",
  17. )
  18. # 机器人回应：Of course, I remember your name, Will.
复制代码
3. 不同上下文隔离

通过不同的 thread_id 区分不同对话上下文：
python
  1. # 新的对话上下文
  2. events = graph.stream(
  3.     {"messages": [{"role": "user", "content": "Remember my name?"}]},
  4.     {"configurable": {"thread_id": "2"}},
  5.     stream_mode="values",
  6. )
  7. # 机器人回应：I apologize, but I don't have any previous context...
复制代码
4. 检查状态快照

可以随时获取当前状态：
python
  1. snapshot = graph.get_state(config)
  2. print(snapshot.values)
  3. # 输出包含完整的消息历史和配置信息
复制代码
四、总结与进阶方向

通过 LangGraph，我们实现了从基础聊天机器人到具备工具调用和记忆能力的智能交互系统。整个过程中，StateGraph 作为状态机的核心思想贯穿始终，通过节点和边的组合，我们可以灵活定义复杂的交互逻辑。
关键技术点回顾：

如果本文对你有帮助，别忘了点赞收藏，关注我，一起探索更高效的开发方式～

原文地址：https://blog.csdn.net/The_Thieves/article/details/148789558



欢迎光临 AI创想 (https://www.llms-ai.com/) Powered by Discuz! X3.4