基于 LangGraph 的 Python 记忆机器人：支持搜索工具与人工干预

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)

# 添加聊天模型
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)

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

运行结果说明

程序启动后，用户可通过命令行输入问题。机器人会根据系统提示词判断是否需要调用搜索工具或请求人工干预。

1. 搜索场景：询问商品价格等信息时，机器人自动调用 Tavily 搜索并返回结果。
2. 人工干预场景：涉及医疗、法律等高风险领域或用户明确要求转人工时，程序会暂停并等待输入人工回复，随后继续流程。

通过 thread_id 保持会话一致性，实现对话记忆的持久化。