做 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( modelgpt-4o-mini, messages[ {role: system, content: system_prompt}, {role: user, content: user_task} ], temperature0, response_format{type: json_object} ) return json.loads(response.choices[0].message.content)[actions]这里的关键是temperature0和 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, threshold0.3, window_size100): 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 先在这个环境里把小任务跑通再放回生产。这个小工具帮我避掉至少一半的早期线上事故。
做 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( modelgpt-4o-mini, messages[ {role: system, content: system_prompt}, {role: user, content: user_task} ], temperature0, response_format{type: json_object} ) return json.loads(response.choices[0].message.content)[actions]这里的关键是temperature0和 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, threshold0.3, window_size100): 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 先在这个环境里把小任务跑通再放回生产。这个小工具帮我避掉至少一半的早期线上事故。
