十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

AI Agent接入业务系统的确定性网关设计:从失控到可控

AI Agent接入业务系统的确定性网关设计:从失控到可控 AI Agent 接入业务系统时最让人头疼的不是模型能力不够而是不可控。模型输出的内容天生带有随机性如果让它直接调用你的订单接口、支付接口、库存接口哪怕只有 1% 的概率调错参数生产环境都可能出大问题。最近看到 Stonefold 这个项目思路非常清晰在 AI Agent 和内部系统之间加一层“确定性网关”让 Agent 只能走预定义好的、校验过的、可审计的通道。本文就从这套思路出发拆解确定性网关的核心设计并给出一个可运行的 Python 示例。1. 为什么 AI Agent 接入业务系统那么难1.1 Agent 调用系统时的“失控”问题先来看一个典型场景。你开发了一个智能助手用户对它说“帮我把上一笔订单退款”。这个请求到了 Agent 之后大模型需要完成几个动作理解用户意图确定要调用“退款”能力。从对话上下文中提取订单号。调用后端系统的退款接口。问题出在第三步。传统接口调用是由代码写死的参数类型、必填项、权限范围在编译期就确定了。但 Agent 调用接口时参数是由大模型现场生成的它可能把订单号格式传错比如多传了一个空格。把“查询订单”理解成“退款订单”。在用户没有退款权限的情况下依然发起了退款请求。连续重试同一个请求导致重复退款。这些问题的根源不是模型不够聪明而是缺少一层“强制约束”。就像公司门禁不能因为访客看起来友善就放他进机房必须刷卡、登记、走指定通道。1.2 什么是确定性网关确定性网关Deterministic Gateway是位于 AI Agent 与业务系统之间的一层中间服务它的核心目标是把 Agent 的“自由发挥”限制在一个预先定义好的框架内。传统做法中Agent 直接调用函数或 API调用链是Agent → 业务接口引入确定性网关后调用链变成Agent → 确定性网关 → 业务接口网关不是简单地转发请求它会做四件关键事情能力说明契约校验只允许调用网关注册过的工具参数格式必须符合 JSON Schema权限控制根据调用者身份、上下文决定是否放行审计记录记录每一次调用的完整输入、输出、耗时、调用者确定性重放相同输入必须产生相同决策链路便于测试和排查Stonefold 正是这一类思路的落地项目。它不是把 Agent 关起来而是给 Agent 修了一条有护栏的高速公路——速度快但不会冲出车道。1.3 Stonefold 的定位与适用场景从项目名称 “a deterministic gateway between AI agents and your systems” 可以看出来Stonefold 关注的是集成层而不是模型层。它不负责训练模型也不负责 Prompt 优化它负责的是 Agent 和系统之间的“翻译安检记录”。适用场景非常典型企业内部知识库 Agent 需要查询 CRM 数据。客服机器人需要调用订单系统和售后系统。运维 Agent 需要执行只读的诊断命令。数据分析助手需要读取数据库并对结果做脱敏。这些场景有一个共同点系统很重要不能出错而且需要追责。如果你只是做个 Demo让 Agent 调用几个公开 API那不需要网关但一旦涉及生产数据和核心链路网关就是必需品。2. 确定性网关的关键设计原则2.1 模式即边界让 Agent 只能走预定义路径确定性网关最重要的原则是——Agent 没有“自由选择调用哪个函数”的权利。Agent 能做的是从网关暴露的工具列表里选一个工具然后按照工具定义的参数格式填写参数。听起来像绕了一圈其实不是。Agent 的“工具选择”依然由大模型决定但网关的注册表里只有你允许它看到的工具。比如你只注册了query_order和create_refund那 Agent 无论如何都调不到delete_order因为它在工具列表里根本不存在。这就像给 Agent 一张菜单菜单上有什么它才能点什么。菜单上没有的菜它再聪明也点不出来。2.2 状态可追溯每一次调用都是可复现的确定性还有一个层面的含义就是“可复现”。当你排查线上问题时你最希望看到的是这个请求是什么时候来的。是哪个会话、哪个用户触发的。Agent 为什么选择调用这个工具。传给工具的原始参数是什么。工具返回了什么结果。网关是否放行如果没有放行原因是什么。这些信息需要用一个全局唯一的request_id串起来并且存储成结构化日志。最好还能把请求的入参做哈希方便后续做回归比对。2.3 权限收敛最小权限 可撤销Agent 调用系统时权限模型不能是“Agent 拥有系统全部权限”。更合理的做法是每个 Agent 绑定一个身份service account。每个身份只能调用特定工具。每个工具的调用需要有操作人上下文。权限可以动态撤销不用重启服务。网关在接到请求时先校验“调用者是否有权限调用这个工具”再校验参数最后才转发给业务系统。这两步校验顺序不能反因为权限校验失败时不需要做参数解析降低无效计算。3. 环境准备与整体架构3.1 环境与版本说明本文的演示代码使用常见环境具体版本需要根据你的项目实际情况调整Python 3.10 或更高版本FastAPI用于演示网关 HTTP 接口Pydantic用于参数校验业务系统使用本地模拟函数代替如果你还没有安装依赖可以先创建虚拟环境python3 -m venv .venv source .venv/bin/activate pip install fastapi uvicorn pydantic3.2 整体架构整个系统的调用链路如下用户 → AI Agent → Stonefold 网关 → 业务系统 │ ├─ 工具注册表 ├─ 权限校验器 ├─ 参数校验器 └─ 审计日志网关部署为独立服务AI Agent 通过网络调用网关网关再调用后端业务系统。这样业务系统不需要直接面对 Agent 的不可控请求。3.3 演示项目结构为了方便阅读我们用一个最小项目来演示stonefold-demo/ ├── gateway.py # 网关主程序 ├── tools.py # 工具注册表与业务模拟 ├── audit.py # 审计日志模块 ├── schema.py # 参数校验定义 └── requirements.txt # 依赖清单下面逐个文件讲解实现思路和代码。4. 从零实现一个 Stonefold 风格网关这里实现的目的是演示确定性网关的核心机制不是 Stonefold 的完整源码。你可以把这份代码当作理解原理的骨架再把真实业务逻辑补充进去。4.1 定义工具注册表工具注册表解决的是“Agent 可以调用什么”的问题。每个工具至少包含名称、描述、参数 Schema、处理函数、权限要求。# 文件路径stonefold-demo/tools.py from typing import Callable, Dict, Any, Optional from datetime import datetime def query_order(order_id: str) - Dict[str, Any]: 模拟查询订单接口 return { order_id: order_id, status: PAID, amount: 299.00, created_at: 2025-01-15 10:30:00, } def create_refund(order_id: str, reason: str) - Dict[str, Any]: 模拟创建退款单接口 return { refund_id: fRF{int(datetime.now().timestamp())}, order_id: order_id, reason: reason, status: PENDING, } # 工具注册表Agent 只能看到这个字典里注册的工具 TOOL_REGISTRY: Dict[str, Dict[str, Any]] { query_order: { description: 根据订单号查询订单信息, parameters: { type: object, properties: { order_id: {type: string, description: 订单号} }, required: [order_id], }, handler: query_order, required_permission: order:read, }, create_refund: { description: 为指定订单创建退款单, parameters: { type: object, properties: { order_id: {type: string, description: 订单号}, reason: {type: string, description: 退款原因}, }, required: [order_id, reason], }, handler: create_refund, required_permission: order:refund, }, }这里有几个设计细节需要注意parameters使用了 JSON Schema 格式好处是可以直接用 Pydantic 或 jsonschema 库做校验而且这个格式也是 OpenAI Function Calling 的标准格式Agent 侧可以直接使用。required_permission是权限标记网关会在校验参数之前检查。handler 是实际执行逻辑的函数在真实项目中这里会替换成 HTTP 调用或 RPC 调用。4.2 实现参数校验与调用分发网关收到 Agent 请求后第一件事不是执行业务逻辑而是做三层检查工具是否存在。调用者是否有权限。参数是否符合 Schema。# 文件路径stonefold-demo/schema.py from jsonschema import validate, ValidationError from tools import TOOL_REGISTRY def validate_tool_call(tool_name: str, arguments: dict) - None: 校验工具参数是否符合定义 if tool_name not in TOOL_REGISTRY: raise ValueError(fUnknown tool: {tool_name}) schema TOOL_REGISTRY[tool_name][parameters] try: validate(instancearguments, schemaschema) except ValidationError as exc: raise ValueError(fInvalid arguments: {exc.message})校验逻辑使用jsonschema库它在 Python 生态中非常成熟也支持复杂的嵌套结构、枚举值、正则校验。4.3 审计日志模块审计日志是确定性网关和普通 API 网关最明显的区别。普通网关关心的是“请求是否成功”确定性网关关心的是“请求为什么这么发、为什么被放行或拒绝、下次如何复现”。# 文件路径stonefold-demo/audit.py import json import hashlib from datetime import datetime from typing import Dict, Any def _hash_request(payload: Dict[str, Any]) - str: 对请求内容做哈希用于后续比对和回归 raw json.dumps(payload, sort_keysTrue, ensure_asciiFalse) return hashlib.sha256(raw.encode(utf-8)).hexdigest() def write_audit_log( request_id: str, agent_id: str, user_id: str, tool_name: str, arguments: Dict[str, Any], decision: str, reason: str, response: Any None, ) - None: 写审计日志 log_entry { request_id: request_id, timestamp: datetime.utcnow().isoformat(), agent_id: agent_id, user_id: user_id, tool_name: tool_name, arguments: arguments, request_hash: _hash_request(arguments), decision: decision, # ALLOW / DENY / ERROR reason: reason, response: response, } # 实际项目中这里应该写入日志系统或消息队列 print(json.dumps(log_entry, ensure_asciiFalse, indent2))这个模块记录了每个请求的完整链路。当线上出现问题时你可以通过request_id快速定位到一条审计记录查看当时的完整上下文。4.4 网关主程序网关主程序把上面的模块串起来对外提供 HTTP 接口。Agent 只需要把工具名和参数 POST 过来网关负责做决定。# 文件路径stonefold-demo/gateway.py import uuid from typing import Dict, Any from fastapi import FastAPI, HTTPException, Header from pydantic import BaseModel, Field from tools import TOOL_REGISTRY from schema import validate_tool_call from audit import write_audit_log app FastAPI(titleStonefold Demo Gateway) class ToolCallRequest(BaseModel): Agent 调用工具的请求体 agent_id: str Field(..., descriptionAgent 身份标识) user_id: str Field(..., description最终操作人标识) tool_name: str Field(..., description要调用的工具名) arguments: Dict[str, Any] Field(default_factorydict, description工具参数) # 简化版权限表真实场景应该对接 RBAC 服务 PERMISSIONS { assistant_v1: [order:read], after_sale_agent: [order:read, order:refund], } def check_permission(agent_id: str, required_permission: str) - bool: 判断 Agent 是否具备调用工具的权限 permissions PERMISSIONS.get(agent_id, []) return required_permission in permissions app.post(/v1/tool_call) def tool_call( request: ToolCallRequest, x_request_id: str Header(default_factorylambda: str(uuid.uuid4())), ): request_id x_request_id or str(uuid.uuid4()) # 第一步工具存在性检查 if request.tool_name not in TOOL_REGISTRY: write_audit_log( request_idrequest_id, agent_idrequest.agent_id, user_idrequest.user_id, tool_namerequest.tool_name, argumentsrequest.arguments, decisionDENY, reasonunknown_tool, ) raise HTTPException(status_code404, detailtool not found) tool_info TOOL_REGISTRY[request.tool_name] # 第二步权限检查 if not check_permission(request.agent_id, tool_info[required_permission]): write_audit_log( request_idrequest_id, agent_idrequest.agent_id, user_idrequest.user_id, tool_namerequest.tool_name, argumentsrequest.arguments, decisionDENY, reasonpermission_denied, ) raise HTTPException(status_code403, detailpermission denied) # 第三步参数校验 try: validate_tool_call(request.tool_name, request.arguments) except ValueError as exc: write_audit_log( request_idrequest_id, agent_idrequest.agent_id, user_idrequest.user_id, tool_namerequest.tool_name, argumentsrequest.arguments, decisionERROR, reasonstr(exc), ) raise HTTPException(status_code400, detailstr(exc)) # 第四步执行调用 try: handler tool_info[handler] result handler(**request.arguments) except Exception as exc: write_audit_log( request_idrequest_id, agent_idrequest.agent_id, user_idrequest.user_id, tool_namerequest.tool_name, argumentsrequest.arguments, decisionERROR, reasonfhandler_error: {exc}, ) raise HTTPException(status_code500, detailinternal handler error) # 第五步记录成功审计 write_audit_log( request_idrequest_id, agent_idrequest.agent_id, user_idrequest.user_id, tool_namerequest.tool_name, argumentsrequest.arguments, decisionALLOW, reasonok, responseresult, ) return { request_id: request_id, tool_name: request.tool_name, result: result, }这个主程序把前面说的“确定性”落到了代码上。无论 Agent 怎么折腾网关只会按固定顺序执行查工具、查权限、查参数、执行、记录。4.5 运行与验证启动服务uvicorn gateway:app --reload --port 9000使用 curl 模拟 Agent 发起一次合法请求curl -X POST http://localhost:9000/v1/tool_call \ -H Content-Type: application/json \ -d { agent_id: after_sale_agent, user_id: user_1001, tool_name: query_order, arguments: {order_id: 20250115001} }预期返回{ request_id: xxx-xxx-xxx, tool_name: query_order, result: { order_id: 20250115001, status: PAID, amount: 299.0, created_at: 2025-01-15 10:30:00 } }再模拟一次越权请求curl -X POST http://localhost:9000/v1/tool_call \ -H Content-Type: application/json \ -d { agent_id: assistant_v1, user_id: user_1001, tool_name: create_refund, arguments: {order_id: 20250115001, reason: 用户不想要了} }预期返回 403因为assistant_v1只有order:read权限没有order:refund权限。场景agent_idtool_name结果合法调用after_sale_agentquery_order200越权调用assistant_v1create_refund403参数错误after_sale_agentquery_order缺参数400未知工具after_sale_agentdelete_order4044.6 确定性评估给 Agent 集成做回归测试网络热词 “demystifying evals for AI agents” 说的正是 Agent 评估问题。很多人以为 Agent 的评估就是跑几个 Prompt 看回答好不好但实际上一旦 Agent 接入了业务系统评估的重点就变成了“工具调用链是否确定”。也就是说与其评估大模型的措辞不如评估它的行为。你可以建立一组“黄金用例”每个用例包含用户输入、预期工具调用序列、预期参数。然后对 Agent 跑回归测试。# 文件路径stonefold-demo/eval_example.py GOLDEN_CASES [ { user_input: 帮我查一下订单 20250115001 的状态, expected_tool: query_order, expected_arguments: {order_id: 20250115001}, }, { user_input: 把订单 20250115001 退款理由是商品破损, expected_tool: create_refund, expected_arguments: { order_id: 20250115001, reason: 商品破损, }, }, ] def evaluate_agent(agent_predict_function, golden_cases): 简单的回归评估函数 passed 0 for case in golden_cases: tool_name, arguments agent_predict_function(case[user_input]) if ( tool_name case[expected_tool] and arguments case[expected_arguments] ): passed 1 else: print( fFAILED: {case[user_input]}\n f expected: {case[expected_tool]} {case[expected_arguments]}\n f actual: {tool_name} {arguments} ) print(fPassed {passed}/{len(golden_cases)}) return passed len(golden_cases)这种回归测试的价值在于当你更换模型版本、修改 Prompt 或调整工具描述时可以快速发现 Agent 的调用行为发生了哪些变化。确定性网关保证的是“网关层一定可靠”而 eval 保证的是“Agent 在网关约束下的行为符合预期”两者配合才构成完整的可靠性体系。5. 常见问题与排查思路在接入确定性网关的过程中团队遇到的高频问题大致如下问题现象常见原因解决思路Agent 经常传错参数工具描述不够清晰模型理解偏差完善工具 description细化参数说明和枚举值调用工具总是 403Agent 身份未配置权限检查网关权限表确认 agent_id 对应的权限同一个请求重复执行Agent 对超时请求自动重试网关增加幂等键根据 request_id 去重线上问题无法排查审计日志不完整缺少上下文确保每次调用都写审计日志并保存完整入参模型升级后行为异常工具选择策略改变运行回归 eval对比升级前后的工具调用分布网关负载过高每次请求都做大量校验参数 Schema 预编译权限结果做短时间缓存下面展开几个排查场景。5.1 参数校验总是失败如果 Agent 调用工具时频繁出现 400首先去查看审计日志中记录的arguments实际值。常见问题是JSON 中多余了空格或转义字符。日期格式不符合 Schema 中的 pattern。枚举值大小写不一致。建议在工具描述中明确写出参数示例比如order_id: 格式为 YYYYMMDD 3 位随机数字例如 20250115001对大模型来说一个具体的示例胜过十行抽象描述。5.2 日志太多无法快速定位当请求量上来之后全量打印审计日志会占用大量存储。建议按重要程度分两级基础日志记录 request_id、agent_id、tool_name、decision、耗时。详细日志记录完整 arguments 和 response仅在需要时开启。在排查问题时先通过基础日志缩小范围再对具体 request_id 拉取详细日志。5.3 网关服务本身出问题怎么办确定性网关是 Agent 调用链路的必经之路如果网关挂了Agent 就完全不可用。因此网关本身需要做到无状态不保存会话数据可以横向扩容。降级如果业务系统不可用网关返回明确的错误码并提示 Agent 稍后重试。监控对每个工具调用的成功率、耗时、拒绝原因做监控。6. 最佳实践与工程建议6.1 权限模型要独立于业务系统不要把权限校验逻辑写在业务系统里。网关层的权限模型应该独立维护便于统一审计。推荐的做法是网关对接企业内部的身份中心或 RBAC 服务Agent 身份使用服务账号Service Account而不是个人账号。6.2 工具的“描述”是模型行为的关键在 Function Calling 场景里工具描述直接影响模型的工具选择准确率。写工具描述时注意几点说清楚工具的职责不要含糊。参数使用业务语言不用内部缩写。给出参数的取值范围或示例。说明工具的副作用比如“此操作会创建退款单不可逆”。6.3 配置与代码分离网关的配置项权限表、工具列表、数据库连接、业务系统地址不应该写死在代码里而应该放到配置中心或环境变量。这样调整权限、新增工具时不需要重新发布服务。# 示例环境变量配置 STONEFOLD_RBAC_ENDPOINThttps://rbac.internal.example.com STONEFOLD_AUDIT_TOPICagent-gateway-audit STONEFOLD_DEFAULT_TIMEOUT_MS30006.4 做好超时和熔断Agent 的调用链比较长任何一个环节超时都可能被模型误判为“工具不可用”。网关调用业务系统时必须有明确的超时时间并且对持续失败的业务系统做熔断避免拖垮整个网关。6.5 评估不是一次性的要纳入 CI/CD如果你在持续迭代 Agent 的 Prompt 或模型版本建议把“黄金用例回归”纳入 CI 流水线。每次改动 Prompt、升级模型、修改工具描述都自动跑一遍评估确保工具调用行为没有发生非预期变化。6.6 安全边界审计日志中如果包含用户敏感信息需要脱敏后存储。网关的 API 需要通过 mTLS 或 API Token 认证不能暴露在公网。权限变更要有审批流程每次变更记录下来。7. 总结与下一步本文从 AI Agent 接入业务系统的失控问题出发介绍了确定性网关的核心思想把 Agent 的自由调用限制在预定义工具集内通过权限校验、参数校验、审计日志和确定性重放来保证系统安全与可追溯。Stonefold 这个项目的定位正好切中这一需求它不改变模型本身而是在模型和业务系统之间加了一层确定性护栏。文章中给出的 Python 示例包含工具注册表、参数校验、权限检查、审计日志和回归评估五部分你可以直接把它改造成自己的网关骨架。下一步建议把示例中的模拟函数替换成真实的业务系统调用。接入你们公司的 RBAC 系统替换示例里的静态权限表。把审计日志接入 ELK 或其他日志平台方便检索。为你的 Agent 建立黄金用例集跑通第一轮回归评估。如果你对 Stonefold 的具体实现感兴趣可以阅读它的源码重点看工具注册和请求校验的实现方式。如果你正在设计 Agent 与内部系统的集成方案不妨把“确定性网关”作为第一层考虑——先保证可控再追求智能。
返回列表