PythonAI算法

Python 基于 LangGraph 搭建带记忆与人工干预的搜索机器人

Python 结合 LangGraph 框架构建智能聊天机器人，实现对话记忆、联网搜索及人工干预功能。集成 Tavily 工具进行实时信息检索，利用 LangSmith 追踪链路。通过 StateGraph 定义节点与边，配置环境变量加载 API Key。演示了用户查询商品价格触发搜索，以及特定场景下请求人工协助的完整流程，适合 Agent 开发入门参考。

花里胡哨发布于 2026/4/60 浏览

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"

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)

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) # 初始化模型 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)

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[][-].pretty_print() (*) () () (*) : : check_and_resume_interrupt(): user_input = () user_input.lower() [, , ]: () stream_graph_updates(user_input) check_and_resume_interrupt() KeyboardInterrupt: () Exception e: ()

Python 基于 LangGraph 搭建带记忆与人工干预的搜索机器人