1. 项目概述为什么你的智能体需要一个“审批门”最近和几个做AI应用落地的朋友聊天发现大家不约而同地踩进了同一个坑辛辛苦苦训练或调教出来的智能体Agent在测试环境里表现堪称完美逻辑清晰、响应迅速可一旦部署到生产环境面对真实、复杂且充满不确定性的用户请求时就开始“放飞自我”。轻则给出一些无关痛痒但略显尴尬的回复重则可能执行一些未经授权的操作比如在未经确认的情况下调用付费API、修改了关键数据库字段甚至对外发送了不合规的内容。这让我意识到单纯追求Agent的“智能”和“自主”是远远不够的我们还需要为它套上一个“缰绳”一个在关键时刻能踩下刹车、让人工介入审查的机制——这就是“审批门”Approval Gate。“给Agent加一个审批门”这个项目核心要解决的就是在自动化流程中嵌入人工监督节点的问题。它不是一个独立的功能而是一套设计模式与工程实践旨在平衡效率与风险。想象一下你构建了一个能自动处理客户报销单的Agent它可以识别发票、核对政策、计算金额。但遇到一张金额巨大、发票模糊或政策边缘的报销单时你肯定不希望它直接批准付款。这时审批门就应该被触发将当前上下文、Agent的分析建议以及待执行的操作打包成一个清晰的审批任务推送给指定的负责人比如部门经理等待他的“同意”或“驳回”指令后流程再继续。这个机制听起来简单但要做好却涉及架构设计、状态管理、用户体验和安全策略等多个层面。它不仅仅是弹出一个确认框那么简单而是需要将审批逻辑深度集成到Agent的决策循环中确保在需要人工介入时整个系统能优雅地暂停、清晰地展示、安全地等待并在获得反馈后准确地恢复执行。接下来我将结合具体的实践拆解如何为你的Agent设计和实现一个健壮、灵活且非侵入式的审批门系统。2. 审批门系统的核心设计思路与架构选型为Agent添加审批门首要问题是如何设计。是让Agent每走一步都问一下还是只在关键风险点介入不同的业务场景需要不同的设计哲学。我的经验是遵循“最小必要”和“上下文完整”两大原则。2.1 基于规则与基于模型的两种触发策略审批门的触发逻辑是整个系统的“大脑”。通常有两种设计思路基于规则的触发策略这是最直接、最可控的方式。你可以预先定义一系列明确的规则当Agent的执行触达这些规则时便触发审批。例如操作敏感度规则任何涉及金钱交易金额大于X元、数据删除、用户权限变更、对外发送重要通知的操作。内容合规规则生成或处理的内容涉及特定关键词如保密信息、敏感话题、或情感倾向过于极端。不确定性阈值规则当Agent自身对决策的信心度confidence score低于某个阈值时。例如一个分类Agent对当前用户意图的分类概率如果只有60%与其让它猜不如交由人工判断。这种策略的优势是规则透明易于审计和调试。缺点是规则库可能会变得庞大且难以维护并且无法处理规则未覆盖的、新的复杂边缘情况。基于模型的触发策略这是一种更“智能”但也更复杂的方式。你可以训练一个专门的“风险预测模型”或使用大语言模型LLM本身来评估当前Agent决策的风险等级。这个模型会分析当前的对话历史、Agent即将执行的操作、以及操作涉及的上下文输出一个“需要人工审批”的概率。当概率超过阈值时则触发审批门。例如你可以设计一个Prompt给LLM“请分析以下Agent即将执行的操作和对话上下文判断其是否存在财务风险、法律风险、声誉风险或操作风险。如果存在任何中等及以上风险请回答‘需要审批’并简述理由。” 这种方式灵活性高能适应更复杂的场景但引入了另一个模型的可靠性和延迟问题并且决策过程可解释性相对较弱。在实际项目中我通常采用混合策略用硬性规则守住底线如涉及金钱、删除必须审批同时用模型策略来覆盖那些模糊的、需要语义理解的灰色地带如判断用户请求是否带有恶意或欺诈倾向。2.2 审批门在Agent架构中的位置审批门不应该成为Agent代码里四处散落的if-else判断而应该作为一个独立的服务或模块以非侵入的方式嵌入到Agent的执行链路中。一个典型的架构是在Agent的“行动层”Action Layer或“工具调用层”Tool Calling Layer进行拦截。以基于函数调用Function Calling的Agent为例其工作流程通常是LLM分析用户请求 - 决定调用某个工具函数 - 执行工具 - 将结果返回给LLM - LLM生成最终回复。审批门的最佳插入点就是在LLM决定调用工具之后、实际执行工具之前。用户请求 | v LLM 分析 决策 | v [审批门拦截点] --- 检查即将调用的工具和参数 | | (需要审批) v 暂停Agent执行 生成审批任务 - 通知审批人 - 等待审批结果 | | (审批通过) v 执行工具函数 | v LLM 生成回复 | v 返回用户在这种架构下Agent核心逻辑完全感知不到审批门的存在它只是正常地输出要执行的动作。审批门服务像一个“过滤器”或“代理”负责检查这些动作并在必要时将其路由到人工审批队列。这种解耦设计使得审批策略可以独立更新和扩展而无需修改Agent本体。2.3 状态管理与流程恢复被暂停的Agent如何“续命”这是实现审批门最具挑战性的部分。当审批被触发Agent的执行流被暂停可能已经消耗了Tokens并且包含了多轮对话的复杂上下文。几天后审批可能需要时间当审批人做出决定后系统如何准确地让Agent从当初暂停的地方继续执行而不是重新开始或丢失状态关键在于上下文快照与任务持久化。在触发审批的瞬间系统必须立即保存一个完整的“执行现场快照”这至少包括会话ID唯一标识当前对话线程。完整的对话历史包括所有用户消息和Agent的响应。被拦截的工具调用请求包括工具名称和传入的参数。Agent的内部状态如果存在例如在多步骤任务中Agent当前处于哪个阶段。触发审批的规则或原因。这个快照需要被持久化到数据库如PostgreSQL, MongoDB中并与一个唯一的“审批任务ID”关联。然后Agent的执行线程可以被安全释放。当审批完成时系统根据“审批任务ID”找回完整的快照重新实例化或唤醒Agent加载相同的对话历史和上下文然后根据审批结果通过或驳回注入一条特殊的系统消息例如“关于之前申请执行工具X的操作已获得批准。请继续。” 或 “关于之前申请执行工具X的操作已被驳回。驳回原因是Y。请据此调整你的回复。”实操心得保存对话历史时要注意Token长度限制。如果历史很长可以考虑只保存最近N轮或采用更智能的摘要方式。同时重新实例化Agent时要确保模型参数、提示词模板等与暂停前完全一致否则行为可能发生漂移。3. 审批门的具体实现与核心环节理论讲完我们进入实战环节。我将以一个虚拟的“智能客服Agent”为例它拥有“处理订单退款”的权限。我们将为这个退款操作添加一个审批门当退款金额超过500元时需要人工确认。3.1 定义审批规则与元数据首先我们需要一个地方来管理规则。可以创建一个简单的规则配置表在代码中可以用字典或配置文件实现。# approval_rules.py APPROVAL_RULES [ { id: rule_001, name: 大额退款审批, description: 单笔退款金额超过阈值时需要经理审批, condition_type: tool_call, # 规则作用于工具调用 tool_name: process_refund, # 监控的工具名称 condition: lambda context: context.get(refund_amount, 0) 500, # 判断逻辑 approver_role: finance_manager, # 审批人角色 approval_form_template: { # 审批任务表单模板 title: 大额退款申请审批, fields: [ {key: order_id, label: 订单号, value: context[order_id]}, {key: refund_amount, label: 退款金额, value: context[refund_amount]}, {key: reason, label: 申请原因, value: context[customer_reason]}, {key: agent_analysis, label: Agent分析, value: context.get(agent_analysis, )} ] } }, # 可以定义更多规则... ]这里的condition是一个函数它接收一个context字典里面包含了Agent试图调用工具时的所有参数。规则引擎会在拦截点执行这个函数来判断是否触发审批。3.2 实现Agent执行拦截器接下来我们需要在Agent调用工具的地方插入拦截逻辑。假设我们使用LangChain框架可以通过自定义Tool类或中间件来实现。# approval_gate_middleware.py import asyncio from typing import Any, Dict from langchain.tools import BaseTool from .approval_rules import APPROVAL_RULES from .approval_task_service import create_approval_task, wait_for_approval class ApprovalGateMiddleware: 审批门中间件包装原始的Tool def __init__(self, original_tool: BaseTool): self.original_tool original_tool async def run(self, tool_input: str, **kwargs) - str: # 1. 检查规则 context {tool_name: self.original_tool.name, **kwargs} triggered_rule None for rule in APPROVAL_RULES: if rule[tool_name] self.original_tool.name: try: if rule[condition](context): triggered_rule rule break except Exception as e: # 规则执行出错出于安全考虑可以触发审批或直接拒绝 print(f规则执行错误 {rule[id]}: {e}) return 系统检测到规则异常操作已被暂停请联系管理员。 # 2. 如果触发规则创建审批任务并等待 if triggered_rule: approval_task_id create_approval_task( rule_idtriggered_rule[id], contextcontext, form_templatetriggered_rule[approval_form_template] ) # 通知审批人可通过邮件、钉钉、飞书机器人等 notify_approver(approval_task_id, triggered_rule[approver_role]) # 等待审批结果这里可以是轮询数据库或使用异步事件 approval_result await wait_for_approval(approval_task_id, timeout72*3600) # 超时72小时 if not approval_result or approval_result[status] ! approved: reason approval_result.get(reason, 未说明原因) if approval_result else 审批超时或未通过 return f您的操作未获批准。原因{reason}。请调整您的请求或联系相关人员。 # 审批通过注入批准信息到上下文可选供后续Agent参考 kwargs[_approval_note] f操作已由 {approval_result[approver]} 于 {approval_result[time]} 批准。 # 3. 执行原始工具逻辑 return await self.original_tool.arun(tool_input, **kwargs) # 在初始化Agent时用中间件包装所有工具 def wrap_tools_with_gate(tools): wrapped_tools [] for tool in tools: wrapped_tool ApprovalGateMiddleware(original_tooltool) # 注意需要将wrapped_tool适配成LangChain能识别的Tool对象这里省略适配器代码 wrapped_tools.append(wrapped_tool) return wrapped_tools3.3 构建审批任务管理与用户界面审批任务需要被管理、展示和操作。我们需要一个简单的服务来处理任务的创建、查询、审批操作。# approval_task_service.py from datetime import datetime import uuid from database import db # 假设的数据库连接 def create_approval_task(rule_id: str, context: dict, form_template: dict) - str: 创建审批任务并存入数据库 task_id str(uuid.uuid4()) # 根据模板和上下文渲染审批表单内容 form_data render_form(form_template, context) task { task_id: task_id, rule_id: rule_id, status: pending, # pending, approved, rejected, cancelled context_snapshot: context, # 保存完整的上下文快照 form_data: form_data, created_at: datetime.utcnow(), updated_at: datetime.utcnow(), } db.approval_tasks.insert_one(task) return task_id async def wait_for_approval(task_id: str, timeout: float): 异步等待审批结果模拟生产环境用消息队列更好 start_time asyncio.get_event_loop().time() while (asyncio.get_event_loop().time() - start_time) timeout: task db.approval_tasks.find_one({task_id: task_id}) if task and task[status] in [approved, rejected]: return {status: task[status], reason: task.get(reject_reason), approver: task.get(approver)} await asyncio.sleep(5) # 每5秒检查一次 return None # 超时 # 一个简单的HTTP端点供审批界面调用 from fastapi import APIRouter router APIRouter(prefix/api/approval) router.get(/tasks) def list_pending_tasks(role: str): 列出待当前用户审批的任务 # 根据角色查询相关规则触发的任务 rules [r for r in APPROVAL_RULES if r[approver_role] role] rule_ids [r[id] for r in rules] tasks db.approval_tasks.find({rule_id: {$in: rule_ids}, status: pending}) return list(tasks) router.post(/tasks/{task_id}/approve) def approve_task(task_id: str, approver: str): task db.approval_tasks.find_one_and_update( {task_id: task_id, status: pending}, {$set: {status: approved, approver: approver, updated_at: datetime.utcnow()}}, return_documentTrue ) # 这里应该触发一个事件通知被暂停的Agent流程可以继续了 # 例如向一个消息队列发送 task_id 和 “approved” 状态 notify_resume_event(task_id, approved) return {success: True} router.post(/tasks/{task_id}/reject) def reject_task(task_id: str, approver: str, reason: str): task db.approval_tasks.find_one_and_update( {task_id: task_id, status: pending}, {$set: {status: rejected, approver: approver, reject_reason: reason, updated_at: datetime.utcnow()}}, return_documentTrue ) notify_resume_event(task_id, rejected, reason) return {success: True}对于用户界面可以是一个简单的内部Web页面列出所有待审批任务展示form_data中的详细信息并提供“批准”和“驳回”按钮。对于集成场景也可以将审批任务推送至钉钉、飞书或企业微信的审批流。注意事项审批界面务必清晰展示Agent做出此决策的依据即关键的上下文信息。例如在退款审批中要展示用户的原始投诉内容、订单历史、Agent分析出的退款原因等帮助审批人快速理解情况而不是仅仅看到一个“申请退款500元”的孤零零请求。4. 进阶优化与生产环境考量一个基础的审批门搭建起来后要使其能在生产环境稳定、高效运行还需要考虑以下几个进阶问题。4.1 性能、超时与异步通信同步等待审批结果如上面代码中的wait_for_approval轮询在Web服务中是不可行的它会阻塞请求线程导致资源耗尽。必须采用异步架构。事件驱动当审批任务创建后Agent流程应立即结束当前请求并向用户返回一个中间状态如“您的请求已提交正在等待审批审批通过后将自动处理”。同时将任务ID与会话ID关联存储。消息队列审批完成批准/驳回时审批服务向消息队列如Redis Pub/Sub, RabbitMQ, Kafka发布一个事件包含task_id和结果。Agent恢复有一个独立的“流程恢复服务”订阅该消息队列。当收到事件后根据task_id从数据库取出之前保存的完整上下文快照重新创建或唤醒对应的Agent实例注入审批结果并继续执行后续流程。执行完成后可以通过WebSocket、服务器推送或轮询通知的方式将最终结果告知用户。4.2 审批链、升级与自动审批审批链复杂操作可能需要多级审批。可以在规则中定义approval_flow例如[“team_lead”, “department_head”, “finance”]。系统按顺序创建审批任务只有上一级通过才流转到下一级。审批升级如果一个审批任务长时间未处理如24小时系统可以自动通知上一级管理者或备用审批人。自动审批基于历史决策对于大量重复且模式固定的审批可以引入机器学习模型。系统记录所有历史审批决策特征包括操作类型、参数、上下文、审批结果训练一个预测模型。当新任务触发时如果模型预测“通过”的概率极高如99%且置信度高可以尝试自动批准并记录日志供后续审计。这能极大减轻人工负担但需谨慎设置阈值并保留人工复核入口。4.3 安全、审计与合规性这是企业级应用的生命线。权限隔离审批人只能看到和操作其权限范围内的任务。确保审批接口有严格的身份认证和授权检查。操作不可篡改所有的审批任务、上下文快照、审批操作谁、何时、批准/驳回、理由都必须作为不可变日志持久化存储方便事后审计和追溯。数据脱敏在审批界面展示时对敏感信息如用户身份证号、银行卡号后几位进行脱敏处理保护用户隐私。合规性检查审批规则本身需要定期复审确保符合最新的内部政策和外部法规要求。5. 常见问题排查与实战避坑指南在实际部署和运行审批门系统时我遇到并总结了一些典型问题及其解决方案。5.1 状态恢复失败Agent“失忆”问题现象审批通过后Agent继续执行但好像忘记了之前的对话内容或者执行了错误的操作。排查思路检查上下文快照完整性确保保存的context_snapshot包含了重启Agent所需的全部信息特别是多轮对话的历史消息。检查是否有字段遗漏或被意外截断。检查模型和Prompt一致性恢复执行时使用的LLM模型版本、系统提示词System Prompt必须与暂停前完全一致。任何细微差别都可能导致行为偏离。检查注入信息的方式将审批结果如“已批准”注入上下文的方式要自然。最好是以一条“系统消息”或“用户消息”的形式插入到对话历史中让LLM能正确理解其语义。避坑技巧在开发阶段为每个审批任务保存的上下文快照增加一个“调试副本”并实现一个“重放”功能可以手动用同样的快照触发恢复流程方便复现和调试问题。5.2 审批延迟导致用户体验断裂问题现象用户提交请求后进入审批等待期间无法进行其他操作或者不知道进度如何体验很差。解决方案明确预期管理触发审批时Agent的回复必须清晰告知用户“您的请求涉及XX需要人工审核通常需要1-2个工作日。审核结果将通过站内信/邮件通知您。” 避免让用户傻等。提供任务查询入口给用户一个查询审批进度的功能如输入“查询我的退款审批状态”。支持后续补充交互设计系统时应考虑用户在等待审批期间是否可以就同一件事进行补充说明或咨询。这需要系统能将新的对话内容关联到已暂停的审批任务上下文中。5.3 规则冲突与循环触发问题现象定义了多条规则它们可能同时被触发或者规则A触发审批审批通过后执行的操作又触发了规则B陷入死循环。解决策略规则优先级与互斥为规则定义优先级。当多个规则被触发时只执行优先级最高的那个。或者定义规则组组内规则互斥。审批后操作白名单在审批通过后、继续执行原始操作前可以设置一个“安全区”允许此次执行绕过某些规则检查避免循环。例如在上下文快照中设置一个_approved_operation标志。规则模拟测试在上线新规则前用历史日志数据进行模拟测试检查是否存在冲突或循环触发的可能。5.4 审批界面信息过载或不足问题审批人面对一个任务要么信息太多看不懂重点要么信息太少无法决策。优化方案结构化展示利用好approval_form_template将信息分组展示如“用户信息”、“订单信息”、“Agent分析”、“申请操作详情”。关键信息高亮自动提取并高亮金额、风险关键词、政策条款冲突点等。提供一键查询入口在审批界面嵌入链接或按钮让审批人能一键跳转到相关的用户详情页、订单系统等获取更全面的背景信息。总结与建议让Agent在触发审批时不仅提交原始数据也提交一个简短的“审批建议摘要”说明它为什么建议执行此操作以及潜在风险点。这能极大提升审批效率。为Agent添加审批门本质上是在“自动化”与“可控性”之间寻找最佳平衡点。它不是一个限制创新的枷锁而是一套让AI应用能更安全、更可靠地服务于复杂现实世界的保障系统。从简单的规则拦截到智能的风险预测从单点审批到复杂的流程编排这个系统的深度和广度可以根据业务需求不断扩展。
