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

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)

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

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

运行结果示例

系统成功调用 Tavily 搜索工具，返回相关商品信息摘要。当用户提出涉及医疗、法律或明确要求转人工的问题时，流程会暂停并等待人工输入，随后恢复执行。

例如查询商品价格时，Agent 自动调用 Tavily 搜索京东实时数据；当用户询问手机故障维修政策并要求转人工时，触发 human_assistance 工具，控制台显示干预提示，输入回复后继续对话。