Python 基于 LangGraph 搭建带记忆与人工干预的搜索机器人
前言
本文介绍如何使用 Python 和 LangGraph 框架构建具备记忆功能、支持联网搜索及人工干预机制的智能聊天机器人。通过集成 Tavily 搜索工具和自定义中断逻辑，实现复杂任务处理与专业领域的人工接管。
前置准备
环境配置
- Python 3.9+
- 安装依赖：
pip install langgraph langchain-community langchain-openai tavily-python python-dotenv langsmith
密钥获取
- Tavily: 注册账号后在控制台获取 API Key（每月有免费额度）。
- LangSmith: 创建项目并生成 API Key，用于链路追踪。
- 大模型 API: 如智谱 AI、OpenAI 等，需配置
base_url和model。
.env 配置
在根目录创建 .env 文件：
ZHIPUAI_API_KEY=你的 apikey
TAVILY_API_KEY=Tavily APIKEY
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT=https://api.smith.langchain.com
LANGCHAIN_API_KEY=LangSmith 的 APIKEY
LANGSMITH_PROJECT="langgraph-agent"
核心代码实现
1. 导入依赖与环境加载
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
import os
from dotenv import load_dotenv
from pathlib import Path
from langgraph.types import Command, interrupt
from langchain_core.tools import tool
from langchain_tavily import TavilySearch
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage
# 加载环境变量
env_path = Path(__file__).parent.parent / ".env"
load_dotenv(dotenv_path=env_path)
2. 定义状态与工具
class State(TypedDict):
messages: Annotated[list, add_messages]
# 定义人工干预工具
@tool
def human_assistance(query: str) -> str:
"""Request assistance from a human."""
human_response = interrupt({"query": query})
return human_response["data"]
# 定义搜索工具
tool = TavilySearch(max_results=2)
tools = [tool, human_assistance]
3. 构建图与工作流
graph_builder = StateGraph(State)
# 初始化模型
llm = init_chat_model(
model="glm-5",
model_provider="openai",
model_kwargs={"base_url": "https://open.bigmodel.cn/api/paas/v4/"}
)
llm_with_tools = llm.bind_tools(tools)
SYSTEM_PROMPT = """你是一个智能助手，能够记住之前的对话内容。
## 工具使用规则
你可以使用以下工具：
1. tavily_search: 搜索互联网获取实时信息
2. human_assistance: 请求人工专家协助
## 何时必须调用 human_assistance 工具
以下情况你必须调用 human_assistance 工具请求人工专家协助，不要自己回答：
1. 医疗健康：症状诊断、用药建议、治疗方案
2. 法律咨询：合同审核、法律纠纷、法规解释
3. 金融理财：投资建议、税务规划、保险推荐
4. 用户明确要求：用户提到'专家'、'人工'、'专业人士'、'客服'等
5. 高风险决策：可能对用户造成重大影响的决定
6. 不明确答案：任何你不知道的答案，或者设计私有的回答
调用时，将用户的问题整理后作为 query 参数传递。"""
def chatbot(state: State):
messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
message = llm_with_tools.invoke(messages)
assert len(message.tool_calls) <= 1
return {"messages": [message]}
graph_builder.add_node("chatbot", chatbot)
from langgraph.prebuilt import ToolNode, tools_condition
tool_node = ToolNode(tools=tools)
graph_builder.add_node("tools", tool_node)
graph_builder.add_conditional_edges("chatbot", tools_condition)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)
4. 运行与交互
config = {"configurable": {"thread_id": "1"}}
def stream_graph_updates(user_input: str):
events = graph.stream({"messages": [{"role": "user", "content": user_input}]}, config, stream_mode="values")
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()
def check_and_resume_interrupt():
snapshot = graph.get_state(config)
if snapshot.next:
print("\n" + "="*50)
print("[人工干预请求] AI 需要您的专业协助！")
print("="*50)
if snapshot.values.get("messages"):
last_message = snapshot.values["messages"][-1]
print(f"AI 的问题：{last_message.content}")
human_response = input("请输入您的专业回复：")
human_command = Command(resume={"data": human_response})
events = graph.stream(human_command, config, stream_mode="values")
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()
return True
return False
print("="*50)
print("聊天机器人已启动！输入 'quit' 或 'exit' 退出")
print("当 AI 需要专业协助时，会自动请求人工干预")
print("="*50)
while True:
try:
if check_and_resume_interrupt():
continue
user_input = input("\nUser: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
stream_graph_updates(user_input)
check_and_resume_interrupt()
except KeyboardInterrupt:
print("\nGoodbye!")
break
except Exception as e:
print(f"发生错误：{e}")
break
运行结果示例
系统成功调用 Tavily 搜索工具，返回相关商品信息摘要。当用户提出涉及医疗、法律或明确要求转人工的问题时，流程会暂停并等待人工输入，随后恢复执行。
例如查询商品价格时，Agent 自动调用 Tavily 搜索京东实时数据；当用户询问手机故障维修政策并要求转人工时，触发 human_assistance 工具，控制台显示干预提示，输入回复后继续对话。
