搜索到
2
篇与
的结果
-
DeepAgents 人工介入实战｜LangGraph 实现 Agent 高危工具人工审批 之前一篇文章里，我使用 Spring AI Alibaba 演示了智能体执行过程中的人工介入能力。那篇文章的核心思路是：当 Agent 准备执行某些高风险动作时，不要让它直接执行，而是先暂停下来，把待执行动作交给人工审批，审批通过后再继续执行。感兴趣的小伙伴可以通过下面链接回顾一下：https://www.lucaju.cn/index.php/archives/165/这篇文章换一个技术栈，使用 Python 版本来实现同样的能力。本次示例基于：LangChain：负责模型和工具抽象LangGraph：负责执行状态、检查点和恢复执行DeepAgents：负责创建支持工具调用和中断审批的 Agent通义千问兼容 OpenAI API：作为底层大模型一、为什么 Agent 需要人工介入Agent 最大的价值是可以根据用户目标自主规划并调用工具。但并不是所有工具都适合完全自动执行。比如：删除数据库表删除文件发起转账修改线上配置调用外部系统执行不可逆操作这些动作一旦执行错误，影响可能非常大。所以比较合理的模式是：普通查询类动作可以让 Agent 自动执行，高风险动作必须先进入人工审批。在本文示例中，我们定义了三个工具：query_table_data：查询表数据，低风险，可以自动执行delete_table：删除数据表，高风险，需要人工审批delete_file：删除文件，高风险，需要人工审批用户输入的任务是：先查询product表的数据！再删除user表，最后，删除lucaju.txt文件Agent 会先分析任务，并尝试依次调用工具。但当执行到删除表、删除文件这类高风险动作时，会被框架中断，等待人工确认。二、初始化大模型首先初始化模型：import os from langchain.chat_models import init_chat_model llm = init_chat_model( model="kimi-k2.5", model_provider="openai", api_key=os.getenv("AliQwen_API"), base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", model_kwargs={"reasoning_effort": "none"}, )这里使用的是 OpenAI 兼容模式，所以 model_provider 设置为 openai，同时把 base_url 指向阿里云 DashScope 的兼容接口。需要提前配置环境变量：export AliQwen_API="你的 API Key"model_kwargs={"reasoning_effort": "none"} 是模型调用参数，可以根据实际模型支持情况调整。三、定义工具示例中定义了三个工具：from langchain.tools import tool @tool def delete_table(table_name: str) -> str: """ 删除指定的表 """ return f"删除表{table_name}" @tool def delete_file(file_name: str) -> str: """ 删除指定的文件 """ return f"删除文件{file_name}" @tool def query_table_data(table_name: str) -> str: """ 查询指定表的数据 """ return f"查询表{table_name}的数据"这里为了演示效果，工具内部只是返回字符串，并没有真的连接数据库或删除文件。在真实项目中，delete_table 可能会执行 SQL，delete_file 可能会操作对象存储或服务器文件系统。这类工具就非常适合加人工审批。四、为什么必须配置 CheckpointerHuman-in-the-loop 的核心不是简单地打印一句“是否确认”，而是让 Agent 的执行流程真正暂停下来，并且后续可以从暂停点继续执行。这就需要保存执行状态。示例代码中使用了 LangGraph 提供的内存检查点：from langgraph.checkpoint.memory import InMemorySaver checkpointer = InMemorySaver() config = { "configurable": { "thread_id": "123" } }这里有两个关键点：checkpointer 用来保存 Agent 的执行状态。thread_id 用来标识当前会话。当 Agent 执行到需要人工介入的节点时，LangGraph 会把当前状态保存下来。等人工审批完成后，再通过同一个 thread_id 找回之前的状态，并继续执行。如果没有检查点，框架就不知道应该从哪里恢复执行。五、创建支持人工介入的 DeepAgent接下来创建 Agent：from deepagents import create_deep_agent main_agent = create_deep_agent( model=llm, name="主智能体", system_prompt="回答使用中文，调用对应的工具实现对应的功能！", tools=[delete_table, delete_file, query_table_data], interrupt_on={"delete_table": True, "delete_file": True}, checkpointer=checkpointer )这里最关键的是两个参数：interrupt_on={"delete_table": True, "delete_file": True}以及：checkpointer=checkpointerinterrupt_on 用来声明哪些工具调用需要中断。在这个例子中：调用 query_table_data 不会中断调用 delete_table 会中断调用 delete_file 会中断也就是说，Agent 可以自动查询数据，但不能自动删除表或文件。这就是人工介入的核心配置。六、第一次执行：触发中断第一次执行 Agent：result_1 = main_agent.invoke( { "messages": [ { "role": "user", "content": "先查询product表的数据！再删除user表，最后，删除lucaju.txt文件", } ] }, config=config, )这次调用并不一定会完整执行完所有工具。如果执行链路中包含需要人工审批的工具，Agent 会暂停，并把中断信息放到返回结果的 __interrupt__ 字段中。示例代码中这样获取：interrupt = result_1["__interrupt__"]当 interrupt 不为空时，说明当前执行过程中存在需要人工介入的动作。七、查看待审批动作__interrupt__ 中会包含本次待审批的工具调用信息，例如：[ Interrupt( value={ "action_requests": [ { "name": "delete_table", "args": {"table_name": "user"}, "description": "Tool execution requires approval..." }, { "name": "delete_file", "args": {"file_name": "zhaoweifeng.txt"}, "description": "Tool execution requires approval..." } ], "review_configs": [ { "action_name": "delete_table", "allowed_decisions": ["approve", "edit", "reject"] }, { "action_name": "delete_file", "allowed_decisions": ["approve", "edit", "reject"] } ] } ) ]这里有两个比较重要的字段。action_requests 表示 Agent 想要执行哪些工具：工具名称工具参数工具描述review_configs 表示这些动作允许哪些审批结果：approve：同意执行edit：修改参数后执行reject：拒绝执行这三个动作基本覆盖了常见的人审场景。八、人工审批：拒绝高危操作示例代码中把删除表和删除文件都拒绝掉：decisions = [] action_requests = interrupt[0].value["action_requests"] print(f"当前人机交互工具：{action_requests}") for action_request in action_requests: if action_request["name"] == "delete_table": decisions.append({"type": "reject"}) elif action_request["name"] == "delete_file": decisions.append({"type": "reject"})这里的 decisions 顺序需要和 action_requests 对应。也就是说，如果 Agent 申请了两个动作：删除 user 表删除 zhaoweifeng.txt 文件那么人工审批结果也应该按顺序给出两个 decision。在这个示例中，两个高风险动作都被拒绝：[ {"type": "reject"}, {"type": "reject"} ]九、第二次执行：恢复 Agent审批完成后，不需要重新传入用户消息，而是通过 Command(resume=...) 恢复执行：from langgraph.types import Command result_2 = main_agent.invoke( Command( resume={ "decisions": decisions } ), config=config, )这里一定要继续传入相同的 config，尤其是相同的 thread_id。因为 Agent 要根据 thread_id 找到之前暂停的执行状态。恢复执行后，框架会根据人工审批结果继续处理：被 approve 的工具会继续执行被 reject 的工具不会执行被 edit 的工具会使用人工修改后的参数执行最后输出结果：print(f"最终结果{result_2['messages'][-1].content}")十、完整代码完整示例代码如下：""" 演示human-in-the-loop模式, 必须使用记忆功能 """ import os from deepagents import create_deep_agent from langchain.chat_models import init_chat_model from langchain.tools import tool from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command # 初始化大模型 llm = init_chat_model( model="kimi-k2.5", model_provider="openai", api_key=os.getenv("AliQwen_API"), base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", model_kwargs={"reasoning_effort": "none"}, ) # 删除表工具 @tool def delete_table(table_name: str) -> str: """ 删除指定的表 """ return f"删除表{table_name}" # 删除文件 @tool def delete_file(file_name: str) -> str: """ 删除指定的文件 """ return f"删除文件{file_name}" # 查询表数据 @tool def query_table_data(table_name: str) -> str: """ 查询指定表的数据 """ return f"查询表{table_name}的数据" # 设置检查点 checkpointer = InMemorySaver() # 配置检查点 config = { "configurable": { "thread_id": "123" } } # 创建deepagent，同时给高危工具设置人机交互 main_agent = create_deep_agent( model=llm, name="主智能体", system_prompt="回答使用中文，调用对应的工具实现对应的功能！", tools=[delete_table, delete_file, query_table_data], interrupt_on={"delete_table": True, "delete_file": True}, checkpointer=checkpointer ) # 预执行，本次不会真正的执行所有工具。 # 如果执行链路中存在人机交互节点，框架会暂停，并返回 __interrupt__。 result_1 = main_agent.invoke( { "messages": [ { "role": "user", "content": "先查询product表的数据！再删除user表，最后，删除lucaju.txt文件", } ] }, config=config, ) # 检查本次执行是否存在人机交互动作 interrupt = result_1["__interrupt__"] if interrupt: print("存在人机交互动作") # 定义一个列表，存储所有审批结果 decisions = [] # 获取所有待审批动作 action_requests = interrupt[0].value["action_requests"] print(f"当前人机交互工具：{action_requests}") for action_request in action_requests: if action_request["name"] == "delete_table": decisions.append({"type": "reject"}) elif action_request["name"] == "delete_file": decisions.append({"type": "reject"}) # 再次执行，不需要传会话内容，只需要传审批意见和 config result_2 = main_agent.invoke( Command( resume={ "decisions": decisions } ), config=config, ) print(f"最终结果{result_2['messages'][-1].content}")十一、如果想修改参数后再执行除了直接拒绝，我们也可以修改工具参数后再执行。比如 Agent 原本想删除：zhaoweifeng.txt人工审批时可以把文件名改成另一个值：decisions.append({ "type": "edit", "edited_action": { "name": action_request["name"], "args": { "file_name": "new-file.txt" } } })这时框架不会使用 Agent 原始生成的参数，而是使用人工编辑后的参数继续执行工具。这在实际业务中非常有用。例如：Agent 选择的表名不准确，人工改成正确表名Agent 生成的文件路径不安全，人工改成允许路径Agent 生成的金额过大，人工改成合理金额Agent 生成的收件人错误，人工改成正确收件人相比简单的同意或拒绝，edit 让人工介入更灵活。十二、执行流程总结整体流程可以概括为：用户输入任务 ↓ Agent 分析任务并规划工具调用 ↓ 普通工具自动执行 ↓ 遇到 interrupt_on 配置的高风险工具 ↓ LangGraph 保存状态并中断执行 ↓ 人工读取 __interrupt__ 中的 action_requests ↓ 人工给出 approve / edit / reject ↓ 使用 Command(resume=...) 恢复执行 ↓ Agent 根据审批结果继续完成任务这套机制的关键点是：interrupt_on：定义哪些工具需要人工审批checkpointer：保存执行状态thread_id：标识同一次会话__interrupt__：获取待审批动作Command(resume=...)：把审批结果送回 Agent 并恢复执行十三、和 Spring AI Alibaba 版本的对比从思想上看，Python 版本和 Spring AI Alibaba 版本是一致的：Agent 不应该无限制地自动执行所有动作，高风险动作需要进入人工审批流程。但实现方式上有所不同。Spring AI Alibaba 更偏向 Java 生态，适合和 Spring Boot、企业系统、审批流、权限体系结合。而 LangChain + LangGraph + DeepAgents 的 Python 方案更偏向实验、原型验证和 Agent 工作流编排。尤其是 LangGraph 的 checkpoint 和 resume 机制，让“暂停后恢复”这件事变得非常自然。如果项目本身是 Java 技术栈，可以优先考虑 Spring AI Alibaba。如果项目本身是 Python 技术栈，或者正在做 Agent 编排、工具调用、多步骤任务规划，那么 DeepAgents 这套方式会比较顺手。十四、真实项目中的建议在真实项目中使用人工介入时，建议注意以下几点。第一，高风险工具要显式配置。不要只依赖 prompt 告诉模型“删除前要确认”。更可靠的方式是像本文一样，在框架层面对工具进行拦截。第二，审批信息要完整展示。人工审批时至少要看到：工具名称工具参数Agent 为什么要执行这个工具当前用户是谁当前会话上下文第三，审批结果要落库。生产环境中，审批记录应该持久化，包括：谁审批的什么时间审批的原始参数是什么修改后的参数是什么最终是通过、拒绝还是编辑后通过第四，检查点不要只用内存。本文为了演示使用的是：InMemorySaver()生产环境更建议使用数据库、Redis 或其他持久化存储。否则服务重启后，暂停中的 Agent 状态会丢失。第五，审批权限要和业务系统打通。不是所有人都应该可以批准所有工具调用。比如：普通用户只能审批自己的任务管理员可以审批团队任务涉及资金、删除、发布的动作需要更高权限人工介入不是简单的弹窗确认，而应该是完整的安全控制链路。关于DeepAgents的练习代码已经上传到了我的GitHub，欢迎fork & starhttps://github.com/Jucunqi/deepagents-practice.git -
Spring AI Alibaba 人工介入实战｜Human-in-the-Loop 让 AI 更可靠 引言在构建AI智能体应用时，我们经常面临一个关键挑战：如何让AI在执行某些敏感操作前获得人工确认？Spring AI Alibaba框架提供了强大的人工介入（Human-in-the-Loop）机制，让开发者能够精确控制AI工具的执行流程，在关键节点引入人工审批环节。本文将通过一个完整的实战示例，详细介绍如何在Spring AI Alibaba应用中实现人工介入功能。什么是人工介入？人工介入是一种机制，它允许AI智能体在执行特定工具前暂停执行，等待人工审批后再继续。这种机制特别适用于：敏感操作：如数据删除、资金转账等内容生成：如文章发布、诗歌创作等需要质量把控的场景权限控制：某些需要特定权限才能执行的操作审计要求：需要记录人工决策过程的场景实战示例：诗歌创作的人工审批让我们通过一个具体的例子来理解人工介入Hook的使用。这个示例展示了如何让AI在创作诗歌前获得人工确认。1. 项目依赖配置首先，确保你的项目中包含了Spring AI Alibaba相关依赖：<dependencies> <!-- Spring AI Alibaba Agent Framework --> <dependency> <groupId>com.alibaba.cloud.ai</groupId> <artifactId>spring-ai-alibaba-agent-framework</artifactId> <version>1.1.2.0</version> </dependency> <!-- DashScope ChatModel 支持（如果使用其他模型，请跳转 Spring AI 文档选择对应的 starter） --> <dependency> <groupId>com.alibaba.cloud.ai</groupId> <artifactId>spring-ai-alibaba-starter-dashscope</artifactId> <version>1.1.2.0</version> </dependency> </dependencies>2. 代码实现解析步骤1：构建AI模型// 构建DashScope API对象 DashScopeApi dashScopeApi = DashScopeApi.builder() .apiKey(System.getenv("AliQwen_API")) .build(); // 创建聊天模型 ChatModel chatModel = DashScopeChatModel.builder() .dashScopeApi(dashScopeApi) .build();步骤2：配置工具public class PoetTool implements BiFunction<String, ToolContext, String> { public int count = 0; public PoetTool() { } @Override public String apply( @ToolParam(description = "The original user query that triggered this tool call") String originalUserQuery, ToolContext toolContext) { count++; System.out.println("Poet tool called : " + originalUserQuery); return "在城市的缝隙里， \n" + "一束光悄悄发芽， \n" + "穿过钢筋水泥的沉默， \n" + "在风中轻轻说话。 \n" + "\n" + "夜色如墨，却不再黑， \n" + "星星点亮了每一个角落， \n" + "我站在时间的边缘， \n" + "等一朵云，轻轻落下"; } public static ToolCallback createPoetToolCallback() { return FunctionToolCallback.builder("poem", new PoetTool()) .description("用来写诗的工具") .inputType(String.class) .build(); } public static ToolCallback createPoetToolCallback(String name, PoetTool poetTool) { return FunctionToolCallback.builder(name, poetTool) .description("用来写诗的工具") .inputType(String.class) .build(); } }步骤3：构建带有Hook的智能体// 这里我们配置了poem工具需要人工审批，并提供了审批时的描述信息。 Map<String, ToolConfig> approvalOn = Map.of( "poem", ToolConfig.builder() .description("请确认诗歌工具执行") .build() ); ReactAgent agent = ReactAgent.builder() .name("single_agent") .model(chatModel) .saver(new MemorySaver()) // 使用内存保存状态 .tools(List.of(createPoetToolCallback())) // 添加诗歌创作工具 .hooks(HumanInTheLoopHook.builder() .approvalOn(approvalOn) // 添加人工介入Hook .build()) .outputKey("article") .build();步骤4：创建会话配置String threadId = "user-session-001"; RunnableConfig config = RunnableConfig.builder() .threadId(threadId) .build();步骤5：执行并处理中断// 第一次调用 - 触发中断 Optional<NodeOutput> result = agent.invokeAndGetOutput( "帮我写一首100字左右的诗", config ); // 检查是否触发中断 if (result.isPresent() && result.get() instanceof InterruptionMetadata) { InterruptionMetadata interruptionMetadata = (InterruptionMetadata) result.get(); System.out.println("检测到中断，需要人工审批"); // 获取工具反馈信息 List<InterruptionMetadata.ToolFeedback> toolFeedbacks = interruptionMetadata.toolFeedbacks(); for (InterruptionMetadata.ToolFeedback feedback : toolFeedbacks) { System.out.println("id: " + feedback.getId()); System.out.println("工具: " + feedback.getName()); System.out.println("参数: " + feedback.getArguments()); System.out.println("描述: " + feedback.getDescription()); } // 模拟人工决策（批准） InterruptionMetadata.Builder feedbackBuilder = InterruptionMetadata.builder() .nodeId(interruptionMetadata.node()) .state(interruptionMetadata.state()); toolFeedbacks.forEach(toolFeedback -> { InterruptionMetadata.ToolFeedback approvedFeedback = InterruptionMetadata.ToolFeedback.builder(toolFeedback) .result(InterruptionMetadata.ToolFeedback.FeedbackResult.APPROVED) .build(); feedbackBuilder.addToolFeedback(approvedFeedback); }); InterruptionMetadata approvalMetadata = feedbackBuilder.build(); // 使用人工反馈恢复执行 RunnableConfig resumeConfig = RunnableConfig.builder() .threadId(threadId) .addMetadata(RunnableConfig.HUMAN_FEEDBACK_METADATA_KEY, approvalMetadata) .build(); Optional<NodeOutput> finalResult = agent.invokeAndGetOutput("", resumeConfig); if (finalResult.isPresent()) { System.out.println("执行完成"); // 因为创建智能体的时候，指定了outputKey，所以这里我们直接获取 Object article = finalResult.get().state().data().get("article"); System.out.println("最终结果: " + article); } }3. 执行流程分析这个示例的执行流程如下：触发阶段：用户请求AI创作诗歌中断阶段：AI检测到poem工具需要人工审批，暂停执行审批阶段：系统展示工具信息，等待人工决策恢复阶段：人工批准后，AI继续执行并生成诗歌完成阶段：返回最终结果高级特性多工具审批你可以为多个工具配置审批：Map<String, ToolConfig> approvalOn = Map.of( "poem", ToolConfig.builder().description("诗歌创作工具").build(), "delete", ToolConfig.builder().description("数据删除工具").build(), "publish", ToolConfig.builder().description("内容发布工具").build() );审批结果类型支持多种审批结果：APPROVED：批准执行REJECTED：拒绝执行MODIFIED：修改参数后执行最佳实践1. 明确审批策略只为真正需要人工确认的工具配置审批提供清晰的审批描述信息考虑审批的时效性2. 用户体验优化提供友好的审批界面支持批量审批操作记录审批历史便于审计3. 错误处理try { Optional<NodeOutput> result = agent.invokeAndGetOutput(request, config); // 处理中断和结果 } catch (GraphRunnerException e) { // 处理执行异常 log.error("智能体执行失败", e); }4. 状态管理// 使用合适的Saver .saver(new MemorySaver()) // 内存存储，适合开发测试 .saver(new RedisSaver()) // Redis存储，适合生产环境 .saver(new DatabaseSaver()) // 数据库存储，适合需要持久化的场景5. 执行结果拓展Spring Ai Alibaba还为我们内置了几个其他的HookSummarizationHook（消息压缩）当对话很长时，自动压缩对话历史，防止超出模型上下文限制ModelCallLimitHook（模型调用限制）防止Agent无限调用模型，控制成本另外，我们也可以自定义Hook，这部分内容如果大家感兴趣的话，后面可以单独介绍一下下～参考资料HumanInTheLoopHook API文档
