☰
Agent-Reach:让大模型Agent真正触达业务系统的接入架构
2026/10/27 5:57:45 网站建设 项目流程

"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 calling,Agent就能操作外部系统了。这个误解本身也曾经是我的误解。

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(f"Agent {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分钟搭出一个能真正操作工单系统的Agent

3.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": f"Bearer {self.token}"}, timeout=self.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": f"Bearer {self.token}", "Idempotency-Key": idempotency_key, }, timeout=self.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永远只能是会议室里的演示系统。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询