做 AI Agent 应用最让人头疼的时刻不是我调不通大模型而是模型明明理解了需求却在最后一步够不着数据。窗口订好了回执拿不到报表分析出来了Excel 存在对方服务器里客服机器人分析完投诉却没法在工单系统里自动建单。我们团队今年上半年折腾了一个叫Agent-Reach的开源项目核心就一句话把智能体的触达能力从聊天窗口扩展到真实的业务系统让 Agent 能真正够到数据库、API、工单系统、物联网设备这些外部资源。Agent-Reach 不是什么颠覆性发明它更像一套规范加工具箱先用统一协议定义智能体如何触达外部资源再让各类外部系统以连接器Connector的方式接入最后用权限、审计、限流、可观测性把这条链路管理起来。如果你也在做 Agent 类产品或者正在帮业务方接智能体这篇内容应该有不少可以直接抄走的设计和代码。我会把架构思路、连接器的完整实现、权限模型还有我们踩过的坑一次说清楚。1. Agent-Reach 要解决的核心问题智能体的最后一公里触达1.1 智能体的手和脚从对话到行动现在市面上绝大多数 Agent 应用其实停在一个很尴尬的位置话很多手很短。你问它帮我查一下最近三天未发货的订单它能流畅回答好的我来查询然后……就没有然后了。即便接了 RAG、连了知识库本质上也只是在给模型喂更多的信息模型并没有改变任何现实世界里的状态。想让 Agent 真正产生业务价值必须让它行动查订单、发邮件、建工单、调接口、改配置。而行动就意味着触达——触达数据源、触达外部服务、触达物理设备。这个触达的过程和单纯对话完全是两码事对话只关心模型说什么不关心模型调到什么触达关心的是授权怎么给、连接怎么建、失败了怎么办、会不会把线上数据搞坏。Agent-Reach 管的就是触达这一层。它把Agent 想要做的事翻译成外部系统能执行的操作再把执行结果原样送回给 Agent 继续推理。1.2 为什么现有方案不够用做触达这件事业界已经有一些既有做法但我们实测下来都有各自的短板。第一种做法每个 Agent 直接写死 SDK。比如在 Agent 代码里 import 一个 order_service 的客户端然后直接查数据库。好处是简单坏处是业务逻辑和 Agent 逻辑高度耦合每次换一个业务系统都要重新改代码。更麻烦的是权限散落在各个系统里审计无从谈起。第二种做法通过工具调用协议比如这两年火起来的 MCP 这类标准把能力暴露给 Agent。这个方向是对的但它偏底层更像给 Agent 一根网线至于这根网线连到哪、谁能用、用了之后有没有记录协议本身并不管治理能力几乎要全部自建。第三种做法干脆不给 Agent 触达能力只做好看的问答 Demo。这个最省事但价值天花板也最低。我把这三种做法和 Agent-Reach 的设计放到一起对比一下维度直接写 SDK裸用工具调用协议Agent-Reach接入方式每个系统写一遍业务代码手动注册工具连接器统一接入按能力名路由权限控制散落在各业务系统基本没有统一策略引擎 操作分级审计能力看日志碰运气无每次触达全链路可追溯变更影响改动业务代码影响面大替换或升级连接器即可可观测性无弱埋点 trace 健康检查所以 Agent-Reach 的设计目标很清晰把触达这件事独立出来做成一等公民让业务团队接入系统时只关心连接器怎么写让 Agent 团队只关心能力怎么编排让平台团队只关心权限和稳定性怎么管。2. 架构设计与核心模型把触达做成一等公民2.1 四层架构把触达独立出来Agent-Reach 整体分四层这个分层不是拍脑袋定的是踩过耦合的坑之后沉淀出来的。第一层是资源层就是那些真正的数据源和服务MySQL、PostgreSQL、Redis、SaaS 系统、企业微信、物联网网关等等。Agent-Reach 原则上不直接碰这一层而是通过连接器去碰。第二层是连接器层这是整个项目的灵魂。每个外部系统对应一个连接器连接器负责把外部系统的原生能力封装成统一的操作模型。这一层的核心工作是做适配和隔离——适配不同的协议风格隔离外部系统的差异避免上层 Agent 被某个系统特有的字段格式污染。第三层是调度与治理层负责路由、鉴权、限流、熔断、超时和审计。Agent 发出的触达请求先到这里过一遍安检安检通过后才真正转发给连接器。这一层是 Agent-Reach 和裸连工具最大的区别所在。第四层是应用层也就是 Agent 本体、工作流引擎、业务应用。这一层只跟 Agent-Reach 的运行时通信不直接感知底层资源。这个分层带来最实际的好处是换数据库从 MySQL 换到 PostgreSQL上层 Agent 代码一行不用改只需要换一个连接器或者改一下连接配置。我们在一期项目里就经历过一次工单系统迁移当时上层 Agent 的逻辑完全没动只重新注册了新系统的连接器大概半小时搞定。2.2 连接器模型能力声明 执行器连接器在 Agent-Reach 里是一个能力声明 执行器的组合。能力声明叫 manifest是一个 JSON 文件描述这个连接器有哪些操作operation、每个操作的入参出参格式、访问级别、是否幂等。执行器是真正干活的那段代码负责把标准化的入参转换成外部系统的调用。下面是一个订单库连接器的 manifest 片段看这个就明白连接器模型的骨架长什么样{ name: order-db, version: 1.0.0, operations: [ { name: query_orders, description: 按用户ID查询订单列表支持按时间范围过滤, input_schema: { type: object, properties: { user_id: { type: string }, start_time: { type: string }, end_time: { type: string } }, required: [user_id] }, output_schema: { type: array, items: { type: object, properties: { order_id: { type: string }, amount: { type: number }, status: { type: string }, created_at: { type: string } } } }, access: read, idempotent: true, timeout_ms: 5000 } ] }注意这里几个字段不是随便写的。access标记操作级别后面的策略引擎要靠它决定放不放行。idempotent标记这个操作是否幂等这直接关系到 Agent 超时重试时会不会产生脏数据后面踩坑记录里我会详细讲。timeout_ms是单次操作超时独立于全局超时因为不同操作的耗时差距非常大查一条记录和导一批数据完全不是一回事。2.3 注册与发现按能力名触达而不是按地址触达连接器写完之后要注册到 Agent-Reach 的运行时代理Runtime。运行时维护着一张能力路由表里面记录着操作名 → 连接器实例。Agent 调用时只需要能力名比如order-db.query_orders完全不用关心这个操作背后是 HTTP 接口还是数据库连接。注册机制可以用最简单的内存注册表起步生产环境建议走 Redis 或其他配置中心方便多实例同步。注册的核心代码其实很薄class ConnectorRegistry: def __init__(self): self._connectors: dict[str, BaseConnector] {} def register(self, connector: BaseConnector): manifest connector.manifest() for op in manifest.operations: key f{manifest.name}.{op.name} self._connectors[key] connector logger.info(fregistered connector {manifest.name} v{manifest.version}) def resolve(self, full_operation_name: str) - BaseConnector: return self._connectors[full_operation_name]这里的logger不是摆设生产环境里每次注册、注销、重连都应该有日志排查问题的时候能救命的。3. 手写一个数据库触达连接器完整可复现3.1 项目结构与环境准备我们一期只支持 Python 运行时选它的理由很直接团队对 Python 最熟且 Agent 生态里 Python 的工具链最全。环境要求很简单Python 3.11、FastAPI、SQLAlchemy、pydantic如果走 OpenAI 接口再装一个 openai 的 SDK。项目目录结构如下尽量保持扁平避免为了分层而过度设计agent-reach/ ├── runtime/ │ ├── __init__.py │ ├── registry.py # 连接器注册表 │ ├── gateway.py # 运行时网关FastAPI 应用 │ ├── policies.py # 权限策略引擎 │ └── audit.py # 审计日志 ├── connectors/ │ ├── base.py # 连接器基类 │ ├── sql.py # SQL 数据源连接器 │ ├── http_api.py # 通用 HTTP API 连接器 │ └── manifests/ │ ├── order-db.json │ └── crm-api.json ├── config/ │ └── config.yaml └── examples/ └── openai_integration.py # 接入 OpenAI Function Calling 的示例3.2 连接器基类与 SQL 实现连接器基类只需要定义三个抽象方法返回 manifest、执行操作、健康检查。太多抽象方法反而会让接入的人头疼。class BaseConnector(ABC): abstractmethod def manifest(self) - ConnectorManifest: 返回连接器的能力声明Agent-Reach 运行时靠它生成工具描述 abstractmethod async def execute(self, operation: str, params: dict) - ConnectorResult: 执行具体操作operation 是 manifest 里声明的操作名 abstractmethod async def health_check(self) - HealthStatus: 健康检查运行时周期性调用挂了就从路由表摘掉以 SQL 连接器为例它应该是使用频率最高、也最容易写错的连接器。下面是实际能跑的简化版本重点是参数化查询和只读保护class SQLConnector(BaseConnector): def __init__(self, name: str, dsn: str, read_only: bool True): self._name name self._dsn dsn self._read_only read_only self._engine create_async_engine(dsn) self._registry load_manifest(fmanifests/{name}.json) def manifest(self): return self._registry async def execute(self, operation: str, params: dict) - ConnectorResult: op_cfg self._registry.get_operation(operation) if self._read_only and op_cfg.access ! read: raise PermissionDenied(f{operation} is not allowed in read-only mode) query params.get(query) if not query: raise InvalidParameter(missing required field: query) # 只读模式下再做一道保险拦截非 SELECT 请求 if self._read_only and not query.strip().upper().startswith(SELECT): raise PermissionDenied(only SELECT is allowed) limit min(params.get(limit, 50), 200) async with self._engine.connect() as conn: result await conn.exec_driver_sql(query f LIMIT {limit}) rows result.fetchall() columns list(result.keys()) return ConnectorResult( data[dict(zip(columns, row)) for row in rows], truncatedlen(rows) limit )这段代码里有个容易被忽略的细节limit min(params.get(limit, 50), 200)。这个 200 的上限必须写死否则大模型一旦生成一个SELECT * FROM orders不带条件连接器会把整张表倒出来几百万行数据直接把网络打爆上下文窗口更是想都不用想。连接器在返回结果的时候同样要限制返回条数宁可让 Agent 多问几次也不能一次喂太多。3.3 接入 OpenAI Function Calling连接器写完之后最关心的问题就是怎么让大模型调起来。Agent-Reach 的做法是运行时启动时把连接器的 manifest 转换成 OpenAI 的 function 格式直接作为 tools 传给模型。def to_openai_tool(connector: BaseConnector): operations [] for op in connector.manifest().operations: operations.append( { type: function, function: { name: f{connector.manifest().name}.{op.name}, description: op.description, parameters: op.input_schema, }, } ) return operations转换之后接入 Agent 的主体逻辑就非常清爽了import openai client openai.AsyncOpenAI() SYSTEM_PROMPT 你是订单客服助手。查询用户订单时必须使用 order-db.query_orders。 如果用户询问的字段不在返回结果中直接告诉用户当前系统不提供该信息不要编造。 async def chat_with_order_agent(user_message: str): messages [{role: system, content: SYSTEM_PROMPT}] messages.append({role: user, content: user_message}) while True: resp await client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolsto_openai_tool(order_db_connector), ) msg resp.choices[0].message messages.append(msg) if not msg.tool_calls: return msg.content for call in msg.tool_calls: result await gateway.execute( operationcall.function.name, paramsjson.loads(call.function.arguments), trace_id..., principalcustomer-service-agent, ) messages.append( { role: tool, tool_call_id: call.id, content: json.dumps(result.data, ensure_asciiFalse), } )注意这里 Agent 调用的是网关的execute而不是直接碰连接器。所有请求经过网关权限策略、审计、限流才能生效。把这句话多念几遍这是 Agent-Reach 整个设计的命门。调用链路走通之后的效果大概是这样的用户说查一下用户 10086 最近三天的订单模型先输出一个 tool call 请求order-db.query_orders网关校验权限、生成 trace连接器执行 SQL结果回填给模型模型再组织成自然语言回答用户。整个过程对用户来说是一体的但我们内部把对话和触达拆得清清楚楚。3.4 配置与启动配置走 YAML原则一切和环境相关的参数都外置代码里不要出现任何账号密码。下面是 config.yaml 的骨架runtime: host: 0.0.0.0 port: 8000 log_level: info connectors: - type: sql name: order-db dsn: ${ORDER_DB_DSN} # 从环境变量读取 read_only: true health_check_interval_sec: 30 policies: - principal: customer-service-agent allow: [order-db.query_orders, order-db.query_order_detail] deny: [*]启动命令就是常规的 uvicornexport ORDER_DB_DSNpostgresqlasyncpg://readonly_user:xxx10.0.0.8:5432/orders uvicorn runtime.gateway:app --host 0.0.0.0 --port 80004. 安全边界与权限控制放开手脚之前先拴好链子4.1 最小权限从数据库账号到字段级过滤权限这件事我们总结了一个血泪教训给 Agent 的权限永远比你想给的再少一级。Agent 是概率模型它生成的操作参数可能出现各种意外所以底层防护必须是最严格的不能指望模型应该不会做某事。最小权限在 Agent-Reach 里分三层落地第一层是基础设施账号。连接器连接数据库用的账号绝不能是 DBA 或者开发账号应该单独创建只读账号或者最小权限账号。我们内部的规范是生产环境连接器账号只授予SELECT权限和少量表的INSERT或UPDATE权限根据业务需要逐表授权。第二层是连接器内部校验。就是上面 SQL 连接器里做的即使账号有权限连接器也会拦截非预期的操作。比如query_orders这个操作声明了access: read连接器会在执行前检查入参是否包含用户 ID如果没有就拒绝执行。第三层是字段级过滤。有些系统字段比如用户的手机号、身份证号不能让 Agent 随便看。我们在 manifest 的output_schema里扩展了一个masked_fields字段连接器返回结果前对敏感字段做脱敏。这个功能用下来对于客服场景特别重要客服 Agent 不需要也不应该看到完整的用户敏感信息。4.2 操作分级与策略引擎Agent-Reach 把连接器里的操作分成三级read只读查询、write可写单条操作、admin批量写、删除、高危操作。策略引擎的核心是一个简单的匹配器每个 principal 对应一组 allow/deny 规则。规则支持通配符默认拒绝只有显式 allow 才放行。这个策略文件越简单越好复杂的规则引擎反而容易让配置的人自己都搞不清楚。policies: - principal: customer-service-agent allow: - order-db.query_* - crm-api.create_ticket deny: - order-db.delete_* - *.*.drop_*策略引擎的代码逻辑很直观class PolicyEngine: def __init__(self, policies: list[Policy]): self._policies policies async def check(self, principal: str, operation: str) - bool: for policy in self._policies: if principal not in policy.principals: continue if any(fnmatch(operation, deny) for deny in policy.deny): return False if any(fnmatch(operation, allow) for allow in policy.allow): return True return False这里有个实际工程上的细节fnmatch用*通配符做匹配符合大多数操作命名的场景但操作名多了之后容易误命中比如order-db.query_orders_archive也会被order-db.query_*放行。如果业务对权限特别敏感建议改成精确匹配或者引入更精细的路径匹配规则。高危操作比如批量更新、删除只靠策略 Agent 这层其实是不太够的我们还会叠加一层人工审批高危操作先进入 Pending 队列由一个轻量审批 API 通知相关责任人批准之后才真正执行。这个东西在第一期可以不做但做生产级 Agent 应用省不掉。4.3 审计追踪每次触达都可回放权限能拦住不该发生的操作但真要出事的时候审计日志才是你最好的朋友。Agent-Reach 的审计核心是给每次触达分配一个trace_id全链路记录下来。审计日志至少包含以下字段字段含义trace_id一次触达的全链路 IDprincipal调用方身份比如哪个 Agentoperation完整操作名params请求参数敏感字段脱敏result_summary返回结果概要记录行数和状态allowed是否通过权限校验duration_ms执行耗时created_at时间戳审计日志的写入放在网关层面统一做连接器不感知。之前我们犯过一个错误把审计逻辑写在了连接器里结果每个连接器都要重复实现一遍有的连接器漏了有的连接器写日志的格式不统一排查问题时特别痛苦。后来统一收敛到网关事情就简单了。5. 实测数据与踩坑记录5.1 工具描述膨胀上下文窗口的隐形杀手接入 Agent-Reach 的第一个版本我们当时天真地把所有连接器都注册了每个连接器的工具描述都转成 OpenAI function 格式一股脑传给模型。结果上线一测傻眼了光是 tools 这块就占了将近 6000 个 token而一次完整的多轮对话还要算上历史消息、系统提示词上下文窗口很快就满了更糟的是模型开始表现得很犹豫明明该调工具的时候不调净在那答非所问。原因其实不复杂工具描述太长模型对每个工具的注意力被稀释了而且很多工具和当前任务完全无关属于纯噪音。我们做的第一件事是把工具描述精简。比如query_orders的描述从按用户ID查询订单列表支持按时间范围过滤返回订单ID、金额、状态、创建时间四个字段压缩成按用户ID查订单列表。第二件事是引入动态加载结合 Agent 的当前意图只加载相关的连接器。比如用户聊的是订单查询就只加载 order-db 和 crm-api其他连接器统统不加载。这两步做完tools 的 token 占用从 6000 降到 1800 左右单次交互的 token 费用大概降了 30%模型的工具调用准确率反而明显上升。这个反向指标很有意思——工具不是越多越好而是越精准越好。5.2 非幂等操作与重试风暴踩的第二个大坑和 Agent 的超时重试有关。有一个场景是客服 Agent 帮用户创建售后工单连接器的create_ticket操作调用比较慢第一次调用其实已经在业务系统里成功建单了但响应超时了模型认为失败了就自动重试结果一个用户投诉被创建了三条工单。这个问题本质上是因为 create_ticket 不是幂等的。解决方案分两步第一步在 manifest 里对每个操作显式声明idempotent字段。只读操作天然幂等可以直接标记 true写操作要看业务系统支不支持幂等键。第二步对非幂等操作强制要求调用方传入idempotency_key这个 key 可以由 Agent 在发起触达时生成比如trace_id 操作名 参数哈希的组合。网关在接到请求后先查一下这个 key 有没有被处理过如果处理过就直接返回上次的结果不再真正执行连接器。这一步的代码逻辑是这样class IdempotencyMiddleware: def __init__(self, redis: Redis): self._redis redis async def guard(self, operation: str, params: dict) - ConnectorResult | None: op_meta get_operation_meta(operation) if op_meta.idempotent: return None key params.get(idempotency_key) if not key: raise InvalidParameter(idempotency_key required for non-idempotent operation) cached await self._redis.get(fidem:{key}) if cached: return ConnectorResult.from_json(cached) return None async def record(self, operation: str, params: dict, result: ConnectorResult): op_meta get_operation_meta(operation) if op_meta.idempotent: return key params.get(idempotency_key) await self._redis.set(fidem:{key}, result.to_json(), ex3600 * 24)5.3 返回结果需要强约束第三个坑是连接器返回结果太野直接把数据库里的原始类型扔给模型。比如 PostgreSQL 里金额是DecimalJSON 序列化之后变成字符串模型一看到amount: 100.50可能就当作文本来回答或者日期时间格式不一致模型也会犯迷糊。我们在连接器层做了一层输出规范化统一约束返回结构数值类型统一转成 float 或 int时间类型统一输出 ISO 8601 格式字符串长文本字段超过 200 个字符截断并加truncated标记空字段直接去掉不返回 null省 context。模型的 token 预算很宝贵给它的结果越结构化、越紧凑它处理得就越准。这条经验后来我们也分享给了很多做 Agent 的朋友大家反馈一致省 token 就是省钱清晰的 schema 就是降低错误率。5.4 超时治理不能无限等做 Agent 和做普通 API 不一样普通 API 超时了用户等几秒重试就是Agent 超时了模型会纠结会重试会编造答案甚至会把一次超时理解成这个操作有问题。我们把超时分成三层连接层超时TCP/Socket 连接、单次操作超时根据 manifest 里的timeout_ms、全局网关超时。真实场景里这三个要分开配置之前图省事只设了一个网关全局超时 10 秒结果遇到一个慢查询Agent 已经等了 8 秒其中 7 秒都花在 TCP 连接和排队上真正执行只用了不到 1 秒。后来把连接层超时设为 2 秒操作层每条操作独立设置网关兜底 30 秒问题才解决。超时之后还要有降级策略。我们最常用的降级是把触达失败变成一个普通消息回给模型前置一段提示触达失败原因连接池已满。请告知用户稍后重试不要重复尝试。这样模型就不会陷入疯狂重试的循环。6. 后续演进方向与几点个人体会6.1 向协议生态演进Agent-Reach 目前是一个自研框架但我们对它未来的定位是协议优先的生态。连接器的 manifest 格式如果和社区主流的协议兼容就能天然借用整个社区已经写好的大量连接器不用自己从零维护。我们在设计 manifest 的时候刻意保持了工具的独立性它不绑定任何特定的大模型品牌而是提供标准 schema 描述再由适配层把 schema 转成各家模型需要的工具格式。6.2 降低接入门槛SDK 与脚手架代理系统的成败很大程度上取决于接入成本。如果让每个业务团队都从连接器基类开始写 manifest、写执行器、写健康检查他们会非常排斥。我们的计划是提供一个脚手架工具执行一条命令就能生成一个连接器项目的模板业务团队只需要填充自己的业务逻辑然后跑测试用例、提交注册。另外还在做可视化调试面板输入一个操作名直接测试连接器的返回结果真实请求和响应一眼可见。这个面板看着简单实际效果比写一堆文档强多了。6.3 做 Agent 触达最大的收获做 Agent-Reach 这段时间我最大的体会其实是Agent 落地最难的部分不在模型也不在连接器的代码而在治理。你有再多会调用工具的 Agent没有受控的权限边界、没有审计、没有幂等保护它们本质上只是碰运气反之把触达链路管住了Agent 反而能稳健地创造价值。还有一点是关于少即是多。我们在项目推进过程中有段时间沉迷于添加各种连接器数据库、CRM、IM、工单恨不得把公司所有系统都接进来。结果 Agent 的能力列表越来越长模型越来越糊涂。后来砍掉一半低频连接器只保留跟核心业务强相关的几个整个系统一下子就顺了。触达能力不是越宽越好而是越准确越好。最后分享一个建议如果你想在自己的项目里引入类似 Agent-Reach 的能力从最小的范围开始。先只接一个数据库只定义三个操作只让一个客服 Agent 用起来跑通完整链路后再逐步扩大。这个策略听起来保守实际上比一开始就铺开做要快得多。系统上线第一天就出事和系统上线第一个月没事带来的信心完全不一样。