Python 构建带记忆与人工干预的搜索机器人

定义人工干预机制

利用 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 的问题：用户询问手机坏掉的原因及售后处理流程
请输入您的专业回复：您好，手机坏掉了，如果是人为损坏的，我们是不进行退货退款的哟，但是您可以申请售后，进行售后维修。

后续机器人会将人工回复整合进对话上下文反馈给用户。