做 Agent 项目久了，我发现一个挺有意思的现象：大家最热衷讨论的是提示词怎么设计、模型选哪个、上下文窗口开多大，可真到落地的时候，卡住我们的从来不是模型智商，而是那个 Agent 能不能真的触达它该碰的东西。我的项目起名 Agent-Reach，直译过来就是“智能体的触达系统”。它解决的问题只有一个：让 LLM 智能体在执行多步骤任务时，能够稳定地碰得到外部工具、内部数据、远程服务，以及别的智能体，并且碰完之后还能拿回一个可信的结果。这听起来不像论文里那些炫酷概念，但恰恰是这类脏活累活，决定了你的 Agent 是停留在 Demo 阶段，还是能真正跑在业务流程里。
这篇文章我打算把 Agent-Reach 的设计思路、核心代码骨架、实测踩坑和一些边界场景的处理原原本本写出来。适合谁看？正在做 AI Agent、RPA 替代、内部自动化工具的开发者，以及准备把大模型接入生产系统的技术负责人。内容偏工程实践，不聊玄学，尽量让每个人看完都能照着搭一套自己的触达层。
1. 为什么现在的 Agent 普遍“够不着”：触达问题的根源
先说一个我在多个项目里反复遇到的场景。你给 Agent 配了最新的模型，提示词里写清楚了“帮我查一下订单状态，如果超时未发货就发一封催发货邮件”，模型也回答得很好，步骤清晰、逻辑正确。但真正跑起来的时候，第一步就断了：查订单需要访问内部订单库，而 Agent 根本没有数据库连接；就算连上了，查询结果的格式也不是模型熟悉的 JSON，是一坨密密麻麻的日志；就算格式没问题，发邮件这个动作还需要走公司邮件网关的二次认证。每一步都是“够不着”。
我把这种问题归为三类，每一类都有不同的解法。
1.1 触达的三个层次：读、写、执行
第一个层次是读触达。Agent 能不能拿到它决策所需的信息，包括本地文件、数据库查询结果、API 返回、网页内容。很多项目卡在这一层，不是因为没给 API key，而是没有把信息的“获取路径”告诉 Agent，或者返回的数据太原始，模型根本嚼不动。
第二个层次是写触达。Agent 能不能把结果保存下来，能不能修改某个配置，能不能向外部系统提交一条记录。写操作比读操作危险得多，因为一旦写错就是真实影响。我在 Agent-Reach 里对写操作做了严格的动作注册，而不是让模型自由发挥。
第三个层次是执行触达。这不仅是读写数据，而是触发一个完整业务流程，比如调用部署脚本、发起支付、发送通知。执行触达的难点在于：它不是幂等的，而且往往需要确认、回滚、审计。模型天然不擅长处理这类事情，所以触达层必须用代码兜底，把“执行”封装成一个个带语义的动作单元，而不是把裸命令交给模型。
1.2 能力幻觉：模型以为它会，其实它不会
我在测试一个内部运营助手的时候，让 Agent 去调整一个营销活动的预算。模型非常自信地告诉我“已成功将预算从 5000 调整到 8000”，但实际上它根本没有调用任何修改接口，它只是根据对话历史里的数字“推断”出了一个结果，然后编造了执行过程。这就是典型的能力幻觉——模型不是故意撒谎，它是在补全一个它认为合理的答案。
解决这个问题的唯一思路，就是让“执行”这件事脱离模型的自由发挥。Agent-Reach 的做法是：模型只负责产出动作序列，真正的执行由触达层完成，并且执行结果必须以结构化日志的形式回写到状态存储里。模型说“我做完了”不算数，状态存储里查到执行记录才算数。这个原则贯穿了整个项目的设计。
1.3 上下文窗口很贵，触达层帮 Agent 省着用
还有一个容易忽略的点：触达方式会直接影响上下文消耗。如果你把一个完整 API 返回的几千行 JSON 塞给模型，上下文窗口很快就被吃光了，模型反而抓不住重点。好的触达层应该做信息压缩。比如查订单，触达层可以先在代码里做字段过滤，只把订单号、状态、金额、时间这几个关键字段拼成一行摘要给模型。这就像给领导汇报工作，先给结论，再附明细，不能把原始流水直接甩上去。
2. Agent-Reach 的骨架：一个可解释的触达引擎
明确了问题之后，我设计的 Agent-Reach 不是一个模型，也不是一个插件，而是一层位于 Agent 和外部世界之间的“触达引擎”。它的核心组件有三块：ReachEngine、ActionBus、StateStore。这三个东西各干各的活，合在一起形成一条完整的触达流水线。
2.1 ReachEngine：干活的总指挥
ReachEngine 是触达层的调度核心。它接收来自 Agent 的意图，解析成一系列待执行的动作，然后按照依赖关系排队、分发、回收结果。它不关心模型是谁，不关心动作是谁实现的，只关心“这个动作能不能做、做了没有、结果是什么”。
我在设计的时候刻意让 ReachEngine 保持无状态，所有运行时数据都存在 StateStore 里。这样做的直接好处是：引擎可以水平扩展，多个请求可以在不同实例上并行处理，就算某个实例挂了，新的实例可以从 StateStore 恢复上下文，不会中断执行。
# ReachEngine 的简化骨架 class ReachEngine: def __init__(self, action_bus, state_store): self.action_bus = action_bus self.state_store = state_store def submit(self, session_id, plan): # 把 Agent 产出的动作计划登记到状态存储 for step in plan: self.state_store.append_action(session_id, step) return self._execute(session_id) def _execute(self, session_id): pending = self.state_store.get_pending_actions(session_id) results = [] for action in pending: result = self.action_bus.dispatch(action) self.state_store.mark_done(session_id, action.id, result) results.append(result) return results这段代码虽然简单，但它定了三条规矩：动作必须预先登记，执行必须经过 ActionBus，结果必须回写 StateStore。这三条规矩就是防幻觉操作的全部家底。
2.2 ActionBus：动作的总线，也是安全闸门
ActionBus 是触达层和外部世界之间的唯一通道。所有动作，不管是对内查数据库、对外调 API，还是执行一段脚本，都必须经过 ActionBus 分发。ActionBus 内部维护了一个注册表，每个动作对应一个执行函数，以及它的权限级别、超时时间、重试策略、是否幂等。
我最开始犯过一个错误：让 Agent 直接返回 Python 代码片段，然后动态执行。听起来很灵活，实际上非常危险。模型生成的代码你不信任过，一旦有一次生成出rm -rf或者一个错误的循环，代价就是实打实的。后来我把这种做法彻底砍掉，改成动作注册表模式。Agent 只能在注册表里选已经封装好的动作，不能自由发挥。
注册表的动作条目长这样：
# 动作注册表条目示例 { "name": "query_order", "description": "Query order status by order_id", "handler": "handlers.order.query_order", "args_schema": { "order_id": {"type": "string", "required": True} }, "permission": "read", "timeout_ms": 15000, "retry": {"max_attempts": 2, "backoff": "exponential"}, "idempotent": True }注意两件事。第一，args_schema是必须的，它让模型知道这个动作需要什么参数，也方便触达层做校验。第二，permission标记了读写类型，在审计和权限控制时非常有用。我见过不少 Agent 框架把工具定义直接扔给模型，不做权限标记，这在生产环境是不可接受的。
2.3 StateStore：触达行为的事实来源
第三块是 StateStore，我选用的是一个简单的 PostgreSQL 表结构，存储每个会话的动作计划、执行状态、结果摘要和错误信息。为什么不用 Redis 之类的内存存储？因为动作执行需要审计追溯，一旦出了问题，你要能回答“这个动作是谁发起的、什么时候执行的、参数是什么、结果是什么”。关系型数据库在这类场景里最合适。
StateStore 的核心表设计不复杂：
CREATE TABLE reach_session ( session_id TEXT PRIMARY KEY, created_at TIMESTAMP DEFAULT NOW(), status TEXT DEFAULT 'active' ); CREATE TABLE reach_action_log ( id BIGSERIAL PRIMARY KEY, session_id TEXT NOT NULL REFERENCES reach_session(session_id), action_name TEXT NOT NULL, args JSONB NOT NULL, status TEXT NOT NULL, -- pending / done / failed / skipped result JSONB, error_message TEXT, started_at TIMESTAMP, finished_at TIMESTAMP );每次动作执行完毕，无论成功失败，都要往reach_action_log里写一条记录。这样模型可以从 StateStore 里读到“哪些动作已经完成、结果是什么、哪些还没做”，而不是靠记忆猜。这也是前文说的“模型说做完了不算数，StateStore 里有记录才算数”的落地实现。
3. 最小复现：带 Agent-Reach 跑通一个跨系统任务
理论说了一堆，接下来上一段完整的实操。我挑了一个非常典型的场景：让 Agent 查最近一周的销售数据，如果某个品类的销量下滑超过 10%，就自动在协作群里发一条提醒。这个任务涉及三个系统——数据库、指标计算、群消息 API，恰好能展示 Agent-Reach 在不同触达类型之间的切换。
3.1 准备工作：注册需要的动作
先定义三个动作：查销售数据、计算同比变化、发送群消息。前两个是读操作，最后一个是写操作。在注册表里，它们的定义分别如下。
actions = [ { "name": "fetch_sales_data", "description": "Fetch sales records for a given period and category", "handler": "handlers.sales.fetch_sales_data", "args_schema": { "start_date": {"type": "string", "format": "date"}, "end_date": {"type": "string", "format": "date"}, "category": {"type": "string", "required": True} }, "permission": "read", "timeout_ms": 10000, "idempotent": True }, { "name": "calculate_growth_rate", "description": "Calculate week-over-week growth rate for a numeric series", "handler": "handlers.metrics.calculate_growth_rate", "args_schema": { "current": {"type": "number"}, "previous": {"type": "number"} }, "permission": "read", "timeout_ms": 3000, "idempotent": True }, { "name": "send_group_message", "description": "Send a message to a specific collaboration group", "handler": "handlers.messaging.send_group_message", "args_schema": { "group_id": {"type": "string", "required": True}, "text": {"type": "string", "required": True} }, "permission": "write", "timeout_ms": 5000, "idempotent": False } ]注意calculate_growth_rate这个动作。其实完全不用让模型算数，直接在触达层算好，模型只需要确认算出来的值有没有超过阈值。这就是我说的信息压缩——模型不需要关心计算过程，只需要处理结论。
3.2 让模型产出动作计划，而不是直接执行
接下来是 Agent 的主逻辑。我给模型两个输入：任务描述和可用的动作列表。模型的任务是产出动作计划，由 ReachEngine 去执行。
from openai import OpenAI client = OpenAI() def build_plan(user_task: str, available_actions: list) -> list: system_prompt = f""" 你是一个任务规划器。你的任务是将用户的请求拆解为一系列动作。 你只能使用以下动作，禁止发明不存在的动作： {available_actions} 输出格式为 JSON 列表，每个元素包含 action_name 和 args 两个字段。 """ response = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_task} ], temperature=0, response_format={"type": "json_object"} ) return json.loads(response.choices[0].message.content)["actions"]这里的关键是temperature=0和 JSON 输出约束。规划任务不需要创造性，稳定性比什么都重要。模型输出的动作序列会被 ReachEngine 逐一验证，任何一个动作名不在注册表里都会被直接拒绝，不会进入执行环节。
3.3 执行链路完整跑一遍
假设模型产出的动作序列是：
fetch_sales_data参数是最近两周按天拆分的销售记录calculate_growth_rate参数是本周合计和上周合计send_group_message参数是提前写好的提醒文案
ReachEngine 会这样处理：先执行fetch_sales_data，拿到结果后，把本周合计和上周合计传给calculate_growth_rate，算出增长率。如果增长率为负且绝对值超过 10%，才会执行send_group_message。
send_group_message是非幂等动作，一旦发出去就会真的影响别人。所以触达层在处理这类动作时有一个额外的确认步骤：在执行前先把动作参数渲染成一个可读的文案，让负责调度的人或上层策略决定是否放行。我在很多场景里把这个确认步骤做成了人工审批，动作会在pending_confirmation状态停住，等审批通过才真正发出去。
# 非幂等动作的确认逻辑 def _requires_confirmation(self, action): return not self.action_bus.get_action(action.name).get("idempotent", True) def _execute(self, session_id): pending = self.state_store.get_pending_actions(session_id) for action in pending: if self._requires_confirmation(action): self.state_store.mark_pending_confirm(session_id, action.id) continue # 等待确认回调，不立即执行 result = self.action_bus.dispatch(action) self.state_store.mark_done(session_id, action.id, result)这个设计在跑真实业务流程时救过我很多次。模型判断“销量下滑超过 10% 需要发提醒”的逻辑没问题，但它给出的提醒文案偶尔会写得过重，直接说“请立即停止该品类的投放”，这种措辞如果没有任何人工确认就发进协作群，后果可想而知。
3.4 结果回写与循环任务
动作执行完成后，StateStore 里出现了一条完整的链路记录。后续如果其他 Agent 或者运营同事问“这个提醒是谁发的、数据依据是什么”，直接查reach_action_log就能拿到全部事实。
我在这个最小复现的基础上，还加了一个定时触发的壳：每天早上九点自动跑一次销售数据检查，生成当天的提醒。这个定时壳不是用模型实现的，而是用传统 cron 表达式控制，到点就调用 ReachEngine 的 submit 接口。工程实践告诉我，能用定时器解决的问题，不要交给 Agent 自己决定什么时候干活，后者太不可控。
4. 触达层的真实事故：权限、幻觉与重试风暴
前面讲的都是顺利的情况。真正让我把 Agent-Reach 这个项目从毛坯房建成精装房的，是一连串线上事故。我用一个章节专门说说这些事故，因为它们比任何框架设计文档都有说服力。
4.1 第一跳的崩塌：令牌过期与静默失败
第一次把 Agent-Reach 接入公司内部系统时，遇到的最诡异问题是：Agent 明明拿到了数据，但返回的结果全是旧数据。排查了半天，最后发现是内部 API 的访问令牌在凌晨过期了，触达层用的缓存重试逻辑只重试了网络连接错误，没有处理 401 认证错误。于是每次请求都静默失败，返回了上一次成功的缓存结果。这个 bug 在产品上表现为“数据延迟一天”，在日志上表现为“看起来一切正常”。
修复方法是在 ActionBus 里加了一层认证中间件，每次分发动作之前先把令牌刷新流程跑一遍。更重要的是，我在触达层统一处理了 HTTP 状态码：401 和 403 不会被当作普通错误吞掉，而是直接标记动作失败，并且在结果里附上失败原因“authentication_expired”，这样模型和人都能一眼看出问题在哪。
经验教训：触达层最可怕的不是报错，而是静默失败。所有外部调用必须显式处理认证状态，哪怕多写几行样板代码，也好过让 Agent 拿着脏数据做决策。
4.2 幻觉动作拦截：给每一个动作挂验证器
第二个事故是关于幻觉动作的。有一次 Agent 在处理一个客户退款请求，模型生成的动作计划里出现了一个我没有定义过的动作refund_external_wallet，参数也很完整。如果触达层不做校验，直接把它丢给某个模糊匹配的执行器，后果就是真的发起一笔退款。
我在 ActionBus 里加了一个动作名精确匹配校验，任何一个动作名不在注册表里的，直接拒绝并返回错误“unknown_action”，同时把这个意外记入审计日志。不要小看这个简单的校验，它是 Agent 触达外部世界的第一道安全闸门。后来我还加了一个相似度告警，如果模型输出的动作名和注册表里的某个动作非常相似（比如差一两个字母），系统会提示“你是否想说 xxx 动作”，帮助快速定位。
另外，每个动作都必须定义args_schema并做类型校验。我见过不少框架为了省事，把参数校验交给模型自动完成。但模型对“这个字段是不是必填”“这个数值应该在什么范围”这类问题的判断并不可靠。把校验下沉到触达层，用 JSON Schema 做硬校验，才是最稳妥的做法。
4.3 重试风暴：有限状态机与熔断器
第三个事故非常经典。一个外部服务突然响应变慢，超时时间设的 5 秒，实际响应要 15 秒。ActionBus 的重试逻辑一看超时，立即触发重试。由于触达层是并发处理的，同一个动作瞬间产生了 20 个并发请求打到那个慢服务上。慢服务本来只是慢，被这一波重试打进来之后直接熔断了，影响范围从一个 Agent 任务扩散到整条业务线。
这是一个典型的“好心办坏事”案例。重试本来是提高成功率的手段，但如果重试的不当，反而会制造雪崩。我在触达层引入了两个机制解决这个问题。
第一个是熔断器。记录每个外部服务的最近失败率，如果失败率超过阈值，直接打开熔断器，后续请求快速失败，不再打向那个服务。
class CircuitBreaker: def __init__(self, threshold=0.3, window_size=100): self.threshold = threshold self.window = [] # 最近 window_size 次调用的结果 def allow_request(self) -> bool: if len(self.window) < self.window_size: return True recent_failures = sum(1 for r in self.window if r == "fail") return recent_failures / len(self.window) < self.threshold def record(self, result: str): self.window.append(result) if len(self.window) > self.window_size: self.window.pop(0)第二个是重试次数硬限制。每个动作的重试次数在注册表里就定义好了，最多重试 2 次，且重试必须使用退避策略，比如第一次等 1 秒，第二次等 4 秒。把重试策略从代码逻辑挪到配置层之后，出问题的概率大大降低。这也是一个原则：触达参数应该显式配置，而不是在代码里随意硬编码。
4.4 触达失败后的“模型补救”是否可靠？
还有一次比较有意思。某个动作执行失败后，我一度尝试让模型自己分析失败原因并生成补救方案。实验下来效果非常不稳定。模型对“网络超时”这种失败能给出合理的补做建议，但遇到“权限不足”“参数格式错误”这类问题，模型经常给出“再试一次”的无效建议，甚至会提出一个绕过权限控制的激进方案。从合规和工程两个角度，我都不建议让模型自由决定补救策略。更稳妥的做法是：失败后先把错误分类，如果是可重试的（网络闪断、服务暂时不可用），走退避重试；如果是不可重试的（权限、参数、业务规则），直接返回给上层做人工处理，不要给模型自由发挥的空间。
5. Agent 之间的触达：多智能体协作的边角料
拆完单个 Agent 的触达问题，最后聊聊多智能体之间的触达。这个方向现在很热，很多团队都在尝试让几个 Agent 分工协作，比如一个负责分析数据，一个负责写文案，一个负责审核。Agent-Reach 在这个层面的实践，一句话总结：先解决通信，再解决分工，最后解决冲突。
5.1 消息通道与地址问题：Agent 如何找到彼此
多 Agent 协作的第一步，是让 Agent 能互相发消息。我采用的方案是给每个 Agent 实例分配一个逻辑地址，格式类似agent://sales_analysis/v1，消息通过一个内部消息队列转发。发消息的动作也注册在 ActionBus 里，像一个普通动作一样被调用。
这里踩过的一个坑是消息格式。最开始我给 Agent 们发的是自然语言消息，比如“请分析最近一周的销售趋势”。模型之间互相理解自然语言没有大问题，但当协作链路过长、消息经过多轮转发之后，自然语言的语义损耗非常明显。一个 Agent 转述另一个 Agent 的结论时，很容易漏掉关键数据，甚至添油加醋。
后来我强制要求消息走结构化格式：每条消息携带sender、receiver、message_type、payload四个字段，其中payload按消息类型约定为 JSON 对象。比如数据分析 Agent 给文案 Agent 发送的结果消息，payload 里固定包含data_summary和confidence字段。这样每个 Agent 接收消息时，不需要重新理解自然语言，直接按字段取值，准确率高得多。
{ "sender": "agent://data_analysis/v1", "receiver": "agent://content_writer/v1", "message_type": "data_report", "payload": { "data_summary": "智能家居品类销售额环比下滑12.3%", "confidence": 0.87 } }5.2 冲突仲裁：给每个协作会话加一个主持人
多 Agent 协作比单 Agent 复杂，最大的原因不在技术，在于冲突。两个 Agent 各有各的目标，A 想最大化营销曝光，B 想控制预算，它们生成的方案天然是矛盾的。如果任由它们互相讨论，理论上模型可能通过对话达成一致，但实践中经常陷入各说各话的僵局，甚至 A 说服 B、B 又说服 A 的循环里。
我在 Agent-Reach 的协作模式里加入了一个仲裁者组件。仲裁者不对具体业务内容做判断，它只负责三件事：
- 定义协作会话的目标和约束条件（比如“预算上限 5 万”“必须在一小时内完成”）
- 让每个 Agent 提交自己的方案，但不允许直接互相质疑，而是提交给仲裁者做汇总
- 在必要的时候直接终止讨论，指定某个方案作为最终输出
这个设计借鉴了一个很朴素的会议原则：每场会都要有主持人，主持人不参与讨论，只负责控场。把主持人从模型人设中剥离出来，变成一个独立的程序组件，能够避免很多模型之间的“礼貌性废话”。
5.3 对外触达的留痕与可解释性
最后聊聊合规感。多 Agent 协作一旦对外发出动作，审计的复杂度会成倍上升。比如两个 Agent 协作完成了一个对外报价的生成，最终报价单发出的动作可能由第三个 Agent 执行。这时候如果出了问题，你要回答的是“整条链路中每一步是谁做的、依据是什么”。
Agent-Reach 的解决办法简单粗暴：每一条对外动作都记录完整的推理链路 ID。这个 ID 串起了发起 Agent、消息传递记录、最终执行动作和参数。产品端的表现就是，任何一个外部动作都可以一键追溯到完整链路，点开可以看到每一轮的模型输入和输出。这个能力在客户质问“为什么发错了”的时候是救命的。
写在最后
Agent-Reach 做到现在，我的体感是：它其实不是一个“很科技”的项目，反而更像一个经验沉淀库。AI 模型负责聪明的部分，触达层负责把聪明变成可靠的结果。每次踩坑之后，我都在触达层里加一条规则，现在这个项目的很多核心逻辑都不是模型生成的，而是被真实事故逼出来的。
如果你也在做类似的 Agent 项目，我的建议是从最小闭环开始：先让 Agent 能稳定地做一个读动作和一个写动作，把注册表、状态存储、审计日志这三样东西立起来。然后再慢慢加动作、加 Agent、加协作。不要一开始就追求花哨的多智能体编排，否则你会在触达层的手脚都没长齐的情况下，先被一堆意外事故淹没。最后分享一个实操小技巧：给你的触达层加一个“演练模式”，在这个模式里所有写动作都被替换成模拟实现，不产生真实外部影响，让 Agent 先在这个环境里把小任务跑通，再放回生产。这个小工具帮我避掉至少一半的早期线上事故。
