Agent-Reach这个名字是我去年年底一个项目跑崩之后想出来的。当时我做了一套基于大模型的工作流机器人模型本身已经能理解用户的模糊意图逻辑链路也没问题但到了真正要去改工单状态、去查数据库、去给群机器人发消息的时候它什么都够不着——代码在调用外部系统的那一刻频繁出错要么权限不对要么字段对不上要么返回的数据把上下文撑爆。我一度以为是模型能力不够换了好几个型号问题照旧。后来才意识到错不在大脑在手脚。这篇文章就围绕Agent-Reach展开完整记录我是怎么从失败现场出发重新设计了一个能让Agent真正触达业务系统的接入架构以及落地过程中的教训和优化。如果你也在做大模型Agent应用发现自己的Agent能聊不能干这篇文章应该能帮上忙。1. 第一个版本能聊不能干彻底暴露了Agent的触达短板1.1 当时的项目背景与失败现场项目最初的目标很简单给运维团队做一个能自动处理客户工单的机器人。客户在IM群里说一句我们线上有个订单一直没同步麻烦看一下Agent要能理解这句话然后去工单系统里查这个订单相关的工单更新状态并且在群里回复处理结果。听起来没什么难度对吧我当时也是这么想的。第一版直接给大模型接了一套function calling定义了好几个函数比如查询工单更新工单状态添加工单备注。模型在对话中会自己判断该调用哪个函数并且生成正确的参数。说实话在纯对话测试里这一套表现非常好模型几乎每次都能选出正确的函数、填出合理的参数。结果一上真实环境就崩了。工单系统不是只有查询工单一个接口它有几十个接口每个接口有不同的认证方式、字段命名规范、分页逻辑、权限范围。我需要给模型暴露一个统一的工具但这个工具内部要把几十个接口的差异全部消化掉。一开始我图省事把这堆接口直接包装成了几个粗粒度的大函数比如查询工单详情这个函数内部自动关联查询多个子接口。跑了几轮之后发现模型时不时会拿着工单号却传进了客户ID的位置或者查询参数超出了一个工单系统的枚举范围接口直接返回400。那段时间我每天的工作变成了教模型怎么正确调用API提示词里写了各种注意xxx字段必须是数字之类的补充说明。结果模型在对话环境里表现依然很好一接到真实请求就犯各种低级错误。后来我冷静下来发现问题的根源不在模型理解能力上——模型确实理解要做什么但它对外部系统的真实约束一无所知。1.2 Function Calling 只解决了想得到没解决够得着很多人做大模型Agent都有个误解只要接入了function callingAgent就能操作外部系统了。这个误解本身也曾经是我的误解。function calling解决的是模型决定调用一个工具这一步它让模型能把自然语言意图转化为一次结构化的函数调用请求。但是函数真正被执行的时候后面还有一整个外部系统的复杂性在等着要不要鉴权这个操作有没有写权限返回的字段哪些是模型需要的哪些字段包含大量冗余信息接口出错时错误信息能不能被模型看懂我把这些统称为触达问题。模型想得到要做什么但它的手够不到业务系统。你给它一把钥匙它不知道这把钥匙开哪扇门你把门打开了它又不知道该拿屋里哪样东西。想通这一点之后我把题目从优化提示词让模型更听话改成了设计一个能让模型安全、高效触达外部系统的中间层。这正是Agent-Reach的起点。1.3 我把项目改名为Agent-Reach的契机改名这件事挺突然的。那天晚上我在纸上画了三个框模型层、工具层、业务系统层。模型层负责理解工具层负责把模型意图映射成真实API调用。我盯着工具层这三个字看了很久意识到它不应该只是简单的API聚合而应该是一层带有语义翻译性质的能力封装——它决定了一个Agent到底能触及哪些系统、以什么方式触及、以多高的效率触及。于是我把项目代号改成了Agent-Reach。Reach这个英文词我觉得比接入集成精准得多。接入只是把线连上Reach还包含够得着覆盖得到的意味。后来的所有调整都围绕如何让Agent触达得更准、更可控、更便宜这三个词展开。2. Agent-Reach的三层结构每层解决一类具体问题2.1 连接器层把外部系统的复杂性消化在边界之内Agent-Reach的第一个核心设计是给每个外部系统配备一个独立的连接器Connector。连接器不是一个函数更不是直接暴露给模型看的工具它是一个语义翻译器。举个例子。工单系统真实API里有PATCH /tickets/{id}这种接口请求体有一堆字段其中有个字段叫hardware_order_status代表硬件订单的流转状态。如果让模型直接面对这种字段它根本不知道要不要传、传什么值因为自然语言里几乎没人会说硬件订单流转状态。连接器要做的事情是把这一坨复杂系统翻译成一小段模型听得懂的业务语言。我在连接器层定义了五个操作原语查询工单、变更工单状态、追加处理记录、分配处理人、查询工单历史。原语的名字必须使用业务人员能理解的动词而不是技术接口的名称。下面是我当时为工单系统写的连接器基类和实现代码是Python写的# connector_base.py from abc import ABC, abstractmethod class BaseConnector(ABC): 所有连接器必须实现统一入口 reach(). def __init__(self, config): self.name config[name] self.base_url config[base_url] self.token config[token] self.timeout config.get(timeout, 5) self.read_only config.get(read_only, False) abstractmethod def reach(self, operation: str, params: dict): 统一触达入口。operation 为业务原语params 为业务参数。 返回数据必须经过 ContextGate 裁剪。 pass def _check_write_permission(self, operation: str): if self.read_only and operation not in self.READ_ONLY_OPERATIONS: raise PermissionError(f当前连接器为只读不允许执行 {operation})每个业务系统继承这个基类内部再去调用各自的SDK或者HttpClient。模型的prompt里完全不出现PATCH /tickets/{id}它只看得见变更工单状态这五个字和少量参数说明。API再怎么变连接器内部改掉就行模型侧完全不受影响。连接器还有一个很重要的边界每个连接器只暴露极少数的操作原语。我之前吃过亏觉得功能越全越好把系统的几十个接口都暴露给了模型。结果是模型陷入选择困难而且每次生成参数时可选范围太大容易填错。收敛到5到10个原语之后模型的调用准确率明显上升。2.2 上下文闸门防止Agent一次触达就烧掉几万token第一版系统还有一个让我肉疼的问题token消耗非常夸张。原因很直接接口返回的数据模型根本消化不了那么多。工单详情接口一次会返回46个字段从流水号到内部审核代码再到各种时间戳总文本量三四千字。模型拿到这些数据后会把整段内容都塞进上下文再在这个基础上做后续推理。一次查询就是几千token一个复杂工单流程做下来动不动就上万token。更麻烦的是模型看到大量无关字段时偶尔还会编造一些字段的含义在回答里给用户错误的信息。Agent-Reach的第二层就叫上下文闸门ContextGate。名字听起来高大上核心职责只有三个字段白名单、记录条数上限、按需加载。字段白名单好理解就是我明确告诉闸门工单实体只能返回id、title、status、priority、created_at这五个字段其他的一概不要。比如真实接口返回了46个字段闸门过滤后只给模型看5个。大多数Agent任务根本不需要全部字段。记录条数上限是对列表类接口的。查询历史记录时如果明文条数是200条闸门默认只放行最近5条。只有模型明确请求我要看完整处理历史闸门才允许加载全量而且在返回前会标记这条数据量很大之类的提示让模型谨慎使用。按需加载是我后来补的一个机制。有些字段必须给但平时没人看比如工单关联的子订单列表。闸门先把存在关联数据这个事实告诉模型但默认不加载明细等模型在推理过程中明确说需要查看子订单明细时再发起第二次触达。这个机制既省token又避免了模型被大量次要信息干扰。经过闸门裁剪后一次典型工单查询的token占用从三四千降到了四五百而且模型的回复准确率反而上升了——它不用再去分辨哪些字段是有用的。2.3 编排层多Agent协作时统一决定谁来触达、触达什么Agent-Reach的第三层是编排层。我项目后面慢慢不止一个Agent了除了工单处理Agent还有一个负责整理每日报告的报告Agent一个负责在群里做初步响应的接待Agent。如果每个Agent都自己连工单系统那权限、审计、缓存这些全都各写一套根本维护不过来。编排层做的事情很简单上游Agent把需要触达的意图告诉编排层编排层检查这个Agent有没有权限然后路由到对应的连接器去执行执行完的数据统一过闸门再返回给上游Agent。这个设计让我想到了office里唯一的打印机你不需要给每个人配一台打印机只需要进办公室的走廊里把打印任务统一交给一个插口谁有权限、谁先打、打几份都集中管理。Agent触达外部系统也是同一个道理。编排层还有一个很实用的状态管理功能磁滞开关。如果某个连接器在最近一分钟内不可用编排层不会让Agent继续傻乎乎地反复重试而是直接在返回值里告诉Agent该连接器暂不可用请稍后再试。这事之前是没人管的模型拿了一次超时错误后可能会用另一个参数又发起一次调用不仅浪费token还可能让外部系统压力变大。下面这段是编排层路由逻辑的核心代码我保留了当时的基本骨架# router.py class ReachRouter: def __init__(self): self.connectors {} # name - connector instance self.policies {} # agent_name - allowed ops def register_connector(self, connector): self.connectors[connector.name] connector def bind_agent_policy(self, agent_name: str, allowed_ops: list[str]): self.policies[agent_name] allowed_ops def route(self, agent_name: str, connector_name: str, operation: str, params: dict): # 1. 权限校验 if operation not in self.policies.get(agent_name, []): raise PermissionError(fAgent {agent_name} 无权执行 {operation}) # 2. 连接器是否存在 connector self.connectors.get(connector_name) if not connector: raise ConnectionError(f连接器 {connector_name} 未注册) # 3. 放行到连接器 raw_result connector.reach(operation, params) # 4. 上下文闸门裁剪 gated_result ContextGate.trim(raw_result, connector_name, operation) return gated_result这个结构的价值在三方各取所需模型只关心业务语义连接器只关心系统对接细节编排层只关心权限和路由。每一层都做自己擅长的事系统的复杂度就降下来了。3. 最小实现30分钟搭出一个能真正操作工单系统的Agent3.1 环境准备和最小依赖选型做完设计我把整个系统重写了一遍。这次的目标非常明确跑通一个最小闭环。我选的技术栈如下每一行都说明一下原因Python 3.10生态成熟写AI相关的东西最顺手。FastAPI不是必须的但Agent内部服务将来一定需要处理并发请求用FastAPI把触达层做成HTTP服务后面接其他Agent会很方便。Redis做语义缓存。如果只求跑通可以先用Python字典顶着。但缓存是必需的后面会专门讲。LLM API我这边用的是一家兼容常见大模型接口的厂商服务配置环境变量即可不用绑定到具体某个品牌。我不建议一上来就搞Kafka、微服务、K8s这些东西Agent-Reach的第一版就是单进程扛住测试环境的几十个并发绰绰有余。先把业务闭环跑通瓶颈在哪一目了然再谈扩展也不迟。3.2 一个可复用的连接器实现我拿工单系统做例子写一个最小版本的TicketConnector。它内部调用一个模拟的外部工单系统客户端但接口命名、参数设计都按真实情况来# ticket_connector.py import time, hashlib, httpx from connector_base import BaseConnector class TicketConnector(BaseConnector): # 只暴露模型能看到的业务原语 OPERATIONS [ query_ticket, # 查询工单状态 update_ticket_status, # 变更工单状态 add_comment # 追加处理记录 ] # 只读原语写操作必须显式声明目标状态 READ_ONLY_OPERATIONS [query_ticket] def reach(self, operation: str, params: dict): self._check_write_permission(operation) if operation query_ticket: return self._query_ticket(params) elif operation update_ticket_status: return self._update_status(params) elif operation add_comment: return self._add_comment(params) else: raise ValueError(f未知原语: {operation}) def _query_ticket(self, params): ticket_no params[ticket_no] # 真实环境这里会去调工单系统HTTP接口 resp httpx.get( f{self.base_url}/tickets/{ticket_no}, headers{Authorization: fBearer {self.token}}, timeoutself.timeout, ) resp.raise_for_status() return resp.json() def _update_status(self, params): # 关键点所有写操作都在连接器内部做幂等键 idempotency_key hashlib.sha256( f{params[ticket_no]}:{params[target_status]}.encode() ).hexdigest() resp httpx.post( f{self.base_url}/tickets/{params[ticket_no]}/status, json{target_status: params[target_status]}, headers{ Authorization: fBearer {self.token}, Idempotency-Key: idempotency_key, }, timeoutself.timeout, ) resp.raise_for_status() return {updated: True, ticket_no: params[ticket_no]}这个连接器暴露给模型的操作原语只有三个。我故意砍掉了分配处理人查历史这些操作先让路由复杂度降到最低。跑通之后再加。事实证明这个顺序是对的。3.3 把Agent调用链跑通的关键配置环境准备完毕、连接器写好之后剩下的就是把几块拼起来。我当时的启动脚本是这样组织的# main.py from ticket_connector import TicketConnector from router import ReachRouter # 1. 初始化连接器 ticket_connector TicketConnector({ name: ticket_system, base_url: http://127.0.0.1:9000, token: dev-token, timeout: 3, read_only: False, }) # 2. 注册到路由 router ReachRouter() router.register_connector(ticket_connector) # 3. 给 Agent 绑定权限 router.bind_agent_policy(ops_agent, [ query_ticket, update_ticket_status, add_comment, ])这里有一个我特别想强调的点权限绑定是Agent级别的不是角色级别的。同一个连接器可以被多个Agent使用但每个Agent能执行的原语列表是单独声明的。这样将来某个Agent因为业务需要被收窄权限时改一行代码就行不会影响到其他Agent。启动之后整个调用链条长这样用户说你们看一下单号T20240415为什么还停在处理中 。接待Agent识别意图后把请求送到编排层。编排层查到ops_agent有权调用query_ticket于是让工单连接器去查单。连接器调外部接口拿到原始数据闸门把字段裁剪成id、title、status、priority、created_at五个返回给模型。模型据此生成回复这个工单当前状态为处理中优先级是P1已经超过SLA时限。跑通这个链路的那一刻我才觉得这个项目真正成立了。4. 上线后我踩过的四个坑每一个都是真金白银换来的4.1 权限放得太宽Agent把已关闭工单全部重新打开系统上线测试环境第一周出了一件让我哭笑不得的事。有一个测试账号的Agent在执行变更工单状态时把一批状态为已关闭的历史工单全部重新置为处理中了。根因很简单我一开始给Agent赋予的权限是所有工单操作包括重新打开已关闭工单这种操作。当时我觉得模型足够聪明不会做这种糊涂事。但我低估了大语言模型对用户自然语言的跟随程度——用户说帮我看看还有哪些已关闭的工单Agent的推理链走到某一步时擅自理解了需要重新打开它们以便进一步处理。从那之后我定了一条铁律Agent的权力边界必须由代码定义不能依赖模型自己的判断力。我在连接器里加了状态机校验比如已关闭的工单不能直接变更为处理中必须先变为待重新审核。模型拿到的操作说明里也明确写了这条限制。这相当于给Agent的手脚加上物理约束而不是靠劝诫来让它懂事。4.2 没有上下文裁剪一次鏖战烧掉十几万token这是我在第2章第2小节提到的教训的完整版本。测试群里有个用户连续追问了同一批工单的多个维度Agent需要反复调用接口查询。每一次接口返回的都是全量工单详情什么都不剪直接全部塞进上下文。我后来看了日志一次十五分钟的对话模型累计消耗了十几万token其中绝大部分都是重复返回的工单历史记录、内部备注、操作日志。这个成本按当时的价格算一天跑下来完全不可接受。我加上上下文闸门之后同一类对话的token消耗降到了原来的十五分之一左右。而且闸门带来的额外好处是模型回答的质量明显变稳定了因为它不再被几十个无关字段干扰注意力全部集中在真正重要的状态、优先级、最近跟进记录上。这件事给我的启发是大模型应用的成本优化很多时候不是靠换便宜的模型而是控制输入数据的质量。数据越干净模型越省token正确率也越高。4.3 重试逻辑放错层导致同一张工单被创建了两次这个坑是我自己埋的。有一次测试环境网络抖动Agent发起创建工单请求时超时了。我当时在Agent调用层写了一个自动重试只要请求超时就原样再发一次。结果第一次请求其实已经在服务端执行成功了只是响应丢失。重试发出后工单系统里出现了两张一模一样的工单。这个问题的本质是重试和幂等没有对应起来。解决办法有两个层面第一重试逻辑要收敛在连接器内部而不是让Agent自己判断该不该重试。Agent只需要知道这次触达的结果是成功还是失败失败的详细原因由连接器决定怎么处理。第二凡是可能产生业务数据的写操作必须携带幂等键。我在前面那段代码里已经把幂等键加上了用ticket_no加目标状态做哈希同一个请求意图不管重试多少次服务端都只认第一次。当时我排查这问题花了大半天因为两张工单的ID不一样但内容完全一样怎么看都像是Agent编造了数据。后来凭时间戳对比才发现是重试导致的。这个坑埋得深写代码时根本想不到。4.4 错误信息不够模型友好Agent只能瞎猜这个问题虽然排在第四但它几乎每天都会遇到。外部系统返回的错误信息往往是这样的{code: 50001, message: SYSTEM_INTERNAL_ERROR}模型拿到这种错误完全不知道发生了什么。它会自己脑补一个原因然后用错误的原因去调整参数再试一次。多数时候越试越错。我后来给连接器加了一个错误语义化层对外部系统的错误响应做一次翻译生成模型能理解的、带下一步建议的错误结构{ error_code: TICKET_LOCKED, meaning: 该工单当前被另一个流程锁定不能执行状态变更, suggestion: 等待5分钟后再试或先通过query_ticket确认解锁状态 }模型看到这样的错误信息才知道自己应该等待还是换个操作。这项改动表面上只影响了报错逻辑实际效果却是Agent的自主修复能力提升了一个档次。在那之前我总抱怨模型自作聪明后来才明白是我没给够它判断所需的上下文。5. 从能用走向好用我给Agent-Reach加了三个增强5.1 语义缓存让高频触达成本趋近于零工单系统有个高频场景好几个Agent在短时间内都要查询同一个工单号。如果没有缓存每次查询都会穿透到外部系统既耗token也占用外部系统接口配额。Agent-Reach在编排层加了缓存。最开始的版本是简单的参数哈希缓存也就是判断同一连接器、同一操作、同一参数就直接复用上一次返回值。但实际场景中不同Agent的措辞不一样查询参数却经常一样。于是我又补了一层语义判断把Agent发来的查询意图做向量化与缓存里近三十分钟的触达请求比对相似度超过阈值的直接命中缓存。当然带语义判断的缓存要注意一个边界凡是有写操作的数据缓存必须立刻失效否则Agent会拿到旧数据做出错误决策。我的做法是在连接器更新工单状态成功后主动通知编排层把该工单号相关缓存全部删除。这个逻辑叫触达缓存一致性没做好的话缓存优化反而会变成事故放大器。从实际效果看高频查询场景的接口调用量降了一半以上token消耗也同步降了。更重要的是Agent的响应速度提升了因为缓存命中时根本不需要等外部系统返回。5.2 触达审计每一个操作都有迹可循Agent越做越聪明最大的风险反而变成了不可预测。上线第三周测试环境中出现了一次误操作我需要快速回答三个问题谁做的做了什么为什么这么做答案来自审计日志。我在编排层加了一段简单的全局记录逻辑每条触达记录包含Agent名称、连接器名称、操作原语、参数摘要、返回状态、耗时、时间戳。存储介质我就用了SQLite数据量不大单表就够。有了审计日志之后权限配置也逐渐科学起来。我看了一周审计数据发现某一类只读查询被某Agent调用了五千多次说明它的业务逻辑可能存在重复查询另一个Agent频繁触发某个写操作但经常以失败告终说明它的参数生成有系统性偏差。这些数据反过来帮助我更准确地调整Agent的提示词和闸门配置。审计这个功能平时用不上但一旦出问题它就是最快定位系统的唯一线索。做Agent项目宁可先花一个小时把日志写好也不要等出事故了再熬夜翻代码查原因。5.3 这套结构可以迁移到哪些场景Agent-Reach的核心思路我后来在好几个不同的业务场景里复用了一遍。只要是智能体要对外部系统做操作的项目三层结构基本都成立数据库查询Agent。连接器只暴露一组白名单SQL查询原语闸门限制返回行数和字段编排层按Agent身份控制可见的表范围。团队里有人说Agent直接生成SQL太危险但加了这层控制之后Agent就只能在我们允许的查询模板里做选择风险大大降低。群机器人自动回复。连接器接IM消息网关原语是向指定群发送消息和读取群里最近的对话。闸门限制消息长度和频率。编排层限制每个机器人一天能主动发多少条消息防止Agent情绪上头刷屏。定时巡检Agent。连接器接监控系统的只读接口原语是查询主机负载查询服务状态获取最近告警。闸门裁剪只保留指标名称、阈值和当前值。编排层设定触发频率并统一写审计。这套结构最让我惊喜的一点是换场景时几乎不需要改动编排层和闸门只需要写一个新的连接器。三层解耦的效果比预期更好。如果让我重新做一遍这个项目我可能不会一上来就写代码而是先画一张图模型有哪些操作原语连接器背后是什么系统编排层怎么分配权限。许多问题在设计图上就能暴露出来。最后再分享一个实际体会Agent项目的成败往往不在模型选得多强而在于那层又细又碎的接入层做得是否扎实。它不性感不会出现在任何演示PPT上但没有它Agent永远只能是会议室里的演示系统。 Agent-Reach这个名字是我去年年底一个项目跑崩之后想出来的。当时我做了一套基于大模型的工作流机器人模型本身已经能理解用户的模糊意图逻辑链路也没问题但到了真正要去改工单状态、去查数据库、去给群机器人发消息的时候它什么都够不着——代码在调用外部系统的那一刻频繁出错要么权限不对要么字段对不上要么返回的数据把上下文撑爆。我一度以为是模型能力不够换了好几个型号问题照旧。后来才意识到错不在大脑在手脚。这篇文章就围绕Agent-Reach展开完整记录我是怎么从失败现场出发重新设计了一个能让Agent真正触达业务系统的接入架构以及落地过程中的教训和优化。如果你也在做大模型Agent应用发现自己的Agent能聊不能干这篇文章应该能帮上忙。1. 第一个版本能聊不能干彻底暴露了Agent的触达短板1.1 当时的项目背景与失败现场项目最初的目标很简单给运维团队做一个能自动处理客户工单的机器人。客户在IM群里说一句我们线上有个订单一直没同步麻烦看一下Agent要能理解这句话然后去工单系统里查这个订单相关的工单更新状态并且在群里回复处理结果。听起来没什么难度对吧我当时也是这么想的。第一版直接给大模型接了一套function calling定义了好几个函数比如查询工单更新工单状态添加工单备注。模型在对话中会自己判断该调用哪个函数并且生成正确的参数。说实话在纯对话测试里这一套表现非常好模型几乎每次都能选出正确的函数、填出合理的参数。结果一上真实环境就崩了。工单系统不是只有查询工单一个接口它有几十个接口每个接口有不同的认证方式、字段命名规范、分页逻辑、权限范围。我需要给模型暴露一个统一的工具但这个工具内部要把几十个接口的差异全部消化掉。一开始我图省事把这堆接口直接包装成了几个粗粒度的大函数比如查询工单详情这个函数内部自动关联查询多个子接口。跑了几轮之后发现模型时不时会拿着工单号却传进了客户ID的位置或者查询参数超出了一个工单系统的枚举范围接口直接返回400。那段时间我每天的工作变成了教模型怎么正确调用API提示词里写了各种注意xxx字段必须是数字之类的补充说明。结果模型在对话环境里表现依然很好一接到真实请求就犯各种低级错误。后来我冷静下来发现问题的根源不在模型理解能力上——模型确实理解要做什么但它对外部系统的真实约束一无所知。1.2 Function Calling 只解决了想得到没解决够得着很多人做大模型Agent都有个误解只要接入了function callingAgent就能操作外部系统了。这个误解本身也曾经是我的误解。function calling解决的是模型决定调用一个工具这一步它让模型能把自然语言意图转化为一次结构化的函数调用请求。但是函数真正被执行的时候后面还有一整个外部系统的复杂性在等着要不要鉴权这个操作有没有写权限返回的字段哪些是模型需要的哪些字段包含大量冗余信息接口出错时错误信息能不能被模型看懂我把这些统称为触达问题。模型想得到要做什么但它的手够不到业务系统。你给它一把钥匙它不知道这把钥匙开哪扇门你把门打开了它又不知道该拿屋里哪样东西。想通这一点之后我把题目从优化提示词让模型更听话改成了设计一个能让模型安全、高效触达外部系统的中间层。这正是Agent-Reach的起点。1.3 我把项目改名为Agent-Reach的契机改名这件事挺突然的。那天晚上我在纸上画了三个框模型层、工具层、业务系统层。模型层负责理解工具层负责把模型意图映射成真实API调用。我盯着工具层这三个字看了很久意识到它不应该只是简单的API聚合而应该是一层带有语义翻译性质的能力封装——它决定了一个Agent到底能触及哪些系统、以什么方式触及、以多高的效率触及。于是我把项目代号改成了Agent-Reach。Reach这个英文词我觉得比接入集成精准得多。接入只是把线连上Reach还包含够得着覆盖得到的意味。后来的所有调整都围绕如何让Agent触达得更准、更可控、更便宜这三个词展开。2. Agent-Reach的三层结构每层解决一类具体问题2.1 连接器层把外部系统的复杂性消化在边界之内Agent-Reach的第一个核心设计是给每个外部系统配备一个独立的连接器Connector。连接器不是一个函数更不是直接暴露给模型看的工具它是一个语义翻译器。举个例子。工单系统真实API里有PATCH /tickets/{id}这种接口请求体有一堆字段其中有个字段叫hardware_order_status代表硬件订单的流转状态。如果让模型直接面对这种字段它根本不知道要不要传、传什么值因为自然语言里几乎没人会说硬件订单流转状态。连接器要做的事情是把这一坨复杂系统翻译成一小段模型听得懂的业务语言。我在连接器层定义了五个操作原语查询工单、变更工单状态、追加处理记录、分配处理人、查询工单历史。原语的名字必须使用业务人员能理解的动词而不是技术接口的名称。下面是我当时为工单系统写的连接器基类和实现代码是Python写的# connector_base.py from abc import ABC, abstractmethod class BaseConnector(ABC): 所有连接器必须实现统一入口 reach(). def __init__(self, config): self.name config[name] self.base_url config[base_url] self.token config[token] self.timeout config.get(timeout, 5) self.read_only config.get(read_only, False) abstractmethod def reach(self, operation: str, params: dict): 统一触达入口。operation 为业务原语params 为业务参数。 返回数据必须经过 ContextGate 裁剪。 pass def _check_write_permission(self, operation: str): if self.read_only and operation not in self.READ_ONLY_OPERATIONS: raise PermissionError(f当前连接器为只读不允许执行 {operation})每个业务系统继承这个基类内部再去调用各自的SDK或者HttpClient。模型的prompt里完全不出现PATCH /tickets/{id}它只看得见变更工单状态这五个字和少量参数说明。API再怎么变连接器内部改掉就行模型侧完全不受影响。连接器还有一个很重要的边界每个连接器只暴露极少数的操作原语。我之前吃过亏觉得功能越全越好把系统的几十个接口都暴露给了模型。结果是模型陷入选择困难而且每次生成参数时可选范围太大容易填错。收敛到5到10个原语之后模型的调用准确率明显上升。2.2 上下文闸门防止Agent一次触达就烧掉几万token第一版系统还有一个让我肉疼的问题token消耗非常夸张。原因很直接接口返回的数据模型根本消化不了那么多。工单详情接口一次会返回46个字段从流水号到内部审核代码再到各种时间戳总文本量三四千字。模型拿到这些数据后会把整段内容都塞进上下文再在这个基础上做后续推理。一次查询就是几千token一个复杂工单流程做下来动不动就上万token。更麻烦的是模型看到大量无关字段时偶尔还会编造一些字段的含义在回答里给用户错误的信息。Agent-Reach的第二层就叫上下文闸门ContextGate。名字听起来高大上核心职责只有三个字段白名单、记录条数上限、按需加载。字段白名单好理解就是我明确告诉闸门工单实体只能返回id、title、status、priority、created_at这五个字段其他的一概不要。比如真实接口返回了46个字段闸门过滤后只给模型看5个。大多数Agent任务根本不需要全部字段。记录条数上限是对列表类接口的。查询历史记录时如果明文条数是200条闸门默认只放行最近5条。只有模型明确请求我要看完整处理历史闸门才允许加载全量而且在返回前会标记这条数据量很大之类的提示让模型谨慎使用。按需加载是我后来补的一个机制。有些字段必须给但平时没人看比如工单关联的子订单列表。闸门先把存在关联数据这个事实告诉模型但默认不加载明细等模型在推理过程中明确说需要查看子订单明细时再发起第二次触达。这个机制既省token又避免了模型被大量次要信息干扰。经过闸门裁剪后一次典型工单查询的token占用从三四千降到了四五百而且模型的回复准确率反而上升了——它不用再去分辨哪些字段是有用的。2.3 编排层多Agent协作时统一决定谁来触达、触达什么Agent-Reach的第三层是编排层。我项目后面慢慢不止一个Agent了除了工单处理Agent还有一个负责整理每日报告的报告Agent一个负责在群里做初步响应的接待Agent。如果每个Agent都自己连工单系统那权限、审计、缓存这些全都各写一套根本维护不过来。编排层做的事情很简单上游Agent把需要触达的意图告诉编排层编排层检查这个Agent有没有权限然后路由到对应的连接器去执行执行完的数据统一过闸门再返回给上游Agent。这个设计让我想到了office里唯一的打印机你不需要给每个人配一台打印机只需要进办公室的走廊里把打印任务统一交给一个插口谁有权限、谁先打、打几份都集中管理。Agent触达外部系统也是同一个道理。编排层还有一个很实用的状态管理功能磁滞开关。如果某个连接器在最近一分钟内不可用编排层不会让Agent继续傻乎乎地反复重试而是直接在返回值里告诉Agent该连接器暂不可用请稍后再试。这事之前是没人管的模型拿了一次超时错误后可能会用另一个参数又发起一次调用不仅浪费token还可能让外部系统压力变大。下面这段是编排层路由逻辑的核心代码我保留了当时的基本骨架# router.py class ReachRouter: def __init__(self): self.connectors {} # name - connector instance self.policies {} # agent_name - allowed ops def register_connector(self, connector): self.connectors[connector.name] connector def bind_agent_policy(self, agent_name: str, allowed_ops: list[str]): self.policies[agent_name] allowed_ops def route(self, agent_name: str, connector_name: str, operation: str, params: dict): # 1. 权限校验 if operation not in self.policies.get(agent_name, []): raise PermissionError(fAgent {agent_name} 无权执行 {operation}) # 2. 连接器是否存在 connector self.connectors.get(connector_name) if not connector: raise ConnectionError(f连接器 {connector_name} 未注册) # 3. 放行到连接器 raw_result connector.reach(operation, params) # 4. 上下文闸门裁剪 gated_result ContextGate.trim(raw_result, connector_name, operation) return gated_result这个结构的价值在三方各取所需模型只关心业务语义连接器只关心系统对接细节编排层只关心权限和路由。每一层都做自己擅长的事系统的复杂度就降下来了。3. 最小实现30分钟搭出一个能真正操作工单系统的Agent3.1 环境准备和最小依赖选型做完设计我把整个系统重写了一遍。这次的目标非常明确跑通一个最小闭环。我选的技术栈如下每一行都说明一下原因Python 3.10生态成熟写AI相关的东西最顺手。FastAPI不是必须的但Agent内部服务将来一定需要处理并发请求用FastAPI把触达层做成HTTP服务后面接其他Agent会很方便。Redis做语义缓存。如果只求跑通可以先用Python字典顶着。但缓存是必需的后面会专门讲。LLM API我这边用的是一家兼容常见大模型接口的厂商服务配置环境变量即可不用绑定到具体某个品牌。我不建议一上来就搞Kafka、微服务、K8s这些东西Agent-Reach的第一版就是单进程扛住测试环境的几十个并发绰绰有余。先把业务闭环跑通瓶颈在哪一目了然再谈扩展也不迟。3.2 一个可复用的连接器实现我拿工单系统做例子写一个最小版本的TicketConnector。它内部调用一个模拟的外部工单系统客户端但接口命名、参数设计都按真实情况来# ticket_connector.py import time, hashlib, httpx from connector_base import BaseConnector class TicketConnector(BaseConnector): # 只暴露模型能看到的业务原语 OPERATIONS [ query_ticket, # 查询工单状态 update_ticket_status, # 变更工单状态 add_comment # 追加处理记录 ] # 只读原语写操作必须显式声明目标状态 READ_ONLY_OPERATIONS [query_ticket] def reach(self, operation: str, params: dict): self._check_write_permission(operation) if operation query_ticket: return self._query_ticket(params) elif operation update_ticket_status: return self._update_status(params) elif operation add_comment: return self._add_comment(params) else: raise ValueError(f未知原语: {operation}) def _query_ticket(self, params): ticket_no params[ticket_no] # 真实环境这里会去调工单系统HTTP接口 resp httpx.get( f{self.base_url}/tickets/{ticket_no}, headers{Authorization: fBearer {self.token}}, timeoutself.timeout, ) resp.raise_for_status() return resp.json() def _update_status(self, params): # 关键点所有写操作都在连接器内部做幂等键 idempotency_key hashlib.sha256( f{params[ticket_no]}:{params[target_status]}.encode() ).hexdigest() resp httpx.post( f{self.base_url}/tickets/{params[ticket_no]}/status, json{target_status: params[target_status]}, headers{ Authorization: fBearer {self.token}, Idempotency-Key: idempotency_key, }, timeoutself.timeout, ) resp.raise_for_status() return {updated: True, ticket_no: params[ticket_no]}这个连接器暴露给模型的操作原语只有三个。我故意砍掉了分配处理人查历史这些操作先让路由复杂度降到最低。跑通之后再加。事实证明这个顺序是对的。3.3 把Agent调用链跑通的关键配置环境准备完毕、连接器写好之后剩下的就是把几块拼起来。我当时的启动脚本是这样组织的# main.py from ticket_connector import TicketConnector from router import ReachRouter # 1. 初始化连接器 ticket_connector TicketConnector({ name: ticket_system, base_url: http://127.0.0.1:9000, token: dev-token, timeout: 3, read_only: False, }) # 2. 注册到路由 router ReachRouter() router.register_connector(ticket_connector) # 3. 给 Agent 绑定权限 router.bind_agent_policy(ops_agent, [ query_ticket, update_ticket_status, add_comment, ])这里有一个我特别想强调的点权限绑定是Agent级别的不是角色级别的。同一个连接器可以被多个Agent使用但每个Agent能执行的原语列表是单独声明的。这样将来某个Agent因为业务需要被收窄权限时改一行代码就行不会影响到其他Agent。启动之后整个调用链条长这样用户说你们看一下单号T20240415为什么还停在处理中 。接待Agent识别意图后把请求送到编排层。编排层查到ops_agent有权调用query_ticket于是让工单连接器去查单。连接器调外部接口拿到原始数据闸门把字段裁剪成id、title、status、priority、created_at五个返回给模型。模型据此生成回复这个工单当前状态为处理中优先级是P1已经超过SLA时限。跑通这个链路的那一刻我才觉得这个项目真正成立了。4. 上线后我踩过的四个坑每一个都是真金白银换来的4.1 权限放得太宽Agent把已关闭工单全部重新打开系统上线测试环境第一周出了一件让我哭笑不得的事。有一个测试账号的Agent在执行变更工单状态时把一批状态为已关闭的历史工单全部重新置为处理中了。根因很简单我一开始给Agent赋予的权限是所有工单操作包括重新打开已关闭工单这种操作。当时我觉得模型足够聪明不会做这种糊涂事。但我低估了大语言模型对用户自然语言的跟随程度——用户说帮我看看还有哪些已关闭的工单Agent的推理链走到某一步时擅自理解了需要重新打开它们以便进一步处理。从那之后我定了一条铁律Agent的权力边界必须由代码定义不能依赖模型自己的判断力。我在连接器里加了状态机校验比如已关闭的工单不能直接变更为处理中必须先变为待重新审核。模型拿到的操作说明里也明确写了这条限制。这相当于给Agent的手脚加上物理约束而不是靠劝诫来让它懂事。4.2 没有上下文裁剪一次鏖战烧掉十几万token这是我在第2章第2小节提到的教训的完整版本。测试群里有个用户连续追问了同一批工单的多个维度Agent需要反复调用接口查询。每一次接口返回的都是全量工单详情什么都不剪直接全部塞进上下文。我后来看了日志一次十五分钟的对话模型累计消耗了十几万token其中绝大部分都是重复返回的工单历史记录、内部备注、操作日志。这个成本按当时的价格算一天跑下来完全不可接受。我加上上下文闸门之后同一类对话的token消耗降到了原来的十五分之一左右。而且闸门带来的额外好处是模型回答的质量明显变稳定了因为它不再被几十个无关字段干扰注意力全部集中在真正重要的状态、优先级、最近跟进记录上。这件事给我的启发是大模型应用的成本优化很多时候不是靠换便宜的模型而是控制输入数据的质量。数据越干净模型越省token正确率也越高。4.3 重试逻辑放错层导致同一张工单被创建了两次这个坑是我自己埋的。有一次测试环境网络抖动Agent发起创建工单请求时超时了。我当时在Agent调用层写了一个自动重试只要请求超时就原样再发一次。结果第一次请求其实已经在服务端执行成功了只是响应丢失。重试发出后工单系统里出现了两张一模一样的工单。这个问题的本质是重试和幂等没有对应起来。解决办法有两个层面第一重试逻辑要收敛在连接器内部而不是让Agent自己判断该不该重试。Agent只需要知道这次触达的结果是成功还是失败失败的详细原因由连接器决定怎么处理。第二凡是可能产生业务数据的写操作必须携带幂等键。我在前面那段代码里已经把幂等键加上了用ticket_no加目标状态做哈希同一个请求意图不管重试多少次服务端都只认第一次。当时我排查这问题花了大半天因为两张工单的ID不一样但内容完全一样怎么看都像是Agent编造了数据。后来凭时间戳对比才发现是重试导致的。这个坑埋得深写代码时根本想不到。4.4 错误信息不够模型友好Agent只能瞎猜这个问题虽然排在第四但它几乎每天都会遇到。外部系统返回的错误信息往往是这样的{code: 50001, message: SYSTEM_INTERNAL_ERROR}模型拿到这种错误完全不知道发生了什么。它会自己脑补一个原因然后用错误的原因去调整参数再试一次。多数时候越试越错。我后来给连接器加了一个错误语义化层对外部系统的错误响应做一次翻译生成模型能理解的、带下一步建议的错误结构{ error_code: TICKET_LOCKED, meaning: 该工单当前被另一个流程锁定不能执行状态变更, suggestion: 等待5分钟后再试或先通过query_ticket确认解锁状态 }模型看到这样的错误信息才知道自己应该等待还是换个操作。这项改动表面上只影响了报错逻辑实际效果却是Agent的自主修复能力提升了一个档次。在那之前我总抱怨模型自作聪明后来才明白是我没给够它判断所需的上下文。5. 从能用走向好用我给Agent-Reach加了三个增强5.1 语义缓存让高频触达成本趋近于零工单系统有个高频场景好几个Agent在短时间内都要查询同一个工单号。如果没有缓存每次查询都会穿透到外部系统既耗token也占用外部系统接口配额。Agent-Reach在编排层加了缓存。最开始的版本是简单的参数哈希缓存也就是判断同一连接器、同一操作、同一参数就直接复用上一次返回值。但实际场景中不同Agent的措辞不一样查询参数却经常一样。于是我又补了一层语义判断把Agent发来的查询意图做向量化与缓存里近三十分钟的触达请求比对相似度超过阈值的直接命中缓存。当然带语义判断的缓存要注意一个边界凡是有写操作的数据缓存必须立刻失效否则Agent会拿到旧数据做出错误决策。我的做法是在连接器更新工单状态成功后主动通知编排层把该工单号相关缓存全部删除。这个逻辑叫触达缓存一致性没做好的话缓存优化反而会变成事故放大器。从实际效果看高频查询场景的接口调用量降了一半以上token消耗也同步降了。更重要的是Agent的响应速度提升了因为缓存命中时根本不需要等外部系统返回。5.2 触达审计每一个操作都有迹可循Agent越做越聪明最大的风险反而变成了不可预测。上线第三周测试环境中出现了一次误操作我需要快速回答三个问题谁做的做了什么为什么这么做答案来自审计日志。我在编排层加了一段简单的全局记录逻辑每条触达记录包含Agent名称、连接器名称、操作原语、参数摘要、返回状态、耗时、时间戳。存储介质我就用了SQLite数据量不大单表就够。有了审计日志之后权限配置也逐渐科学起来。我看了一周审计数据发现某一类只读查询被某Agent调用了五千多次说明它的业务逻辑可能存在重复查询另一个Agent频繁触发某个写操作但经常以失败告终说明它的参数生成有系统性偏差。这些数据反过来帮助我更准确地调整Agent的提示词和闸门配置。审计这个功能平时用不上但一旦出问题它就是最快定位系统的唯一线索。做Agent项目宁可先花一个小时把日志写好也不要等出事故了再熬夜翻代码查原因。5.3 这套结构可以迁移到哪些场景Agent-Reach的核心思路我后来在好几个不同的业务场景里复用了一遍。只要是智能体要对外部系统做操作的项目三层结构基本都成立数据库查询Agent。连接器只暴露一组白名单SQL查询原语闸门限制返回行数和字段编排层按Agent身份控制可见的表范围。团队里有人说Agent直接生成SQL太危险但加了这层控制之后Agent就只能在我们允许的查询模板里做选择风险大大降低。群机器人自动回复。连接器接IM消息网关原语是向指定群发送消息和读取群里最近的对话。闸门限制消息长度和频率。编排层限制每个机器人一天能主动发多少条消息防止Agent情绪上头刷屏。定时巡检Agent。连接器接监控系统的只读接口原语是查询主机负载查询服务状态获取最近告警。闸门裁剪只保留指标名称、阈值和当前值。编排层设定触发频率并统一写审计。这套结构最让我惊喜的一点是换场景时几乎不需要改动编排层和闸门只需要写一个新的连接器。三层解耦的效果比预期更好。如果让我重新做一遍这个项目我可能不会一上来就写代码而是先画一张图模型有哪些操作原语连接器背后是什么系统编排层怎么分配权限。许多问题在设计图上就能暴露出来。最后再分享一个实际体会Agent项目的成败往往不在模型选得多强而在于那层又细又碎的接入层做得是否扎实。它不性感不会出现在任何演示PPT上但没有它Agent永远只能是会议室里的演示系统。