Python 构建带记忆与人工干预的搜索机器人
前言
在 LangGraph 官方文档的启发下，我决定尝试构建一个具备长期记忆、支持联网搜索并能触发人工干预的智能助手。虽然背景是 Java 开发，但 AI 应用开发的快速迭代让我意识到掌握 Agent 工作流的重要性。本文将分享基于 LangGraph 搭建此类机器人的实战经验。
前置准备
环境配置
推荐使用 Python 3.9+（本文基于 3.12）：
pip install langgraph langchain-community langchain-openai tavily-python python-dotenv langsmith
python-dotenv 用于读取根目录下的 .env 配置文件。
工具介绍
- Tavily：专为 AI 设计的搜索引擎，提供实时联网能力。每月有免费额度，适合学习和测试。
- LangSmith：LangChain 团队的可观测性平台，用于调试和监控 LLM 应用。支持免费额度，可追踪每一步执行细节。
密钥获取与环境配置
- Tavily：登录后进入首页，在左侧菜单添加 API Key。
- LangSmith：登录后进入仪表盘，点击左下角账号设置生成 API Key，并创建一个新的 Project 用于追踪。
- 大模型 API：支持智谱、阿里云等，需自行获取。
在根目录创建 .env 文件，填入如下内容：
# 智谱 API Key
ZHIPUAI_API_KEY=你的 apikey
# Tavily 智能体搜索工具
TAVILY_API_KEY=Tavily APIKEY
# LangSmith
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT=https://api.smith.langchain.com
LANGCHAIN_API_KEY=LangSmith 的 APIKEY
LANGSMITH_PROJECT="langgraph-agent"
核心代码实现
依赖导入与环境加载
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
from langgraph.checkpoint.memory import InMemorySaver
# 加载父目录的 .env 文件
env_path = Path(__file__).parent.parent / ".env"
load_dotenv(dotenv_path=env_path)
定义人工干预机制
利用 interrupt 实现人机协作，当遇到高风险或不确定问题时暂停流程等待人工输入：
@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]
# 配置聊天模型
print("env 中的智谱 APIKEY：", os.getenv("ZHIPUAI_API_KEY"))
os.environ["OPENAI_API_KEY"] = os.getenv("ZHIPUAI_API_KEY")
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)
# 系统提示词：引导 AI 在特定场景调用人工干预
SYSTEM_PROMPT = """
你是一个智能助手，能够记住之前的对话内容。
## 工具使用规则
你可以使用以下工具：
1. tavily_search: 搜索互联网获取实时信息
2. human_assistance: 请求人工专家协助
## 何时必须调用 human_assistance 工具
以下情况你必须调用 human_assistance 工具请求人工专家协助，不要自己回答：
1. **医疗健康**: 症状诊断、用药建议、治疗方案
2. **法律咨询**: 合同审核、法律纠纷、法规解释
3. **金融理财**: 投资建议、税务规划、保险推荐
4. **用户明确要求**: 用户提到"专家"、"人工"、"专业人士"、"客服"等
5. **高风险决策**: 可能对用户造成重大影响的决定
6. **不明确答案**: 任何你不知道的答案，或者设计私有的回答
调用时，将用户的问题整理后作为 query 参数传递。
"""
构建状态图 (StateGraph)
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
def chatbot(state: State):
# 在消息列表开头添加 system prompt
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)
# 工具执行后返回 chatbot 节点
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
# 编译图并添加内存
memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)
运行逻辑
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 搜索最新信息并汇总结果。
场景二：人工干预触发
当用户询问涉及医疗、法律等高风险问题或明确要求转人工时，程序会暂停并打印 [人工干预请求]，等待控制台输入回复后继续执行。
例如：
[人工干预请求] AI 需要您的专业协助！
AI 的问题：用户询问手机坏掉的原因及售后处理流程
请输入您的专业回复：您好，手机坏掉了，如果是人为损坏的，我们是不进行退货退款的哟，但是您可以申请售后，进行售后维修。
后续机器人会将人工回复整合进对话上下文反馈给用户。
