彻底说清 Human-in-the-Loop：企业级 Agent 系统的关键挑战与LangGraph解法【上】

from langgraph.types import interrupt, Command
...
#这是一个Agent的某个人工参与的节点
def human_review_node(state: State):

   # 暂停执行，输出需人工审核的数据
    review_data = {"question": "请审核以下内容:", "output": state["llm_output"]}
    decision = interrupt(review_data) 

    # 恢复后将根据人工决策更新状态或跳转
    if decision == "approve":
        return Command(goto="approved_node")
    else:
        return Command(goto="rejected_node")
...

...调用agent客户端程序...
    result = graph.invoke(initial_state, config) #调用Agent启动工作流
    interrupt_info = result['__interrupt__'][0].value

    ...显示中断信息，人类交互与反馈...

    # 假设 thread_id 标识此次任务，再次调用invoke恢复运行即可
    user_decision = "approve"  # 这里模拟用户最后的反馈
    result = graph.invoke(Command(resume=user_decision), config={"configurable": {"thread_id": thread_id}})
...

...
# 初始化 PostgreSQL 检查点保存器
with PostgresSaver.from_conn_string("postgresql://postgres:yourpassword@localhost/postgres?sslmode=disable") as checkpointer:
        checkpointer.setup()
        graph = builder.compile(checkpointer=checkpointer)

def human_approval_node(state: State):

        .....从历史消息或者状态获得工具调用消息:tool_calls.....
        tool_calls_info = []
        
        # 获取工具调用信息（这里暂时只取第一个演示）
        tc = tool_calls[0]
        tool_id = tc.get("id", "未知工具ID")
        tool_name = tc.get("name", "未知工具")
        tool_args = tc.get("args", {})
        tool_calls_info.append(f"{tool_name}({tool_args})")
            
        # 非高风险工具自动批准
        if tool_name not in HIGH_RISK_TOOLS:
            return {"human_approved": True}
        
        tool_calls_str = "\n - ".join(tool_calls_info)
        
        # 高风险工具：中断并等待人工审批
        value = interrupt({
            "tool_calls": tool_calls_str,
            "message": "请输入 'ok' 批准工具使用，或输入 'reject' 拒绝"
        })
...

async def tavily_search(query: str, search_depth: Optional[str] = "basic"):
    ...
    # 中断执行，等待人工审核
    response = interrupt({
        "tool": "tavily_search",
        "args": {
            "query": query,
            "search_depth": search_depth
        },
        "message": f"准备使用Tavily搜索：\n- 查询内容: {query}\n- 搜索深度: {search_depth}\n\n是否允许继续？\n输入 'yes' 接受，'no' 拒绝，或 'edit' 修改查询关键词",
    }) 

    # 处理人工响应
    if response["type"] == "accept":
        pass
    elif response["type"] == "edit":
        query = response["args"]["query"]
    else:
        returnf"该工具被拒绝使用，请尝试其他方法或拒绝回答问题。"
    
    ...开始执行真正的工具逻辑...

@human_in_the_loop()
def tavily_search(query: str, search_depth: str = "basic"):
    """使用Tavily进行网络搜索"""
    try:
......