尧图网页设计 
当 AI Agent 从聊天窗口走向生产环境核心变化不是模型更强了而是它开始真正改变外部状态发邮件、改订单、执行命令、写数据库。模型输出只是文本这些动作却会产生真实后果。于是工程问题变成了如何让 AI 在行动之前先证明这次行动是应该执行的。这正是“AI learned to act. I built the gate that makes it prove it should”这句话背后的技术意图Agent 学会了行动那网关要做的就是要求它提交理由、证据、权限上下文然后决定放行还是拒绝。本文实现一个可运行的 AI Action Gate用 FastAPI 搭建检查服务在 Agent 的工具调用链路上加入五层校验并通过实例演示允许、拒绝和异常处理。1. 为什么 AI Agent 需要一道门而不是一套提示词1.1 行动能力的代价从生成文本到执行操作早期的 LLM 应用主要是“生成内容”模型输出一段文字用户自己判断是否采用。这个阶段即使模型产生错误影响也局限在文本层面用户有最后决定权。AI Agent 出现后模型输出不再只是文字而是变成结构化的工具调用例如send_email、delete_file、create_order。这些调用一旦执行就会改变系统状态而且往往是不可逆的。以一个简单的客服 Agent 为例。用户说“帮我给客户发送一份包含对账单的邮件”Agent 需要调用邮件工具。如果只是在提示词里写“发送前要检查收件人”遇到模型幻觉、参数拼写错误、上下文拼接错误仍然可能把邮件发给错误的人。更危险的是如果 Agent 同时被注入了恶意指令提示词约束很难挡住所有异常路径。Gate 的存在就是把这些“可能出错但又不能完全靠模型自觉”的动作放到一个显式的、可编程的检查点上。它不是用来替代 Agent 的能力而是确保 Agent 的每一次动作在进入执行器之前都经过规则、权限和语义层面的验证。1.2 提示词约束为什么不可靠很多人会问既然模型已经理解了规则为什么不直接在 system prompt 里写“未经验证不得执行外部调用”原因有四点。第一LLM 的输出是概率性的同样的 prompt 在不同输入下可能产生不同行为。第二工具调用参数来自模型生成模型可能把to和cc字段填反或者忽略必填参数。第三提示词注入会改变模型对规则的判断攻击者可以把恶意内容藏在用户输入里让 Agent 误以为某个操作是合法的。第四提示词本身不产生审计记录如果操作出了问题事后很难回答“当时模型为什么这样调用”。所以提示词约束适合做“行为引导”不适合做“安全边界”。安全边界必须由代码强制实现。Gate 就是代码层的一道边界它要求每个工具调用都提交一个标准请求包括tool_name、params、reason、evidence等字段然后由独立的检查链决定是否允许执行。1.3 Gate 的本质把“证明”变成强制检查点“让 AI 证明它应该行动”这句话很容易被理解成一种理念让模型解释自己的行为。但在工程实现里它必须变成协议和代码。Gate 不关心模型内心怎么想只关心这次请求是否满足预先定义的条件。一次完整的请求至少要回答四个问题动作是什么调用哪个工具参数是什么。谁在发起哪个 Agent、哪个用户。为什么发起理由是什么有没有用户指令或任务上下文作为依据。证据在哪里有没有工单编号、授权记录、会话引用。Gate 根据这些问题执行检查。规则层负责判断“能不能做”语义层负责判断“应不应该做”。两层都通过才允许执行。这个设计让 AI 的行动从“模型说了算”变成“策略说了算”也让它留下的每一步都具备可追溯性。2. 门禁系统设计一次工具调用五层检查2.1 核心数据流Action Gate 本质上是一个独立的服务或模块部署在 Agent 的执行器之前。Agent 生成工具调用后不能直接调用执行函数而是先把请求发给 GateGate 返回allow或deny只有allow才能进入执行器。数据流如下用户 - Agent | v Agent 工具调用请求 (ToolCallRequest) | v Action Gate 服务 ├─ 1. Schema 校验 ├─ 2. 权限校验 ├─ 3. 频率/预算校验 ├─ 4. 关键词规则校验 └─ 5. LLM 语义审查 | v 允许 - 执行器执行 拒绝 - 返回原因并记录日志这个链路的关键点在于“不可跳过”。Agent 内部可以有自己的规划逻辑但最终的工具调用必须统一经过 Gate。如果某个 Agent 绕过 Gate 直接调用执行器Gate 就无法发挥作用。2.2 五层检查Schema、权限、预算、语义、审计这里设计五层检查每一层解决一类问题。前四层是规则校验成本低、速度快第五层是语义校验成本高、判断更灵活。检查层作用通过条件失败结果Schema 校验检查参数是否完整、格式是否正确必填字段存在参数结构符合工具定义不进入执行器返回明确原因权限校验检查 Agent / 用户是否有调用该工具的权限Agent 在允许列表中工具未被禁用拒绝并标记越权行为频率 / 预算校验防止无限循环和成本失控单位时间调用数未超过阈值成本未超预算拒绝并提示稍后重试关键词规则校验拦截危险参数和敏感信息参数中不包含 deny_keywords拒绝并记录触发关键词LLM 语义审查判断请求理由与用户意图是否一致模型认为理由合理且置信度达标拒绝保留人工复核入口Schema 和权限通常由配置文件维护适合静态场景。频率和预算需要动态计数适合在线场景。关键词规则可以快速拦截“明显不该出现的内容”LLM 语义审查则处理规则无法覆盖的模糊场景。2.3 为什么需要请求-凭据协议Gate 要判断“证明是否成立”前提是请求里必须有可验证的信息。如果请求只包含tool_name和paramsGate 只能做参数和权限校验无法判断这次行动是否真的有用户意图支撑。因此需要设计一个标准的请求模型。下面是一个请求示例{ request_id: req_20250101_001, agent_id: customer-support-agent, user_id: user_42, tool_name: send_email, params: { to: opsexample.com, subject: 部署完成通知, body: 生产环境 v2.3 部署完成 }, reason: 用户要求部署完成后发送通知, evidence: ticket#1234, task_iddeploy_20250101 }reason是模型给出的行动理由evidence是支撑这个理由的上下文信息。语义审查层可以根据这两个字段判断请求是否合理。比如用户只要求“查询订单”但 Agent 却尝试调用“删除订单”这就是明显的意图偏差。请求模型定了Gate 的实现边界也就清晰了。下面进入环境准备和代码实现。3. 环境准备与项目结构3.1 技术选型与依赖示例使用 Python 3.10 及以上版本Web 框架使用 FastAPI数据校验使用 Pydantic v2。之所以选 FastAPI是因为它天然支持 Pydantic 模型适合快速搭建 JSON API也方便后续接入中间件、限流和日志。需要安装的依赖如下。建议先创建虚拟环境避免污染系统 Pythonpython -m venv .venv source .venv/bin/activate pip install fastapi uvicorn pydantic PyYAML httpx pytest各依赖的用途如下依赖用途fastapi提供 Web API 框架uvicorn本地启动 ASGI 服务pydantic定义请求和响应模型PyYAML读取策略配置文件httpx调用外部 LLM 接口pytest编写自动化验证用例3.2 项目目录结构项目可以按模块拆分而不是把所有逻辑堆在单个文件里。这里给出一个便于扩展的目录结构action_gate/ ├── main.py ├── models.py ├── validators.py ├── llm_reviewer.py ├── agent_loop.py ├── config.yaml └── tests/ └── test_gate.pymodels.py定义请求、检查结果、最终决策的数据结构。validators.py实现 ActionGate 的规则检查链。llm_reviewer.py实现基于 LLM 的语义审查。agent_loop.py演示 Agent 如何被打断并强制经过 Gate。config.yaml维护 Agent、工具、策略等配置。tests/test_gate.py写自动化验证。3.3 配置文件Gate 的策略不应该写死在代码里。先准备一个config.yaml后续无论是调整允许工具还是修改限流阈值都不需要重新发版。agents: agent-demo: role: assistant allowed_tools: - send_email - query_order tools: send_email: required_keys: [to, subject, body] params_schema: to: string subject: string body: string deny_keywords: [password, token] cost_weight: 2 query_order: required_keys: [order_id] params_schema: order_id: string deny_keywords: [] cost_weight: 1 policies: max_calls_per_minute: 10 llm_reviewer: enabled: false base_url: https://api.example.com/v1 model: review-model api_key_env: LLM_REVIEW_API_KEY temperature: 0.0 max_tokens: 256 timeout_seconds: 10这里把llm_reviewer.enabled设为false方便先跑通规则链路。如果不配置 API KeyLLM 审查应该降级为“直接放行规则检查结果”而不是让整个服务报错。注意config.yaml里的地址和 Key 只是示例。生产环境不要把密钥直接写在文件里建议通过环境变量或配置中心管理。3.4 学习环境与生产环境差异本地单进程跑一个 FastAPI 服务和在生产环境接入多个 Agent 实例差别非常大。下面列一张对照表帮助定位当前阶段应该关注什么。维度学习环境生产环境限流计数内存字典重启丢失Redis 或独立计数服务策略存储本地 YAML配置中心支持灰度发布LLM 审查可关闭必须开启并设置超时与降级审计日志打印到控制台推送到日志平台保留足够周期高可用单实例多实例部署Gate 做无状态设计人工审批不实现高风险动作接入人工复核队列这些差异决定了代码架构不能把“内存变量”看得太重至少要把数据结构和接口抽象出来后续换成 Redis 时不需要改所有业务代码。4. 最小实现用 FastAPI 搭建 Action Gate4.1 定义请求和响应模型models.py负责定义所有核心数据结构。先定义ToolCallRequest这是 Agent 提交给 Gate 的标准请求再定义CheckResult表示单个检查项的结果最后定义GateDecision表示 Gate 的最终决策。# models.py from typing import Any from pydantic import BaseModel, Field class ToolCallRequest(BaseModel): request_id: str agent_id: str user_id: str tool_name: str params: dict[str, Any] Field(default_factorydict) reason: str evidence: str class CheckResult(BaseModel): name: str passed: bool reason: str metadata: dict[str, Any] Field(default_factorydict) class GateDecision(BaseModel): request_id: str allowed: bool decision: str checks: list[CheckResult] Field(default_factorylist) message: str 这里的reason和evidence是语义审查的核心输入。如果某个 Agent 调用没有填写reasonGate 可以把它视为“行动理由不足”即使参数合法也可以拒绝。4.2 实现配置加载policies.py不需要写复杂逻辑只要把 YAML 文件读取成 Python dict 即可。考虑到后续可能接入配置中心这里单独拆一个模块方便替换实现。# policies.py import yaml from pathlib import Path def load_config(path: str config.yaml) - dict: config_path Path(path) with open(config_path, r, encodingutf-8) as f: return yaml.safe_load(f)如果文件不存在这里会抛出文件异常。实际项目中建议在启动时做一次配置合法性检查比如tools必须存在agents必须存在否则直接拒绝启动。4.3 实现多层校验器validators.py是 Gate 的核心。这里实现一个ActionGate类每个检查方法返回一个CheckResult最终聚合所有检查结果。# validators.py import time import threading from models import ToolCallRequest, CheckResult, GateDecision from llm_reviewer import LLMReviewer class ActionGate: def __init__(self, config: dict, reviewer: LLMReviewer | None None): self.config config self.reviewer reviewer self.tools config.get(tools, {}) self.agents config.get(agents, {}) self.max_per_minute config.get(policies, {}).get(max_calls_per_minute, 10) self.recent_calls: list[float] [] self.lock threading.Lock() def check_tool_call(self, req: ToolCallRequest) - GateDecision: checks [] checks.append(self._schema_check(req)) checks.append(self._permission_check(req)) checks.append(self._rate_check(req)) checks.append(self._keyword_check(req)) rule_passed all(c.passed for c in checks) if rule_passed and self.reviewer is not None: checks.append(self.reviewer.review(req, checks)) allowed all(c.passed for c in checks) message if not allowed: failed [c for c in checks if not c.passed] if failed: message failed[-1].reason return GateDecision( request_idreq.request_id, allowedallowed, decisionallow if allowed else deny, checkschecks, messagemessage, ) def _schema_check(self, req: ToolCallRequest) - CheckResult: tool self.tools.get(req.tool_name) if not tool: return CheckResult( nameschema, passedFalse, reasonftool {req.tool_name} not registered, ) for key in tool.get(required_keys, []): if key not in req.params: return CheckResult( nameschema, passedFalse, reasonfmissing required param {key}, ) return CheckResult(nameschema, passedTrue, reasonparams schema ok) def _permission_check(self, req: ToolCallRequest) - CheckResult: agent self.agents.get(req.agent_id) if not agent: return CheckResult( namepermission, passedFalse, reasonfunknown agent {req.agent_id}, ) allowed_tools agent.get(allowed_tools, []) if req.tool_name not in allowed_tools: return CheckResult( namepermission, passedFalse, reasonfagent {req.agent_id} cannot call {req.tool_name}, ) return CheckResult(namepermission, passedTrue, reasonagent permission ok) def _rate_check(self, req: ToolCallRequest) - CheckResult: now time.time() with self.lock: self.recent_calls [t for t in self.recent_calls if now - t 60] if len(self.recent_calls) self.max_per_minute: return CheckResult( namerate, passedFalse, reasonrate limit exceeded, try later, ) self.recent_calls.append(now) return CheckResult(namerate, passedTrue, reasonrate ok) def _keyword_check(self, req: ToolCallRequest) - CheckResult: tool self.tools.get(req.tool_name) if not tool: return CheckResult( namekeyword, passedFalse, reasontool not registered, cannot check keywords, ) deny_keywords tool.get(deny_keywords, []) text str(req.params).lower() for keyword in deny_keywords: if keyword.lower() in text: return CheckResult( namekeyword, passedFalse, reasonfkeyword blocked: {keyword}, ) return CheckResult(namekeyword, passedTrue, reasonkeyword check ok)代码里值得注意的一点是_rate_check使用了线程锁和内存列表这只适合单进程学习环境。多 worker 部署时每个 worker 有独立内存限流会失效需要把计数放到 Redis。4.4 接入 LLM 语义审查llm_reviewer.py负责语义层面的审查。这里的核心是构造一个审查 prompt让模型输出approved、confidence和explanation。如果未配置 LLM就直接返回通过结果避免阻塞主流程。# llm_reviewer.py import json import os import httpx from models import ToolCallRequest, CheckResult class LLMReviewer: def __init__(self, config: dict): self.config config self.enabled config.get(enabled, False) self.base_url config.get(base_url, ) self.model config.get(model, ) self.api_key os.getenv(config.get(api_key_env, ), ) self.timeout config.get(timeout_seconds, 10) self.max_tokens config.get(max_tokens, 256) self.temperature config.get(temperature, 0.0) def review(self, req: ToolCallRequest, rule_checks: list[CheckResult]) - CheckResult: if not self.enabled: return CheckResult( namellm_review, passedTrue, reasonllm review disabled, ) prompt self._build_prompt(req, rule_checks) try: result self._call_model(prompt) except Exception as exc: return CheckResult( namellm_review, passedFalse, reasonfllm review error: {exc}, ) if result.get(approved): return CheckResult( namellm_review, passedTrue, reasonsemantic review approved, metadata{ confidence: result.get(confidence, 0.0), explanation: result.get(explanation, ), }, ) return CheckResult( namellm_review, passedFalse, reasonresult.get(explanation, semantic review denied), metadata{ confidence: result.get(confidence, 0.0), }, ) def _build_prompt(self, req: ToolCallRequest, rule_checks: list[CheckResult]) - str: check_summary \n.join( f- {c.name}: {c.reason} for c in rule_checks ) return ( You are an action reviewer. Determine whether this tool call is justified.\n fuser_id: {req.user_id}\n fagent_id: {req.agent_id}\n ftool_name: {req.tool_name}\n fparams: {json.dumps(req.params, ensure_asciiFalse)}\n freason: {req.reason}\n fevidence: {req.evidence}\n frule check results:\n{check_summary}\n Return JSON only: {approved: bool, confidence: 0.0-1.0, explanation: ...} ) def _call_model(self, prompt: str) - dict: url f{self.base_url}/chat/completions payload { model: self.model, messages: [{role: user, content: prompt}], temperature: self.temperature, max_tokens: self.max_tokens, response_format: {type: json_object}, } headers { Authorization: fBearer {self.api_key}, Content-Type: application/json, } resp httpx.post(url, jsonpayload, headersheaders, timeoutself.timeout) resp.raise_for_status() content resp.json()[choices][0][message][content] return json.loads(content)这里把异常统一转成CheckResult(passedFalse)理由是宁可拒绝一次可疑请求也不要因为 LLM 不稳定而放行。4.5 组装 Gate 服务和 Agent 拦截逻辑main.py暴露一个 POST 接口。通过这个接口所有 Agent 的工具调用请求都会进入 ActionGate 检查。# main.py from fastapi import FastAPI from models import ToolCallRequest, GateDecision from policies import load_config from validators import ActionGate from llm_reviewer import LLMReviewer app FastAPI() config load_config() reviewer LLMReviewer(config.get(llm_reviewer, {})) gate ActionGate(config, reviewerreviewer) app.post(/v1/check_tool_call, response_modelGateDecision) async def check_tool_call(req: ToolCallRequest): return gate.check_tool_call(req)agent_loop.py演示 Agent 在调用执行器前必须先经过 Gate。这个例子省略了模型规划部分聚焦在“拦截”这个关键动作上。# agent_loop.py from models import ToolCallRequest from validators import ActionGate class Executor: def execute(self, tool_name: str, params: dict): # 实际项目中这里会调用邮件、数据库、订单服务等 return fexecuted {tool_name} class Agent: def __init__(self, gate: ActionGate, executor: Executor): self.gate gate self.executor executor def run(self, task: str): # 这里省略真实 Agent 的 planning 过程 # 重点是所有工具调用都必须统一走 Gate。 calls self._plan_tool_calls(task) results [] for call in calls: decision self.gate.check_tool_call(call) if not decision.allowed: results.append({ status: denied, message: decision.message, request_id: call.request_id, }) continue result self.executor.execute(call.tool_name, call.params) results.append({status: executed, result: result}) return results def _plan_tool_calls(self, task: str) - list[ToolCallRequest]: # 示例中直接返回一个请求实际应来自模型输出 return [ ToolCallRequest( request_idreq_demo_1, agent_idagent-demo, user_iduser_42, tool AI Agent 行动安全：用 FastAPI 构建五层校验 Action Gate
当 AI Agent 从聊天窗口走向生产环境核心变化不是模型更强了而是它开始真正改变外部状态发邮件、改订单、执行命令、写数据库。模型输出只是文本这些动作却会产生真实后果。于是工程问题变成了如何让 AI 在行动之前先证明这次行动是应该执行的。这正是“AI learned to act. I built the gate that makes it prove it should”这句话背后的技术意图Agent 学会了行动那网关要做的就是要求它提交理由、证据、权限上下文然后决定放行还是拒绝。本文实现一个可运行的 AI Action Gate用 FastAPI 搭建检查服务在 Agent 的工具调用链路上加入五层校验并通过实例演示允许、拒绝和异常处理。1. 为什么 AI Agent 需要一道门而不是一套提示词1.1 行动能力的代价从生成文本到执行操作早期的 LLM 应用主要是“生成内容”模型输出一段文字用户自己判断是否采用。这个阶段即使模型产生错误影响也局限在文本层面用户有最后决定权。AI Agent 出现后模型输出不再只是文字而是变成结构化的工具调用例如send_email、delete_file、create_order。这些调用一旦执行就会改变系统状态而且往往是不可逆的。以一个简单的客服 Agent 为例。用户说“帮我给客户发送一份包含对账单的邮件”Agent 需要调用邮件工具。如果只是在提示词里写“发送前要检查收件人”遇到模型幻觉、参数拼写错误、上下文拼接错误仍然可能把邮件发给错误的人。更危险的是如果 Agent 同时被注入了恶意指令提示词约束很难挡住所有异常路径。Gate 的存在就是把这些“可能出错但又不能完全靠模型自觉”的动作放到一个显式的、可编程的检查点上。它不是用来替代 Agent 的能力而是确保 Agent 的每一次动作在进入执行器之前都经过规则、权限和语义层面的验证。1.2 提示词约束为什么不可靠很多人会问既然模型已经理解了规则为什么不直接在 system prompt 里写“未经验证不得执行外部调用”原因有四点。第一LLM 的输出是概率性的同样的 prompt 在不同输入下可能产生不同行为。第二工具调用参数来自模型生成模型可能把to和cc字段填反或者忽略必填参数。第三提示词注入会改变模型对规则的判断攻击者可以把恶意内容藏在用户输入里让 Agent 误以为某个操作是合法的。第四提示词本身不产生审计记录如果操作出了问题事后很难回答“当时模型为什么这样调用”。所以提示词约束适合做“行为引导”不适合做“安全边界”。安全边界必须由代码强制实现。Gate 就是代码层的一道边界它要求每个工具调用都提交一个标准请求包括tool_name、params、reason、evidence等字段然后由独立的检查链决定是否允许执行。1.3 Gate 的本质把“证明”变成强制检查点“让 AI 证明它应该行动”这句话很容易被理解成一种理念让模型解释自己的行为。但在工程实现里它必须变成协议和代码。Gate 不关心模型内心怎么想只关心这次请求是否满足预先定义的条件。一次完整的请求至少要回答四个问题动作是什么调用哪个工具参数是什么。谁在发起哪个 Agent、哪个用户。为什么发起理由是什么有没有用户指令或任务上下文作为依据。证据在哪里有没有工单编号、授权记录、会话引用。Gate 根据这些问题执行检查。规则层负责判断“能不能做”语义层负责判断“应不应该做”。两层都通过才允许执行。这个设计让 AI 的行动从“模型说了算”变成“策略说了算”也让它留下的每一步都具备可追溯性。2. 门禁系统设计一次工具调用五层检查2.1 核心数据流Action Gate 本质上是一个独立的服务或模块部署在 Agent 的执行器之前。Agent 生成工具调用后不能直接调用执行函数而是先把请求发给 GateGate 返回allow或deny只有allow才能进入执行器。数据流如下用户 - Agent | v Agent 工具调用请求 (ToolCallRequest) | v Action Gate 服务 ├─ 1. Schema 校验 ├─ 2. 权限校验 ├─ 3. 频率/预算校验 ├─ 4. 关键词规则校验 └─ 5. LLM 语义审查 | v 允许 - 执行器执行 拒绝 - 返回原因并记录日志这个链路的关键点在于“不可跳过”。Agent 内部可以有自己的规划逻辑但最终的工具调用必须统一经过 Gate。如果某个 Agent 绕过 Gate 直接调用执行器Gate 就无法发挥作用。2.2 五层检查Schema、权限、预算、语义、审计这里设计五层检查每一层解决一类问题。前四层是规则校验成本低、速度快第五层是语义校验成本高、判断更灵活。检查层作用通过条件失败结果Schema 校验检查参数是否完整、格式是否正确必填字段存在参数结构符合工具定义不进入执行器返回明确原因权限校验检查 Agent / 用户是否有调用该工具的权限Agent 在允许列表中工具未被禁用拒绝并标记越权行为频率 / 预算校验防止无限循环和成本失控单位时间调用数未超过阈值成本未超预算拒绝并提示稍后重试关键词规则校验拦截危险参数和敏感信息参数中不包含 deny_keywords拒绝并记录触发关键词LLM 语义审查判断请求理由与用户意图是否一致模型认为理由合理且置信度达标拒绝保留人工复核入口Schema 和权限通常由配置文件维护适合静态场景。频率和预算需要动态计数适合在线场景。关键词规则可以快速拦截“明显不该出现的内容”LLM 语义审查则处理规则无法覆盖的模糊场景。2.3 为什么需要请求-凭据协议Gate 要判断“证明是否成立”前提是请求里必须有可验证的信息。如果请求只包含tool_name和paramsGate 只能做参数和权限校验无法判断这次行动是否真的有用户意图支撑。因此需要设计一个标准的请求模型。下面是一个请求示例{ request_id: req_20250101_001, agent_id: customer-support-agent, user_id: user_42, tool_name: send_email, params: { to: opsexample.com, subject: 部署完成通知, body: 生产环境 v2.3 部署完成 }, reason: 用户要求部署完成后发送通知, evidence: ticket#1234, task_iddeploy_20250101 }reason是模型给出的行动理由evidence是支撑这个理由的上下文信息。语义审查层可以根据这两个字段判断请求是否合理。比如用户只要求“查询订单”但 Agent 却尝试调用“删除订单”这就是明显的意图偏差。请求模型定了Gate 的实现边界也就清晰了。下面进入环境准备和代码实现。3. 环境准备与项目结构3.1 技术选型与依赖示例使用 Python 3.10 及以上版本Web 框架使用 FastAPI数据校验使用 Pydantic v2。之所以选 FastAPI是因为它天然支持 Pydantic 模型适合快速搭建 JSON API也方便后续接入中间件、限流和日志。需要安装的依赖如下。建议先创建虚拟环境避免污染系统 Pythonpython -m venv .venv source .venv/bin/activate pip install fastapi uvicorn pydantic PyYAML httpx pytest各依赖的用途如下依赖用途fastapi提供 Web API 框架uvicorn本地启动 ASGI 服务pydantic定义请求和响应模型PyYAML读取策略配置文件httpx调用外部 LLM 接口pytest编写自动化验证用例3.2 项目目录结构项目可以按模块拆分而不是把所有逻辑堆在单个文件里。这里给出一个便于扩展的目录结构action_gate/ ├── main.py ├── models.py ├── validators.py ├── llm_reviewer.py ├── agent_loop.py ├── config.yaml └── tests/ └── test_gate.pymodels.py定义请求、检查结果、最终决策的数据结构。validators.py实现 ActionGate 的规则检查链。llm_reviewer.py实现基于 LLM 的语义审查。agent_loop.py演示 Agent 如何被打断并强制经过 Gate。config.yaml维护 Agent、工具、策略等配置。tests/test_gate.py写自动化验证。3.3 配置文件Gate 的策略不应该写死在代码里。先准备一个config.yaml后续无论是调整允许工具还是修改限流阈值都不需要重新发版。agents: agent-demo: role: assistant allowed_tools: - send_email - query_order tools: send_email: required_keys: [to, subject, body] params_schema: to: string subject: string body: string deny_keywords: [password, token] cost_weight: 2 query_order: required_keys: [order_id] params_schema: order_id: string deny_keywords: [] cost_weight: 1 policies: max_calls_per_minute: 10 llm_reviewer: enabled: false base_url: https://api.example.com/v1 model: review-model api_key_env: LLM_REVIEW_API_KEY temperature: 0.0 max_tokens: 256 timeout_seconds: 10这里把llm_reviewer.enabled设为false方便先跑通规则链路。如果不配置 API KeyLLM 审查应该降级为“直接放行规则检查结果”而不是让整个服务报错。注意config.yaml里的地址和 Key 只是示例。生产环境不要把密钥直接写在文件里建议通过环境变量或配置中心管理。3.4 学习环境与生产环境差异本地单进程跑一个 FastAPI 服务和在生产环境接入多个 Agent 实例差别非常大。下面列一张对照表帮助定位当前阶段应该关注什么。维度学习环境生产环境限流计数内存字典重启丢失Redis 或独立计数服务策略存储本地 YAML配置中心支持灰度发布LLM 审查可关闭必须开启并设置超时与降级审计日志打印到控制台推送到日志平台保留足够周期高可用单实例多实例部署Gate 做无状态设计人工审批不实现高风险动作接入人工复核队列这些差异决定了代码架构不能把“内存变量”看得太重至少要把数据结构和接口抽象出来后续换成 Redis 时不需要改所有业务代码。4. 最小实现用 FastAPI 搭建 Action Gate4.1 定义请求和响应模型models.py负责定义所有核心数据结构。先定义ToolCallRequest这是 Agent 提交给 Gate 的标准请求再定义CheckResult表示单个检查项的结果最后定义GateDecision表示 Gate 的最终决策。# models.py from typing import Any from pydantic import BaseModel, Field class ToolCallRequest(BaseModel): request_id: str agent_id: str user_id: str tool_name: str params: dict[str, Any] Field(default_factorydict) reason: str evidence: str class CheckResult(BaseModel): name: str passed: bool reason: str metadata: dict[str, Any] Field(default_factorydict) class GateDecision(BaseModel): request_id: str allowed: bool decision: str checks: list[CheckResult] Field(default_factorylist) message: str 这里的reason和evidence是语义审查的核心输入。如果某个 Agent 调用没有填写reasonGate 可以把它视为“行动理由不足”即使参数合法也可以拒绝。4.2 实现配置加载policies.py不需要写复杂逻辑只要把 YAML 文件读取成 Python dict 即可。考虑到后续可能接入配置中心这里单独拆一个模块方便替换实现。# policies.py import yaml from pathlib import Path def load_config(path: str config.yaml) - dict: config_path Path(path) with open(config_path, r, encodingutf-8) as f: return yaml.safe_load(f)如果文件不存在这里会抛出文件异常。实际项目中建议在启动时做一次配置合法性检查比如tools必须存在agents必须存在否则直接拒绝启动。4.3 实现多层校验器validators.py是 Gate 的核心。这里实现一个ActionGate类每个检查方法返回一个CheckResult最终聚合所有检查结果。# validators.py import time import threading from models import ToolCallRequest, CheckResult, GateDecision from llm_reviewer import LLMReviewer class ActionGate: def __init__(self, config: dict, reviewer: LLMReviewer | None None): self.config config self.reviewer reviewer self.tools config.get(tools, {}) self.agents config.get(agents, {}) self.max_per_minute config.get(policies, {}).get(max_calls_per_minute, 10) self.recent_calls: list[float] [] self.lock threading.Lock() def check_tool_call(self, req: ToolCallRequest) - GateDecision: checks [] checks.append(self._schema_check(req)) checks.append(self._permission_check(req)) checks.append(self._rate_check(req)) checks.append(self._keyword_check(req)) rule_passed all(c.passed for c in checks) if rule_passed and self.reviewer is not None: checks.append(self.reviewer.review(req, checks)) allowed all(c.passed for c in checks) message if not allowed: failed [c for c in checks if not c.passed] if failed: message failed[-1].reason return GateDecision( request_idreq.request_id, allowedallowed, decisionallow if allowed else deny, checkschecks, messagemessage, ) def _schema_check(self, req: ToolCallRequest) - CheckResult: tool self.tools.get(req.tool_name) if not tool: return CheckResult( nameschema, passedFalse, reasonftool {req.tool_name} not registered, ) for key in tool.get(required_keys, []): if key not in req.params: return CheckResult( nameschema, passedFalse, reasonfmissing required param {key}, ) return CheckResult(nameschema, passedTrue, reasonparams schema ok) def _permission_check(self, req: ToolCallRequest) - CheckResult: agent self.agents.get(req.agent_id) if not agent: return CheckResult( namepermission, passedFalse, reasonfunknown agent {req.agent_id}, ) allowed_tools agent.get(allowed_tools, []) if req.tool_name not in allowed_tools: return CheckResult( namepermission, passedFalse, reasonfagent {req.agent_id} cannot call {req.tool_name}, ) return CheckResult(namepermission, passedTrue, reasonagent permission ok) def _rate_check(self, req: ToolCallRequest) - CheckResult: now time.time() with self.lock: self.recent_calls [t for t in self.recent_calls if now - t 60] if len(self.recent_calls) self.max_per_minute: return CheckResult( namerate, passedFalse, reasonrate limit exceeded, try later, ) self.recent_calls.append(now) return CheckResult(namerate, passedTrue, reasonrate ok) def _keyword_check(self, req: ToolCallRequest) - CheckResult: tool self.tools.get(req.tool_name) if not tool: return CheckResult( namekeyword, passedFalse, reasontool not registered, cannot check keywords, ) deny_keywords tool.get(deny_keywords, []) text str(req.params).lower() for keyword in deny_keywords: if keyword.lower() in text: return CheckResult( namekeyword, passedFalse, reasonfkeyword blocked: {keyword}, ) return CheckResult(namekeyword, passedTrue, reasonkeyword check ok)代码里值得注意的一点是_rate_check使用了线程锁和内存列表这只适合单进程学习环境。多 worker 部署时每个 worker 有独立内存限流会失效需要把计数放到 Redis。4.4 接入 LLM 语义审查llm_reviewer.py负责语义层面的审查。这里的核心是构造一个审查 prompt让模型输出approved、confidence和explanation。如果未配置 LLM就直接返回通过结果避免阻塞主流程。# llm_reviewer.py import json import os import httpx from models import ToolCallRequest, CheckResult class LLMReviewer: def __init__(self, config: dict): self.config config self.enabled config.get(enabled, False) self.base_url config.get(base_url, ) self.model config.get(model, ) self.api_key os.getenv(config.get(api_key_env, ), ) self.timeout config.get(timeout_seconds, 10) self.max_tokens config.get(max_tokens, 256) self.temperature config.get(temperature, 0.0) def review(self, req: ToolCallRequest, rule_checks: list[CheckResult]) - CheckResult: if not self.enabled: return CheckResult( namellm_review, passedTrue, reasonllm review disabled, ) prompt self._build_prompt(req, rule_checks) try: result self._call_model(prompt) except Exception as exc: return CheckResult( namellm_review, passedFalse, reasonfllm review error: {exc}, ) if result.get(approved): return CheckResult( namellm_review, passedTrue, reasonsemantic review approved, metadata{ confidence: result.get(confidence, 0.0), explanation: result.get(explanation, ), }, ) return CheckResult( namellm_review, passedFalse, reasonresult.get(explanation, semantic review denied), metadata{ confidence: result.get(confidence, 0.0), }, ) def _build_prompt(self, req: ToolCallRequest, rule_checks: list[CheckResult]) - str: check_summary \n.join( f- {c.name}: {c.reason} for c in rule_checks ) return ( You are an action reviewer. Determine whether this tool call is justified.\n fuser_id: {req.user_id}\n fagent_id: {req.agent_id}\n ftool_name: {req.tool_name}\n fparams: {json.dumps(req.params, ensure_asciiFalse)}\n freason: {req.reason}\n fevidence: {req.evidence}\n frule check results:\n{check_summary}\n Return JSON only: {approved: bool, confidence: 0.0-1.0, explanation: ...} ) def _call_model(self, prompt: str) - dict: url f{self.base_url}/chat/completions payload { model: self.model, messages: [{role: user, content: prompt}], temperature: self.temperature, max_tokens: self.max_tokens, response_format: {type: json_object}, } headers { Authorization: fBearer {self.api_key}, Content-Type: application/json, } resp httpx.post(url, jsonpayload, headersheaders, timeoutself.timeout) resp.raise_for_status() content resp.json()[choices][0][message][content] return json.loads(content)这里把异常统一转成CheckResult(passedFalse)理由是宁可拒绝一次可疑请求也不要因为 LLM 不稳定而放行。4.5 组装 Gate 服务和 Agent 拦截逻辑main.py暴露一个 POST 接口。通过这个接口所有 Agent 的工具调用请求都会进入 ActionGate 检查。# main.py from fastapi import FastAPI from models import ToolCallRequest, GateDecision from policies import load_config from validators import ActionGate from llm_reviewer import LLMReviewer app FastAPI() config load_config() reviewer LLMReviewer(config.get(llm_reviewer, {})) gate ActionGate(config, reviewerreviewer) app.post(/v1/check_tool_call, response_modelGateDecision) async def check_tool_call(req: ToolCallRequest): return gate.check_tool_call(req)agent_loop.py演示 Agent 在调用执行器前必须先经过 Gate。这个例子省略了模型规划部分聚焦在“拦截”这个关键动作上。# agent_loop.py from models import ToolCallRequest from validators import ActionGate class Executor: def execute(self, tool_name: str, params: dict): # 实际项目中这里会调用邮件、数据库、订单服务等 return fexecuted {tool_name} class Agent: def __init__(self, gate: ActionGate, executor: Executor): self.gate gate self.executor executor def run(self, task: str): # 这里省略真实 Agent 的 planning 过程 # 重点是所有工具调用都必须统一走 Gate。 calls self._plan_tool_calls(task) results [] for call in calls: decision self.gate.check_tool_call(call) if not decision.allowed: results.append({ status: denied, message: decision.message, request_id: call.request_id, }) continue result self.executor.execute(call.tool_name, call.params) results.append({status: executed, result: result}) return results def _plan_tool_calls(self, task: str) - list[ToolCallRequest]: # 示例中直接返回一个请求实际应来自模型输出 return [ ToolCallRequest( request_idreq_demo_1, agent_idagent-demo, user_iduser_42, tool