PythonAI算法
基于 LangGraph 的 Python 记忆机器人：支持搜索工具与人工干预
基于 Python 和 LangGraph 框架构建具备记忆功能的聊天机器人。集成 Tavily 搜索引擎实现联网查询能力，并通过 interrupt 机制实现人工干预流程。配置环境变量后，模型可根据预设规则在特定场景下请求人工协助，确保回答准确性。项目包含完整代码示例及运行逻辑说明，适合大模型应用开发入门参考。
基于 Python 和 LangGraph 框架构建具备记忆功能的聊天机器人。集成 Tavily 搜索引擎实现联网查询能力，并通过 interrupt 机制实现人工干预流程。配置环境变量后，模型可根据预设规则在特定场景下请求人工协助，确保回答准确性。项目包含完整代码示例及运行逻辑说明，适合大模型应用开发入门参考。
本文介绍如何使用 Python 和 LangGraph 框架搭建一个具备记忆功能的聊天机器人。该机器人支持调用 Tavily 搜索引擎获取实时信息，并集成了人工干预机制，在特定场景下可请求专家协助。
确保安装 Python 3.9+ 版本。
pip install langgraph langchain-community langchain-openai tavily-python python-dotenv langsmith
python-dotenv 用于读取根目录 .env 中的配置。
Tavily 是为 AI 和大模型设计的搜索引擎，提供实时联网能力。注册后可在后台获取 API Key。
LangSmith 是 LangChain 团队开发的观测平台，用于调试、测试和监控 LLM 应用。开启链路追踪功能有助于可视化跟踪执行步骤。
在根目录创建 .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.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import InMemorySaver
# 加载父目录的 .env 文件
env_path = Path(__file__).parent.parent / ".env"
load_dotenv(dotenv_path=env_path)
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]
graph_builder = StateGraph(State)
# 添加聊天模型
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)
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)
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)
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
程序启动后，用户可通过命令行输入问题。机器人会根据系统提示词判断是否需要调用搜索工具或请求人工干预。
通过 thread_id 保持会话一致性，实现对话记忆的持久化。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online