做 AI agent 最容易产生的一个错觉是：模型什么都能做。我动手写 Agent-Reach，就是因为被这个错觉坑过。等项目接进真实业务才发现，绝大多数 agent 卡在最后那一公里——它推理出来了答案，却摸不到任何真实世界里的东西。发不了邮件，建不了工单，调不了内部系统，更别说把任务交接给另一个 agent。Agent-Reach 就是为解决这个"能思考但够不着"的问题写的：它是一层极薄的连接层，把 agent 的输出能力向外延伸，通过统一的连接器、路由和权限策略，让 agent 真正"触达"外部系统、其他 agent 和人。
这篇文章会讲清楚我做这个项目的完整思路：先想明白"能力半径"到底是什么，再拆架构，然后讲最小可用版本怎么跑起来，最后是跑通 demo 之后踩到的三个坑和完整的排查链路。适合正在做 agent 应用、尤其是想把原型推上生产的开发者，也适合想搞清楚"function calling 之外到底还缺什么"的朋友。
1. 动手前先想清楚："能力半径"到底是什么问题
近一年我见过太多 agent 项目卡在一个非常尴尬的阶段：demo 里什么都会，接到真实业务里就废掉。原因不是模型不够强，而是 agent 的能力边界被框死在聊天窗口里。我把 agent 实际能触达的外部世界范围叫做"能力半径"——它能调哪些系统、能通知哪些人、能操作哪些数据。模型自身的推理能力决定它的智商，能力半径决定它的价值。
我最早踩到这个点，是在给一个内部服务台做支持机器人原型的时候。它跑在办公群聊里，能准确回答知识库问题，还能根据历史工单给出处理建议。可真正用起来的第一周，大家就烦了："你能帮我建个工单吗？""你能把这个问题转给值班同学吗？"——它什么都答应，但什么都做不了。因为当时我只给它接了模型和检索，没有给它接任何行动通道。
你可能会说，用 function calling 不就行了？对，function calling 能解决"调用函数"的问题，但它解决不了"触达系统"的问题。两者看起来像，骨子里差很远。往下我会专门对比。
1.1 为什么是"Reach"而不是"Tool"
把 function calling 比作给 agent 一双手，它只能拿起来眼前的东西。Tool 是"动作"，Reach 是"通道"——我们不只关心 agent 能不能调用某个函数，还关心这次调用怎么被允许、怎么送达、失败了怎么办、有没有留痕。我列了一张表，方便你理解两者的差异：
| 对比项 | Function Calling | Agent-Reach 的 Reach 层 |
|---|---|---|
| 核心问题 | 模型如何生成结构化调用参数 | 调用如何获准、送达、审计 |
| 失败处理 | 通常交给应用自己兜底 | 统一超时、重试、熔断、降级 |
| 权限粒度 | 大多按函数名硬编码 | 按目标 + 动作 + 内容动态决策 |
| 可观测性 | 靠日志零散排查 | 每次触达都有标准审计记录 |
简单说，工具层解决的是"能不能调用"，Reach 层解决的是"该不该调用、怎么安全到达"。Agent-Reach 做的是后者。这也是为什么我只留了极薄的抽象层：它不负责让模型更聪明，只负责让模型的动作安全地出得去。
顺便说一句，也有人在群里问我：那为什么不干脆写脚本调 API？脚本的问题是，目标、参数、时机都是运行时才决定的，agent 在执行过程中要根据环境变化现场决策，你不可能事先把所有分支写成脚本。这本质上是一次"动态编排"问题，不是自动化问题。
1.2 三个必须覆盖的目标类型
设计初期，我把 agent 需要触达的目标收敛成三类：
- 外部系统与 API：建工单、查库存、发消息，这类最高频，几乎每个 agent 都躲不开。
- 其他 agent：任务需要拆分给多个专用 agent 时，需要一个标准的 agent 间交接协议。
- 人：通知、审批、复核。机器做决定之前，很多动作需要真人确认。
为什么三套场景一定要放进同一个框架？因为如果每个目标各自单独接，权限策略、审计日志、重试逻辑就会各写一套，越到后面越失控。统一抽象是所有工程化 agent 的必经之路，这个判断在我接完第二个连接器之后就彻底验证了。
2. 架构拆解：连接器、路由与策略边界，谁该听谁的
Agent-Reach 的代码量不大，强在结构，不在功能。整体分三层，各管一件事：
- 连接器层（Connector）：吃掉不同外部服务的差异，向上暴露统一接口。
- 路由层（Router）：决定一条触达请求该送到哪个连接器。
- 策略层（Policy）：决定这条请求允不允许发、要不要审批、限不限制频率。
这三层最关键的约束是：路由和策略必须分离。原因是职责边界——路由是纯技术问题（发到哪），策略是治理问题（允不允许、谁能发、花多大代价）。混在一起写，改权限就得动代码，大范围部署时完全不可接受。
2.1 统一消息模型
所有触达请求，不论目标是 API 还是人，都统一成一个消息对象：
# agent_reach/model.py from dataclasses import dataclass, field from typing import Any @dataclass class ReachMessage: target: str # 目标标识，如 "jira"、"email"、"agent.ops" action: str # 动作，如 "create_ticket"、"send" payload: dict[str, Any] # 载荷，具体内容 timeout: float = 10.0 # 单次超时 max_retries: int = 2 # 最大重试次数 request_id: str = "" # 链路追踪用这个模型看着简单，但每个字段都是踩过坑才定下来的。比如request_id，一开始我没加，排查问题只能靠时间戳猜是哪次触达，后来才补上。max_retries更关键，后面讲重试风暴时你就知道它有多重要。
2.2 连接器：世界上没有两个一样的 API
连接器是一个抽象基类，每个外部系统实现一个子类：
# agent_reach/connector.py from abc import ABC, abstractmethod class Connector(ABC): name: str = "" @abstractmethod async def dispatch(self, message: ReachMessage) -> dict: """把 ReachMessage 翻译成目标系统能理解的调用。""" async def health_check(self) -> bool: return True为什么必须有health_check？因为我遇到过：连接器配置错误，agent 还傻乎乎地一直往里发，所有请求全挂在超时上。有了健康检查，路由层可以在连接器不健康时自动降级，而不是继续往里堆请求。
连接器用注册表管理，支持热插拔：
# agent_reach/registry.py class Registry: def __init__(self): self._connectors: dict[str, Connector] = {} def register(self, connector: Connector) -> None: self._connectors[connector.name] = connector def get(self, name: str) -> Connector | None: return self._connectors.get(name)注册表的好处是，新增一个系统不需要改动路由和策略代码，只要注册实现类、再在配置里声明允许触达的权限就行。我后来接第三个连接器时只花了半天，大部分时间都耗在熟悉对方接口上。
还有一个很容易被忽略的细节：连接器的难点从来不在"调用 API"，而在"翻译语义"。同一个create_ticket，在 Jira 里要构造 issue 对象，在邮件里要拼一封标题带 [工单] 的邮件。这个翻译层必须放在连接器内部，不能漏到上层，否则上层代码会越写越脏。
2.3 策略层：不做权限控制的触达是灾难
策略层不需要一开始就做一个复杂的权限引擎，但有几个点必须满足：
- 目标白名单：只允许触达已注册且显式开放的目标。
- 动作级权限：同一目标下不同动作可以设置不同权限，比如"读工单"允许，"删除工单"必须审批。
- 审计日志：每次触达记录结构化日志，包含 request_id、目标、动作、调用方、结果。
- 限流：防止 agent 在异常状态下对下游造成压力。
我用一份 TOML 配置承载绝大多数策略规则：
[targets.jira] allowed_actions = ["read_ticket", "create_ticket", "comment_ticket"] max_rate = "10/min" require_approval = ["delete_ticket"]策略判定放在路由之前，判定不通过直接拒绝，不给连接器造成任何负载。
提示：权限和审计是这套框架里"看着最难、实现却不难"的部分。一张配置表加一个拦截函数，就能覆盖 80% 的场景，别上来就堆权限引擎。
3. 最小可用版本：从零到三百行跑通一次真实触达
我定的最小目标很朴实：让一个 agent 能通过两种连接器（Webhook 和邮件）完成一次"建工单"模拟操作。整个实现不到三百行代码，但五脏俱全。
3.1 先把骨架写出来
按依赖顺序依次写：模型、连接器基类、注册表、策略、路由。策略层最小形态就是一张配置表：
# agent_reach/policy.py class Policy: def __init__(self, rules: dict[str, set[str]]): # 例：{"jira": {"create_ticket", "read_ticket", "comment_ticket"}} self.rules = rules def check(self, message: ReachMessage) -> bool: allowed_actions = self.rules.get(message.target, set()) return message.action in allowed_actions路由主体长这样：
# agent_reach/router.py from agent_reach.model import ReachMessage class Router: def __init__(self, registry, policy, audit): self.registry = registry self.policy = policy self.audit = audit async def route(self, message: ReachMessage): if not self.policy.check(message): self.audit.log(message, status="denied") return {"status": "denied", "reason": "policy_denied"} connector = self.registry.get(message.target) if connector is None: self.audit.log(message, status="failed", reason="no_connector") return {"status": "failed", "reason": "unknown_target"} try: result = await connector.dispatch(message) self.audit.log(message, status="ok", result=result) return {"status": "ok", "result": result} except Exception as exc: self.audit.log(message, status="error", error=str(exc)) # 实际实现里，这里会挂上重试与熔断逻辑 return {"status": "error", "error": str(exc)}这段代码我故意写得直白，没套框架，就是想让第一次读的人一眼看穿每个环节的职责。先保证链路通，再谈优化。
3.2 让 agent 决定"发给谁"
路由输入从哪来？我的做法是"模型分类 + 规则兜底"混合策略：agent 在推理时产出结构化触达意图，包含 target 和 action；但考虑到模型偶尔会输出不存在的目标名，路由层再做一层相似度匹配，命中不了就返回候选名单让 agent 澄清。给模型的意图识别提示词大致长这样：
你正在使用 Agent-Reach 触达外部系统。请基于用户需求输出 JSON： {"target": "目标系统标识", "action": "动作", "payload": {...}} 可选目标：jira(工单系统)、email(邮件)、agent.ops(运维专用agent) 规则：如果用户请求创建工单，target 必须为 jira，action 为 create_ticket。关键点是把可选目标列表直接喂给模型，而不是让它自由发挥。目标是封闭集合，动作也尽量是封闭集合，模型在这个约束下输出稳定得多。实测下来，把候选列表写进提示词之后，目标识别的准确率从七成提升到九成以上，剩下的交给模糊匹配兜底。
3.3 用两个真实连接器验证
Webhook 连接器最简单，一个 HTTP POST 就完事：
# agent_reach/connectors/webhook.py import httpx from agent_reach.connector import Connector from agent_reach.model import ReachMessage class WebhookConnector(Connector): name = "webhook" def __init__(self, url: str, token: str): self.url = url self.headers = {"Authorization": f"Bearer {token}"} async def dispatch(self, message: ReachMessage) -> dict: async with httpx.AsyncClient(timeout=message.timeout) as client: resp = await client.post(self.url, json=message.payload, headers=self.headers) resp.raise_for_status() return resp.json()邮件连接器稍微麻烦一点，要处理收件人、标题、正文，核心逻辑还是同一个：把 payload 翻译成邮件，发送，返回结果。我在实现时特意把凭证从系统变量里取，而不是从对话里抓，这个设计在后面救了我一命。
提示：验证最小版本的时候，别一上来就接真实系统。先用一个返回固定结果的 mock 连接器把路由链路测通，再接真实服务。这样出了问题，你至少能确认问题出在连接器还是出在路由层。
4. 跑通 demo 后踩到的三个坑与完整排查链路
Demo 跑通只是开始，接上真实业务，坑才开始排队。三个印象最深的坑，每一个我都按完整的排查链路讲：现象、查证、根因、修复、验证。
4.1 重试风暴：agent 失败后的"倔强"
现象：下游服务有一次短暂抖动，结果那一小时监控显示，同一个工单创建请求被重复发送了四十多次。查日志发现，agent 每次收到失败结果，就会"再试一次"，而且它还会重新推理、重新生成参数，每次都生成新的 request_id，所以我一开始根本没发现是同一起事。
根因有两层：一层是模型层的重试是"语义级重试"，它不知道下游可能已经处理了一部分请求；另一层是我的路由层没有幂等控制，也没有熔断。修复方案一分为三：
- 对同一语义动作，在策略层生成幂等键，下游按幂等键去重；
- 重试必须指数退避加抖动，不能固定间隔；
- 加熔断逻辑：连接器连续失败达到阈值，直接拒绝后续请求，降级为"已记录，稍后处理"。
import random def next_retry_delay(attempt: int, base: float = 1.0, cap: float = 30.0) -> float: exp_backoff = min(base * (2 ** attempt), cap) return exp_backoff + random.uniform(0, 0.3 * exp_backoff) # 加抖动，避免同步重试这个坑给我的教训很朴素：不要相信 agent 会"克制"，它只会按照策略行事。重试和熔断必须落在基础设施层，而不是交给模型自己决定。
4.2 幻觉目标：agent 把请求发给了不存在的目标
第二个坑是 target 字段的幻觉。模型按提示词应该输出 jira，某次它输出的是 "jirra"（拼错了），路由直接返回 unknown_target。agent 收到错误后又重新推理，这次输出了 "ticket-system"——更离谱。一来一回，用户在群里看到的就是 agent 在循环道歉。
排查过程：先在审计日志里看到 status=failed、reason=unknown_target 的记录，确认是路由层拒绝的；然后把 agent 的原始输出拉出来看，发现是模型的拼写错误加别名自由发挥：
10:03:11 status=failed reason=unknown_target target=jirra action=create_ticket 10:03:12 status=failed reason=unknown_target target=ticket-system action=create_ticket修复：路由层加了目标名模糊匹配，编辑距离低于阈值就自动纠正并记录；同时对金额操作、删除操作这类高成本动作，即使匹配成功也要二次确认。这就是前面说"目标封闭集合"的真正价值——模型不需要理解所有系统，它只需要在约束里选。
4.3 凭证泄漏：上下文把密钥带进了载荷
这个坑最隐蔽，也最该警惕。某次邮件连接器发出的工单内容里，出现了一段 API Key。排查链路：先看到审计日志里 payload 异常，再顺藤摸瓜发现，agent 在对话里"看过"这个密钥，然后把它当成普通文本塞进了要发送的工单描述里。
根因：模型的上下文是复制粘贴式的，它不区分"秘密"和"普通信息"。修复分两层：
- 凭证引用表：连接器需要的密钥只从系统变量读取，绝不从对话内容里取；
- 脱敏过滤器：所有出去的 payload 统一过一遍模式匹配，把疑似密钥、令牌、手机号替换成占位符。
这个坑让我明白，Reach 层不只是"把消息送出去"，它还是信息流的边界。所有要离开系统边界的数据，都要经过一道独立检查，且这道检查绝不能依赖 agent 的推理逻辑。
5. 如果继续做，我会先改这四个点
Agent-Reach 第一版只算验证了方向。有几个明确的改进方向，按优先级排序：
第一是投递语义。现在连接器失败后只记录错误，没有"至少一次"的投递保障。真正生产环境里，"工单到底建没建"是不能含糊的。我会加一个本地持久化队列，先把 ReachMessage 落盘，再由调度器异步投递，成功后才标记完成。
第二是人工审批流。前面提到高成本动作需要确认，第一版只做了"拒绝"，没做"暂缓并通知人"。下一版会给 ReachMessage 加一个approval_required字段，命中策略后转入审批队列，负责人审核通过才真正投递。
第三是测试策略。连接器层建议用 mock 连接器做集成测试：随机返回失败、随机延迟、随机乱序。agent 的不可控性，必须用可控的故障注入来对抗。我在自己项目里加了一组故障注入连接器，才把重试和熔断逻辑测稳。
第四是模型无关。Agent-Reach 本身不依赖具体模型，只要 agent 能输出结构化的 ReachMessage，Reach 层就不用动。这样换模型、换推理框架都影响不到连接通道。
最后说一点个人体会。写 agent 项目最容易犯的错，是一上来就追模型能力。但模型能力再强，也需要有人给它建通道、设边界、记日志。Agent-Reach 这个名字就是在提醒我自己：agent 的价值不在"想"得多漂亮，而在"够"得着多远。把通道建好，边界设好，剩下的事情自然顺。现在我的新项目里，所有 agent 的第一行配置，永远是那份目标白名单。
