“AI learned to act. I built the gate that makes it prove it should”,这句话翻译过来就是:AI 已经学会了行动,而我造了一道闸门,让它在真正动手之前,先证明自己“应该”这么做。
这次我们不聊生图模型,也不聊推理加速,聊一个更偏 AI 工程治理的话题:当 Agent 开始自主调用工具、读写文件、发请求、跑代码时,你靠什么拦住那些不该做的动作?很多团队现在已经在跑多 Agent 系统,模型本身也越来越会“用工具”,但真正到了生产环境,问题往往不是模型不够聪明,而是它太聪明了——你根本不知道它下一步会调用什么,也不知道这个动作是否越权、是否合理、是否可审计。
“AI 行动门控”(Action Gate)要解决的正是这件事。它的核心不是限制模型能力,而是在“模型生成动作意图”和“动作真正执行”之间插入一个证明层:动作必须先通过规则校验、上下文校验和权限校验,生成可解释的证明结果,才允许被执行。这篇文章会完整拆解这套设计思路:门控放在哪里、规则怎么定、验证器怎么写、接口怎么暴露、批量任务和审批队列怎么接,以及最容易踩的坑在哪里。
如果你正在做 AI Agent、工具调用、自动化工作流,或者只是想知道怎么让 AI 在关键操作上“先证明再行动”,这篇文章可以直接收藏。
1. AI 行动门控核心能力速览
| 能力项 | 说明 |
|---|---|
| 项目定位 | 面向 AI Agent 的“动作执行前”的治理/门控/审验机制 |
| 核心思路 | 模型生成动作意图后,先经过规则校验、上下文校验、权限校验,再决定是否放行 |
| 主要功能 | 动作规则校验、权限判定、上下文合理性证明、审批队列、审计日志、动作上限控制 |
| 适用对象 | 自主调用工具的 Agent、多 Agent 协作、RPA 自动化、企业内 AI 工作流 |
| 支持平台 | 与模型和框架无关,可接入 OpenAI Function Calling、LangChain、自研 Agent 等 |
| 启动方式 | 独立 API 服务 / 嵌入 Agent 进程 / 网关中间件 |
| 是否支持 API | 支持,核心是一个可以同步或异步调用的“动作审验接口” |
| 是否支持批量任务 | 支持,通过审批队列和批量任务状态机管理 |
| 显存占用 | 不涉及模型推理,纯规则/校验逻辑,CPU 与内存为主 |
| 部署复杂度 | 中等,依赖策略配置、验证器、存储和日志 |
| 适合场景 | 需要给 AI 动作加权限边界、合规审验、可审计记录的生产环境 |
这个表里最有价值的信息是:Action Gate 不是一个重模型,它不跑大模型推理,不占显存,它的成本主要在策略配置、验证器逻辑和队列管理上。也就是说,哪怕你的 Agent 跑在消费级显卡或者纯 API 模型上,门控层也可以独立部署,几乎不增加推理负担。
很多人听到“给 AI 加门控”会觉得是给 AI 套枷锁,实际上恰好相反。门控做得好的系统,反而可以给 Agent 开放更多执行权限,因为每个动作都有规则兜底和审计记录。没有门控,你就只能依靠提示词约束“你小心一点”,这在生产环境几乎等于裸奔。
2. 适用场景与使用边界
2.1 这类门控适合谁
- 在跑多 Agent 系统的团队:多个 Agent 之间互相调用工具、读写共享资源,最容易出现动作冲突和越权。门控可以作为统一出口,所有 Agent 的动作都从同一个闸口过。
- 做企业 AI 工作流的开发者:Agent 需要连接数据库、发送邮件、创建工单、调用内部 API,这些动作直接影响业务数据,必须做动作前审验。
- 做 RPA 和自动化脚本的人:AI 自动操作浏览器、桌面软件、命令行时,门控能在危险操作(删除文件、转账、批量发消息)之前拦住。
- 研究 AI Agent 安全与可观测性的同学:门控天然产生结构化日志,每个被拦截或放行的动作都有理由和证据,非常适合做安全分析和效果复盘。
2.2 不适合什么场景
- 单轮问答、纯文本生成、没有外部动作的聊天机器人,不需要门控,加了反而增加延迟。
- 需要极限低延迟的推理场景,比如实时语音对话,门控如果设计得太重会拖慢响应。
- 纯研究 demo,只在本地跑一次性的提示词实验,暂时不用上完整门控。
2.3 使用边界与合规提醒
这里必须说清楚:门控不是“让 AI 变安全”的银弹,它只是执行前的最后一道闸口,不能替代数据安全、隐私保护和人工审批制度。
如果要让 Agent 访问用户数据、操作企业系统、调用第三方服务,需要先获得合法授权;涉及人脸、声音、版权素材、个人信息等场景,必须严格遵循相关法规和平台条款。门控规则应该由业务方、安全方和运维方一起评审,而不是由写 Agent 的工程师一个人说了算。
凡是 Agent 主动发起的写操作、转账操作、外发消息操作、删除操作,建议在门控里做成“高风险动作”,默认走人工审批,而不是让 AI 自己证明一下就放行。门控能降低风险,但不能消灭责任。
3. 整体架构:在哪个环节“证明应该做”
3.1 三段式动作流水线
Action Gate 的典型位置在 Agent 的“动作意图生成”和“动作执行器”之间。完整流水线如下:
- Plan:模型根据用户任务生成动作计划,比如“先查询用户订单,再调用退款接口”。
- Prove:动作计划中的每一个动作进入 Gate,门控基于规则和上下文给出“允许执行 / 拒绝执行 / 需要审批”的判定。
- Act:通过判定后,动作才交给执行器真正执行。
这里最容易被忽视的是 Prove 阶段的产物,我建议把它设计成一条可解释的证明链:为什么允许、命中哪条规则、上下文证据是什么、执行范围是什么。这样出了问题,可以回溯到某个具体判断依据,而不是模型的一句“我觉得可以做”。
3.2 门控的四个核心模块
- 规则引擎:负责加载策略配置,匹配动作类型、目标对象、参数范围。规则配置应该使用声明式格式,比如 YAML 或 JSON,方便非开发人员参与维护。
- 上下文校验器:结合当前会话、任务目标、用户身份、历史动作,判断这个动作在上下文里是否合理。比如用户只问了“查询”,Agent 却要调用“删除”,上下文校验就应该直接拦截。
- 权限与审批模块:判断执行主体是否有权做这个动作;对高风险动作,进入审批队列,等待人工或上级系统确认。
- 审计日志模块:记录每次动作判定,包括模型生成的动作原始内容、门控判定结果、命中规则、审批人、执行结果。审计日志是门控存在的最大价值之一。
从实现上看,这四块可以全部放在一个服务里,也可以拆成多个微服务。我建议先做单体服务,把规则引擎和审计日志做扎实,再考虑拆分。
用户请求 -> Agent(LLM) -> 动作列表 -> Action Gate -> 允许/拒绝/审批 -> 执行器 | +-> 规则校验 + 权限校验 + 审计日志4. 环境准备与前置条件
Action Gate 不绑定特定编程语言。下面这组清单是通用前置条件,具体版本以实际技术栈为准。
4.1 需要准备的基础组件
- 操作系统:Linux 或 macOS 比较推荐,Windows 也能跑,但 shell 命令和进程管理需要额外适配。
- 运行时:如果使用 Python,建议 Python 3.10 以上;使用 Node.js 建议 18 以上;具体以项目依赖为准。
- 配置管理:至少有一个 YAML 或 JSON 的规则配置文件。
- 存储:审计日志建议写本地文件或专门的日志系统;审批队列需要一个支持原子状态变更的存储,比如 Redis、PostgreSQL 或 SQLite。
- API 框架:如果独立部署,FastAPI、Flask、Express 都可以,本文示例用 FastAPI。
- 依赖管理工具:Python 用 pip 或 poetry,Node 用 npm 或 yarn。
4.2 功能模块与依赖建议
| 模块 | 建议依赖 | 说明 |
|---|---|---|
| 规则解析 | PyYAML / jsonschema | 解析规则文件并做格式校验 |
| API 服务 | FastAPI + uvicorn | 提供动作审验接口 |
| 审批队列 | Redis / PostgreSQL / SQLite | 保存待审批任务与状态 |
| 审计日志 | 标准 logging / 文件输出 / Elasticsearch | 记录每次判定过程 |
| 测试 | pytest | 维护验证器单元测试 |
如果只是本地验证流程,SQLite + 标准 logging 就够了。生产环境至少要把审计日志落到独立存储里,最好是只追加、不可篡改的日志方案。
4.3 端口和进程规划
门控服务如果独立部署,建议固定一个内网端口,只对 Agent 服务开放,不要直接暴露到公网。端口冲突时,可以在启动参数里指定地址和端口,例如 127.0.0.1:9000 或 0.0.0.0:9000,按实际访问方调整。
5. 门控规则定义与验证器实现
这是整套设计的核心。规则定义的质量,直接决定门控能不能拦住问题,以及误伤率高不高。
5.1 声明式动作规则
用 YAML 定义规则,优点是易读、易改、易评审。下面是一个示例规则文件,覆盖了常见的动作类型。
version: "1.0" actions: - name: "read_file" risk: "low" allowed_roles: ["user", "assistant"] allowed_paths: - "/data/project/workspace/**" deny_paths: - "/data/project/workspace/secrets/**" require_approval: false - name: "write_file" risk: "medium" allowed_roles: ["assistant"] allowed_paths: - "/data/project/workspace/output/**" deny_paths: - "/data/project/workspace/secrets/**" - "/etc/**" require_approval: false - name: "delete_file" risk: "high" allowed_roles: [] allowed_paths: [] require_approval: true - name: "call_external_api" risk: "medium" allowed_roles: ["assistant"] allowed_domains: - "api.example.com" deny_domains: - "*.secret-example.com" require_approval: false max_calls_per_minute: 10 - name: "send_message" risk: "high" allowed_roles: ["assistant"] max_receivers: 20 require_approval: true规则的字段可以按需要扩展。关键的几个要点:
allowed_roles定义哪些角色能执行这个动作,user表示用户明确要求,assistant表示 Agent 自主发起。如果允许集合为空,表示任何角色都不能直接执行。allowed_paths和deny_paths用于文件类操作的路径白名单和黑名单,路径匹配建议使用 glob 或正则,这里用 glob 写法。require_approval表示是否需要人工审批。删除、发送消息、转账这类动作默认 true。max_calls_per_minute是简单限流,防止 Agent 在循环中疯狂调用同一个接口。
5.2 验证器实现
验证器把动作描述和规则进行比对,输出校验结果。下面的代码是 Python 版本的核心验证器,使用pathlib.PurePath做路径匹配、fnmatch做 glob 匹配。
import fnmatch import re from dataclasses import dataclass, field from typing import Any, Dict, List, Optional @dataclass class ActionContext: action_name: str params: Dict[str, Any] role: str session_id: str user_id: str history: List[Dict[str, Any]] = field(default_factory=list) @dataclass class GateResult: allowed: bool reason: str matched_rule: Optional[Dict[str, Any]] = None require_approval: bool = False evidence: Dict[str, Any] = field(default_factory=dict) class ActionGateValidator: def __init__(self, rules: Dict[str, Any]): self.rules = rules self._action_rules = {a["name"]: a for a in rules["actions"]} def validate(self, ctx: ActionContext) -> GateResult: rule = self._action_rules.get(ctx.action_name) if not rule: return GateResult( allowed=False, reason="unknown_action", evidence={"action": ctx.action_name}, ) if ctx.role not in rule.get("allowed_roles", []): return GateResult( allowed=False, reason="role_not_allowed", matched_rule=rule, evidence={"role": ctx.role, "action": ctx.action_name}, ) if rule.get("require_approval") is True: return GateResult( allowed=False, reason="requires_approval", matched_rule=rule, require_approval=True, evidence={"action": ctx.action_name}, ) # 路径类规则校验 if "allowed_paths" in rule or "deny_paths" in rule: path_result = self._validate_path(rule, ctx.params) if not path_result.allowed: return path_result # 域名类规则校验 if "allowed_domains" in rule or "deny_domains" in rule: domain_result = self._validate_domain(rule, ctx.params) if not domain_result.allowed: return domain_result # 限流类规则校验 if "max_calls_per_minute" in rule: limit_result = self._validate_rate_limit(rule, ctx) if not limit_result.allowed: return limit_result return GateResult( allowed=True, reason="allowed_by_rule", matched_rule=rule, evidence={"action": ctx.action_name, "params": ctx.params}, ) def _validate_path(self, rule: Dict[str, Any], params: Dict[str, Any]) -> GateResult: target_path = params.get("path", "") path_obj = PurePath(target_path) for deny_pattern in rule.get("deny_paths", []): if fnmatch.fnmatch(str(path_obj), deny_pattern): return GateResult( allowed=False, reason="path_denied", matched_rule=rule, evidence={"path": target_path, "pattern": deny_pattern}, ) allowed = False for allow_pattern in rule.get("allowed_paths", []): if fnmatch.fnmatch(str(path_obj), allow_pattern): allowed = True break if not allowed: return GateResult( allowed=False, reason="path_not_allowed", matched_rule=rule, evidence={"path": target_path, "patterns": rule.get("allowed_paths")}, ) return GateResult(allowed=True, reason="allowed_by_rule", matched_rule=rule) def _validate_domain(self, rule: Dict[str, Any], params: Dict[str, Any]) -> GateResult: url = params.get("url", "") domain = re.sub(r"^https?://", "", url).split("/")[0] for deny_pattern in rule.get("deny_domains", []): if fnmatch.fnmatch(domain, deny_pattern): return GateResult( allowed=False, reason="domain_denied", matched_rule=rule, evidence={"domain": domain, "pattern": deny_pattern}, ) if rule.get("allowed_domains"): allowed = any( fnmatch.fnmatch(domain, allow_pattern) for allow_pattern in rule["allowed_domains"] ) if not allowed: return GateResult( allowed=False, reason="domain_not_allowed", matched_rule=rule, evidence={"domain": domain, "patterns": rule["allowed_domains"]}, ) return GateResult(allowed=True, reason="allowed_by_rule", matched_rule=rule) def _validate_rate_limit(self, rule: Dict[str, Any], ctx: ActionContext) -> GateResult: # 实际场景应使用 Redis 或内存计数器,这里保留占位 return GateResult(allowed=True, reason="rate_limit_placeholder", matched_rule=rule)这段代码的关键是:每个校验分支失败都返回明确的reason,并且把命中的规则、参数、路径、域名写入evidence。这样接审计日志时,可以直接把GateResult序列化保存,不需要额外拼字段。
5.3 上下文合理性校验
规则校验只能防“明显不合规”,但防不住“在上下文里不合理”。举个例子:用户问“今天天气怎么样”,Agent 生成的动作却是“读取用户私密文件”。这时即使文件路径在允许列表里,上下文校验也应该拦截。
上下文校验的简单实现思路:在验证器之外再加一层ContextJudge,传入会话历史、用户原始输入和动作列表,判断动作是否与当前任务相关。这一步可以用轻量规则,也可以调用小模型做意图相关性判断。
class ContextJudge: def __init__(self, relevant_keywords_by_action: Dict[str, List[str]]): self.relevant_keywords_by_action = relevant_keywords_by_action def is_relevant(self, user_input: str, ctx: ActionContext) -> bool: keywords = self.relevant_keywords_by_action.get(ctx.action_name, []) user_input_lower = user_input.lower() return any(kw in user_input_lower for kw in keywords)当然,纯关键词匹配比较粗糙,生产环境可以换成向量相似度或小模型分类。但这里有一个原则:上下文校验可以保守,不能激进。宁可多拦截一次,也不要放行一个明显无关的高危动作。
6. 把门控接入 Agent:接口 API 与调用示例
门控做成本地函数可以直接嵌入 Agent,但更推荐做成独立 API 服务。独立服务的好处是规则更新、审计日志、审批队列可以集中管理,多个 Agent 共用同一个门控。
6.1 用 FastAPI 暴露动作审验接口
下面是一个最小可运行的服务示例。它加载规则文件,接收 Agent 提交的动作列表,逐条校验,返回批量判定结果。
import yaml from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from typing import List, Dict, Any from gate_validator import ActionGateValidator, ActionContext, GateResult app = FastAPI(title="AI Action Gate Service") with open("action_rules.yaml", "r", encoding="utf-8") as f: rules = yaml.safe_load(f) validator = ActionGateValidator(rules) class ActionRequest(BaseModel): session_id: str user_id: str role: str = "assistant" user_input: str = "" actions: List[Dict[str, Any]] = Field(..., description="模型生成的动作列表") class ActionResponse(BaseModel): results: List[Dict[str, Any]] @app.post("/api/v1/gate/check", response_model=ActionResponse) def check_actions(req: ActionRequest): results = [] for action in req.actions: ctx = ActionContext( action_name=action.get("name", ""), params=action.get("params", {}), role=req.role, session_id=req.session_id, user_id=req.user_id, history=[], ) result: GateResult = validator.validate(ctx) results.append( { "action": ctx.action_name, "allowed": result.allowed, "require_approval": result.require_approval, "reason": result.reason, "evidence": result.evidence, } ) return ActionResponse(results=results) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=9000)这个接口的核心价值是:Agent 不用自己理解规则,它只需要把模型生成的动作列表原样提交给门控,门控返回每个动作的判定。注意role字段,如果是用户明确要求的操作,角色可以传user;如果 Agent 自主发起的操作,角色传assistant。规则可以按角色区别对待。
6.2 用 curl 测试接口
启动服务后,用 curl 提交一组动作,观察判定结果。
curl -X POST http://127.0.0.1:9000/api/v1/gate/check \ -H "Content-Type: application/json" \ -d '{ "session_id": "sess-001", "user_id": "user-001", "role": "assistant", "user_input": "请帮我读取项目输出目录下的报告文件", "actions": [ { "name": "read_file", "params": { "path": "/data/project/workspace/output/report.md" } }, { "name": "send_message", "params": { "receivers": ["user-002"], "content": "我准备删除一个文件" } } ] }'如果规则文件里send_message配置为高风险且需要审批,返回结果中这个动作的allowed应该是 false,require_approval是 true。这正好对应“AI 应该先证明自己可以做,再真正执行”的设计目标。
6.3 在 Agent 调用侧封装 SDK
为了不让 Agent 代码写得太碎,可以在调用侧封装一个AgentGateClient。它负责调用门控接口、过滤被拒绝的动作、拦截需要审批的动作。
import requests from typing import List, Dict, Any class AgentGateClient: def __init__(self, gate_url: str = "http://127.0.0.1:9000"): self.gate_url = gate_url def check_actions( self, session_id: str, user_id: str, role: str, user_input: str, actions: List[Dict[str, Any]], ) -> Dict[str, Any]: payload = { "session_id": session_id, "user_id": user_id, "role": role, "user_input": user_input, "actions": actions, } resp = requests.post( f"{self.gate_url}/api/v1/gate/check", json=payload, timeout=10, ) resp.raise_for_status() return resp.json() def filter_allowed( self, session_id: str, user_id: str, role: str, user_input: str, actions: List[Dict[str, Any]], ) -> List[Dict[str, Any]]: data = self.check_actions(session_id, user_id, role, user_input, actions) allowed_actions = [] for result, raw_action in zip(data["results"], actions): if result["allowed"] and not result["require_approval"]: allowed_actions.append(raw_action) else: print( f"blocked action: {result['action']}, " f"reason={result['reason']}, require_approval={result['require_approval']}" ) return allowed_actions这样 Agent 主流程里只需要调filter_allowed,把放行后的动作交给执行器。被拦截的动作记日志,需要审批的动作进入审批队列。
7. 批量任务与审批队列设计
多 Agent 场景下,门控是高频调用点。如果 Agent 在几秒钟内生成了上百个动作,门控服务需要支持批量校验和审批流的进队/出队。这里给出一个不依赖具体中间件设计的审批队列状态机。
7.1 动作审批状态机
一个动作从门控校验到最终执行,至少要经过这几个状态:
PENDING_APPROVAL->APPROVED/REJECTED->EXECUTING->SUCCEEDED/FAILED
如果是低风险动作,不需要审批,可以直接从PENDING进入EXECUTING。如果审批超时未处理,建议进入TIMEOUT状态,默认拒绝执行。
PENDING -> [需要审批] -> PENDING_APPROVAL -> APPROVED -> EXECUTING -> SUCCEEDED REJECTED -> EXECUTING -> FAILED PENDING -> [无需审批] -> EXECUTING -> SUCCEEDED / FAILED7.2 Redis 队列实现示例
approval_queue.py 给出一个简化实现,使用 Redis List 存储待审批动作,使用 Hash 保存状态。
import json import time import redis r = redis.Redis(host="127.0.0.1", port=6379, db=0) APPROVAL_QUEUE_KEY = "gate:approval_queue" APPROVAL_STATE_KEY = "gate:approval_state" def enqueue_approval(action_id: str, action_payload: dict, ttl_seconds: int = 300): item = { "action_id": action_id, "payload": action_payload, "created_at": time.time(), "ttl": ttl_seconds, } r.rpush(APPROVAL_QUEUE_KEY, json.dumps(item)) r.hset( APPROVAL_STATE_KEY, action_id, json.dumps( { "status": "PENDING_APPROVAL", "created_at": time.time(), "updated_at": time.time(), } ), ) def approve_action(action_id: str, approver: str): state = json.loads(r.hget(APPROVAL_STATE_KEY, action_id)) if state["status"] != "PENDING_APPROVAL": return False state["status"] = "APPROVED" state["approver"] = approver state["updated_at"] = time.time() r.hset(APPROVAL_STATE_KEY, action_id, json.dumps(state)) return True def reject_action(action_id: str, approver: str, reason: str): state = json.loads(r.hget(APPROVAL_STATE_KEY, action_id)) if state["status"] != "PENDING_APPROVAL": return False state["status"] = "REJECTED" state["approver"] = approver state["reason"] = reason state["updated_at"] = time.time() r.hset(APPROVAL_STATE_KEY, action_id, json.dumps(state)) return True这种设计在并发上有一个问题:多个审批人同时操作同一个 action 会覆盖状态。生产环境建议改为 Lua 脚本或数据库行锁,确保状态流转的原子性。这里只是演示状态流转思路。
7.3 批量任务处理建议
- 批量校验接口应该支持一次提交多个动作,减少 Agent 与门控之间的 HTTP 往返。
- 批量任务里的每个动作都是独立判定,一个动作被拒绝不应影响其他动作执行。
- 对高风险批量操作,比如“给 100 个用户发消息”,建议把整个批次作为一个审批单元,而不是逐个审批。
- 批量任务的每个动作都要在日志中保留原始参数,否则出了事只能看到“该动作已执行”,看不到当时传了什么参数。
8. 资源占用与性能观察
Action Gate 不跑大模型,所以资源占用不是显存,而是 CPU、内存和存储。但从生产角度看,它依然是关键链路的一部分,性能不可忽略。
8.1 性能影响点
- 规则加载与解析:每次服务启动时加载一次规则文件即可,不要在每次请求里重新解析 YAML。
- 路径和域名匹配:如果规则很多、路径 glob 很复杂,匹配耗时会上升。建议先过滤 action 类型,再做细粒度匹配。
- 审批队列状态写入:审批操作涉及 Redis 或数据库读写,要关注连接池配置和超时。
- 审计日志写入:每次门控判定都写日志,高频调用下日志量会很大,建议走异步日志或消息队列,不要阻塞请求主流程。
8.2 如何观察性能
- 给门控服务加 Prometheus 指标:
gate_check_total、gate_check_duration_seconds、gate_blocked_total、gate_approval_pending_total。 - 在测试环境压一批动作,观察 P99 延迟。如果 P99 高于 200ms,先查规则匹配和日志写入。
- 关注被拦截动作的比例。如果拦截率过高,说明规则太严或 Agent 生成的动作质量问题很大;如果拦截率几乎为 0,说明门控可能形同虚设,需要检查规则是否覆盖了真实风险动作。
8.3 降低开销的手段
- 缓存解析后的规则对象,避免每次请求重复解析。
- 对路径规则做前缀索引,减少 glob 全量匹配。
- 日志批量写入,例如攒 100 条再 flush 一次。
- 审批队列和审计日志分离部署,避免互相影响。
9. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 接口返回 422 参数校验失败 | 请求体字段名或类型不匹配 | 查看 FastAPI 返回的错误详情 | 按 Pydantic 模型修正请求体字段 |
| 所有动作都被拦截 | 规则配置里 allowed_roles 为空或角色不匹配 | 检查规则文件和传入的 role 字段 | 调整规则角色或请求中的角色 |
| 规则修改后不生效 | 服务启动时缓存了旧规则 | 检查代码是否每次请求都重新加载规则 | 在更新规则文件后重启服务或实现热加载 |
| 审批队列中动作一直处于 PENDING | 审批端没有正确消费队列 | 检查队列消费者日志和 Redis 连接 | 补充审批消费逻辑或手动触发审批 |
| 审计日志丢失 | 日志是异步写入但进程异常退出 | 查看进程退出前日志缓冲 | 改用文件回滚或消息队列,保证落盘 |
| 路径匹配误判 | Windows 路径与 Linux 路径分隔符不一致 | 检查使用 pathlib 还是字符串拼接 | 统一使用 pathlib.PurePath 处理 |
| 门控服务请求超时 | 日志写入阻塞或审批队列阻塞 | 查看服务调用链和数据库慢查询 | 异步化日志写入,优化队列连接池 |
| 高并发下审批状态覆盖 | Redis 状态更新不是原子操作 | 看是否多个审批人同时操作同一 action | 使用 Lua 脚本或数据库行锁 |
10. 最佳实践与使用建议
10.1 规则设计原则
- 默认拒绝,优先放行。凡是没出现在规则里的动作,直接拒绝。
- 高风险动作默认走人工审批,能不能批量执行由业务方决定。
- 规则文件用版本号管理,每次变更要有 changelog。
- 路径、域名、角色这类基础规则要写单元测试,避免改一个 glob 规则把正常动作全部拦截。
10.2 接入流程建议
第一次接入不要追求全量覆盖。按这个顺序推进:
- 先接只读动作,比如查询、读取文件、搜索,跑通门控主流程。
- 再加写操作,比如写文件、改配置,观察误拦截率。
- 最后接高风险动作,比如删除、发送消息、转账,默认走审批。
- 每次调整规则后,回放历史动作日志,确认拦截决策是否符合预期。
10.3 不要忽视审计日志
审计日志的字段越完整,排障越省力。建议至少包含这些字段:
{ "timestamp": "2025-06-01T12:00:00Z", "session_id": "sess-001", "user_id": "user-001", "agent_name": "order-agent", "action_name": "call_external_api", "action_params": { "url": "https://api.example.com/create-order" }, "gate_reason": "allowed_by_rule", "matched_rule_version": "1.0", "approval_status": "NOT_REQUIRED", "execution_result": "SUCCEEDED" }这个 JSON 结构在接入 ELK 或其他日志平台时会非常方便,可以直接按 action_name 聚合、按放行原因过滤。
10.4 安全与合规提醒
- 门控服务不应直接暴露到公网,必须放在内网,并对调用方做鉴权。
- 涉及用户数据、个人隐私、版权内容的动作,必须结合业务方审批流程,不能把审批权完全交给 AI。
- 在真实环境开放前,先在测试环境用脱敏数据走完整流程。
- 定期审查门控规则和审批日志,及时清理过宽或过严的规则。
11. 总结与下一步
这个项目标题最有价值的地方,在于它把 AI 工程问题从“怎么让模型更聪明”拉回到了“怎么让动作更可控”。无论模型能力多强,只要它在真实环境里执行动作,就需要一扇既能放行、也能拦截、还得留下记录的门。Action Gate 的设计并不复杂:模型生成动作列表,门控逐条校验,输出允许、拒绝或审批;规则文件决定边界,验证器执行判定,审计日志记录全过程。
最值得先验证的功能,是让它拦下一个明显不该执行的高风险动作。比如让 Agent 假装生成一个“删除目录”的动作,配置里删除默认需要审批,看门控是否正确进入审批队列。这一步跑通了,整套链路的核心就成立了。
最容易踩的坑有三个:规则写得过松,等于没拦;角色和路径匹配没做单元测试,上线后产生大量误拦截;审批队列没有原子状态流转,多人审批时互相覆盖。建议第一次接入先用只读动作跑小流量,确认门控判定和审计日志都正常后,再逐步放开写操作和高风险操作。
后续可以扩展的方向包括:把上下文校验从关键词升级为模型相关性判断;给门控接入 Prometheus 指标和告警;在审计日志之上做动作异常检测;让审批队列支持多级审批人和超时自动拒绝。如果你已经在做多 Agent 系统,下一步最值得做的,就是把门控接到真实业务动作上,先从小流量、只读操作开始验证,跑通后再放开更多执行权限。