Python 搭建具备记忆与人工干预的 Agent 机器人

2. 定义人工干预机制

这是本项目的核心亮点。利用 interrupt 函数，当触发特定条件时，程序会暂停并等待人工输入，恢复后继续执行。

from langgraph.types import Command, interrupt
from langchain_core.tools import tool

@tool
def human_assistance(query: str) -> str:
    """请求人工专家协助"""
    # 中断当前执行，等待用户输入
    human_response = interrupt({"query": query})
    return human_response["data"]

3. 配置工具与状态图

将搜索工具和人工干预工具注册到 Agent 中，并定义状态结构。

from langchain_tavily import TavilySearch

# 定义工具列表
tool = TavilySearch(max_results=2)
tools = [tool, human_assistance]

# 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

4. 设置聊天模型与提示词

使用系统提示词（System Prompt）明确告知模型何时必须调用人工干预，例如涉及医疗、法律或用户明确要求转人工的场景。

from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage

# 初始化模型 (以智谱为例)
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 参数传递。
"""

5. 构建节点与边

定义聊天节点和工具节点，并通过条件边连接。

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)

# 添加条件边：如果调用工具则去 tools 节点，否则结束
graph_builder.add_conditional_edges("chatbot", tools_condition)
# 工具执行后返回聊天节点
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

6. 编译与运行

启用检查点（Checkpointer）以支持记忆，并编写主循环处理中断恢复。

from langgraph.checkpoint.memory import InMemorySaver

memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)

# 设置会话线程 ID，保持上下文记忆
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

if __name__ == "__main__":
    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

运行效果示例

正常搜索场景

当用户询问商品价格等公开信息时，Agent 会自动调用 Tavily 搜索工具并整合结果回答。

User: 你好，请你为我查看一下今天的京东上面的 iphone17pro 的售价在多少

Ai Message: 我来帮您查询京东上 iPhone 17 Pro 的价格信息...
Tool Calls: tavily_search
...
Ai Message: 根据搜索结果，iPhone 17 Pro 256GB 起售价为 8999 元...

人工干预场景

当用户提出涉及售后维修等敏感问题，或明确要求转人工时，Agent 会触发中断。

User: 那我需要知道一下客服可以怎样为我处理我的手机坏掉的原因？需要转人工

[AI 触发 human_assistance 工具]

==================================================
[人工干预请求] AI 需要您的专业协助！
==================================================
AI 的问题：用户询问手机故障问题... 用户明确要求转接人工客服。

请输入您的专业回复：您好，手机坏掉了，如果是人为损坏的，我们是不进行退货退款的哟，但是您可以申请售后，进行售后维修。

Ai Message: 好的，我已经为您转接人工客服。客服的回复如下：
---
客服回复：您好，手机坏掉了，如果是人为损坏的，我们是不进行退货退款的哟，但是您可以申请售后，进行售后维修。
---

通过这种方式，我们既利用了 AI 的高效处理能力，又在关键节点保留了人类专家的判断权，实现了安全可控的自动化服务。