做了两年多智能体应用,我最大的感受是:让大模型开口说话不难,难的是让它的手“够得着”你要的东西。我最近在整理一个项目,代号 Agent-Reach,直白一点讲,它解决的是一个很具体的问题:AI Agent 凭什么能稳定、安全、可追溯地触达外部的工具、数据源,以及另外一批 Agent。很多 Agent 看起来能聊能写,一旦面对真实业务——查库存、发邮件、调第三方接口、取数据库里的某条记录——就开始原地打转。问题不在模型本身,而在 Agent 和外部世界之间缺了一条可靠的触达通道。这篇文章是我把这个项目从零搭起来、跑通、踩坑之后的完整复盘,包含设计思路、关键代码和排查记录,适合正在做 Agent 应用、工具调用或多智能体协作的开发者参考。
1. 项目定位:Agent-Reach 解决的是“够得着”的问题
1.1 从场景聊起:Agent 卡在哪一步
先说一个很典型的例子。之前我帮一个电商团队做售前客服 Agent,用户问“这个订单还能改地址吗”,模型能理解意图,也能生成一段像模像样的回答,但真正的难题是:改地址要调订单系统的 API,要校验时间窗口,要判断是否已发货,还要把操作结果写进工单。没有一条可靠的触达通道,模型说得再漂亮,最后也只能给用户一句“我帮您看一下,请稍后”,然后就没有然后了。
我们把这类问题拆开看,Agent 卡住的位置非常统一,不是“思考”出了问题,而是“行动”断了线。具体有三层断点:
- 工具不可发现。模型不知道系统里有哪些能力,更不知道每个能力的入参、出参长什么样。
- 调用不可控。就算知道有工具,谁在调、能不能调、调得频不频繁,完全没有约束。
- 链路不可追。一次调用背后经过了哪些系统、花了多长时间、返回了什么,没有任何审计记录。
市面上常见的做法是在 system prompt 里把工具描述堆进去,再用 Function Calling 直接调后端服务。这个思路在小规模演示里很顺,一旦工具数量超过二十个、服务拆成多个团队维护,Prompt 会越来越胖,权限和灰度又没人管,最后变成一团解不开的线。Agent-Reach 的做法,是在模型和外部系统之间加一层专门管“触达”的中间层,统一负责注册、路由、策略和审计。
1.2 方案选型里的两次取舍
做这个项目之前,我先把候选路线列了一遍,最后的选择不是拍脑袋定的,而是被真实场景逼出来的。
| 方案 | 优点 | 主要代价 |
|---|---|---|
| 全部工具描述塞进 Prompt | 零依赖,见效快 | Token 爆炸,维护成本随工具数量指数上升 |
| 只用 Function Calling 直接调用 | 原生支持,链路短 | 无权限、无审计、无路由,多服务场景很难管理 |
| 直接用 MCP 标准协议 | 生态好,社区活跃 | 灵活度高但约束弱,细粒度权限和可观测性还是得自己补 |
| Agent-Reach 中间层 | 可控、可审计、可插拔 | 多一层服务,初期工程量稍大 |
第一次取舍发生在“标准化”和“可控性”之间。MCP 是很好的协议,但它解决的是“接口长什么样”的问题,不解决“谁允许调”和“调完怎么复盘”的问题。Agent-Reach 把 MCP 这类标准当成底层接入方式之一,同时在上层加了策略引擎和审计管道,相当于给 Agent 的每只“手”都装了一道闸门。
第二次取舍发生在“中心化”和“去中心化”之间。早期我试用过直接在 Agent 进程里嵌注册表的路子,轻是真轻,但多个 Agent 共享能力时需要各自维护一份,配置很快就漂移了。后来改成中心化注册加去中心化执行的架构:注册表集中管理,工具实际运行在各自的服务里,中间层只做编排和策略判断。这样既避免了重复配置,又不用把所有调用流量都拉进同一个进程,代价只是多维护一个网关服务。
提示:如果你手头只有两三个工具,不要急着上这套东西。Agent-Reach 的价值要从“工具数量多、权限要求细、链路需要审计”这三个条件同时成立时才开始显现。
2. 核心机制拆解:注册中心、路由与安全边界
2.1 能力注册:让每个工具自带“说明书”
Agent-Reach 里所有的工具都不是写死在代码里的,而是通过注册中心动态加进去的。注册信息包含四个部分:名字、描述、参数 Schema、执行函数。前两项是给模型看的,参数 Schema 用来做校验,执行函数才是真正干活的代码。
名字的命名我强烈建议用“域.动作.对象”三段式,比如warehouse.check_stock、order.update_address、payment.refund_apply。这样命名有三个好处:路由策略可以用通配符批量授权,日志里扫一眼就能定位到业务域,模型在工具太多时也能减少混淆。
描述这一栏非常关键。不要写“查询库存”这种一句话,要写清楚“什么时候用、有什么前置条件、失败时可能因为什么”。原因很简单:模型是靠描述来决定要不要调用这个工具的,描述敷衍,模型就会瞎猜。我见过一个工具描述没写“仅支持已支付订单”,结果模型在未支付订单上反复调用,生成了一堆无意义的报错。
参数 Schema 直接采用 JSON Schema 标准。这里有一个新手容易忽略的点:每个字段都要写 description,枚举值要尽量给全。模型虽然不至于把字符串参数拼错,但面对枚举值的时候,如果你不告诉它可选项,它真的会自己发明一个值出来。
执行函数是唯一的真实逻辑入口。在 Agent-Reach 里,我要求注册进来的 handler 只做一件事:接收已经校验过的参数,调用后端系统,返回结果。所有重试、降级、超时逻辑都放在中间层的路由网关里,不让业务函数自己处理,这样才能保证每一个工具的行为是统一、可控的。
2.2 动态路由:一次调用背后的完整链路
路由层是整个项目的核心。模型产生一个工具调用请求后,请求不会直接打到业务服务,而是先进路由网关走五步流程:
- 权限判断。根据调用方身份、所在 scope、目标工具名查策略表,不允许就直接拒绝并记录原因。
- 参数校验。用注册时的 JSON Schema 校验模型生成的参数,缺字段、类型不对都在这一步拦截。
- 限流计数。从 Redis 里扣减调用额度,超限就返回“频率超限,请稍后重试”。
- 执行调用。把校验后的参数传给工具 handler,同时启动超时计时。
- 审计落库。记录调用方、工具名、入参、出参摘要、耗时、成功失败标记,最终写进审计日志库。
这五步看起来多,但每一步都有明确的职责,少了哪一个,线上都会出事。尤其是第 5 步,很多人嫌麻烦想省掉,真出了“Agent 乱调工具”的投诉时,没有审计日志你连定位都没法定位。
路由过程里有个细节,出参摘要。工具返回值可能会很大,比如一个订单查询返回了三百行明细,直接把全量内容塞回给模型,一方面浪费 Token,另一方面会干扰模型对后续对话的判断。我采用的策略是每个工具返回都做两层裁剪:先由 handler 自己生成一个summary字段,路由层再做一次按字符数的截断,默认 2000 字符以内。这个参数我放在后面详细讲。
2.3 安全边界:权限、限流与审计
很多 Agent 项目把安全想得太简单,以为“只有模型能调工具,所以不需要鉴权”。这是非常危险的想法。Agent 本身可能被提示注入,一个恶意的用户输入完全可以诱导模型去调用高权限工具。Agent-Reach 在安全边界上做了四件事。
第一件,最小权限。给每个 Agent 分配一个身份,策略表只允许它访问完成业务必需的工具。客服 Agent 能查订单、能改收货地址,但绝不能直接调用退款接口。退款要走独立的审批 Agent,两边通过路由网关衔接。
第二件,分 scope 隔离。同一个工具可以注册多个 scope,比如warehouse.check_stock在零售域和服务台域可以有不同的并发额度。scope 本质上是租户隔离,防止一个域的高峰流量把另一个域的额度吃光。
第三件,调用链上下文。每个 Agent 会话都有唯一的agent_id,每次路由都会生成trace_id,这两个 ID 贯穿所有审计日志。出问题的时候,只要拿到用户的一句话,就能顺着 trace_id 把所有工具调用记录串起来。
第四件,敏感信息过滤。工具的原始返回里可能包含身份证号、手机号、内部备注。路由层在把结果交给模型之前,会做一次敏感字段掩码。这一步不能依赖模型自觉,必须在系统层面强制。
注意:审计日志同样需要权限保护。日志里记录了入参和出参摘要,如果日志库本身不设防,等于把钥匙挂在门旁边。我见过不止一个团队把审计日志存在同一个库同一个账号下,最后排查问题时看到权限混乱,反而不敢信日志了。
2.4 关键参数表和配置建议
参数配置是 Agent-Reach 里最容易被忽略、但影响最大的部分。先给一份我在生产环境用的推荐值:
| 参数 | 默认值 | 推荐值 | 说明 |
|---|---|---|---|
| tool_call_timeout | 10s | 15s | 单次工具调用的超时时间,超过则返回错误 |
| max_retry | 1 | 2 | 网络类错误的自动重试次数,业务错误不重试 |
| context_budget | 2000 字符 | 2000 字符 | 每个工具结果回传给模型的最大长度 |
| max_iterations | 8 | 8 | Agent 单轮对话中最多连续调用工具的次数 |
| rate_limit | 60/min | 按业务定 | 每个 Agent 每分钟最多调用某一类工具的次数 |
| allow_reroute | false | false | 是否允许工具结果再次触发其他工具,默认关闭 |
max_iterations是防“鬼打墙”的关键参数。模型有时候会在回答不出来的时候反复调用同一个工具,不加这个上限,一次对话能烧掉你几百次 API 调用。allow_reroute默认关闭也是同样的原因,工具结果自动触发下一个工具,链路一旦出现逻辑环,很难打断,所以多 Agent 场景下的链式调用我都改成显式路由,让 Agent 自己决定下一步调什么。
context_budget的设置要结合模型上下文窗口来看。窗口大不代表可以随意塞,工具返回的信息是为了让模型做决策的,不是让它背诵的。我在实验里把预算从 2000 调到 8000,模型的回答准确率没有明显提升,反而更容易在冗余信息里抓到次要字段。
3. 实操落地:从零搭一个最小可用的 Agent-Reach
3.1 目录结构和基础依赖
先给目录结构。我没有用复杂的微服务框架,而是把网关、注册、策略、审计拆成包,方便各自演进。
agent-reach/ ├── gateway/ # FastAPI 服务,统一入口 │ └── router.py ├── registry/ # 工具注册与管理 │ ├── manager.py │ └── schema.py ├── connectors/ # 外部系统适配器 │ ├── warehouse.py │ └── order.py ├── policies/ # 权限与限流策略 │ ├── default.yaml │ └── engine.py ├── audit/ # 审计日志 │ └── logger.py ├── agents/ # Agent 调用循环 │ └── loop.py └── examples/ # 场景示例 ├── single_agent.py └── approval_chain.py基础依赖尽量精简,我用的是 Python 3.10、FastAPI、Redis 客户端、SQLAlchemy。大模型接口做了适配层,统一走 OpenAI 兼容格式,方便切不同厂商的模型。Redis 在这里的用途是存限流计数和路由表缓存,审计日志写 PostgreSQL。
3.2 工具注册模块实现
注册模块最关键的是保存 JSON Schema 和执行函数,同时维护一份给模型看的工具列表。下面这段是我实际使用的简化版:
# registry/manager.py from __future__ import annotations import time from typing import Awaitable, Callable, Dict ToolHandler = Callable[..., Awaitable[dict]] class ToolRegistry: def __init__(self) -> None: self._tools: Dict[str, dict] = {} def register( self, name: str, description: str, parameters: dict, handler: ToolHandler, scope: str = "default", ) -> str: self._tools[name] = { "description": description, "parameters": parameters, "handler": handler, "scope": scope, "created_at": int(time.time()), "invoke_count": 0, } return name def lookup(self, name: str, scope: str): tool = self._tools.get(name) if tool is None or tool["scope"] != scope: return None return tool def list_for_llm(self, scope: str) -> list: tools = [] for name, info in self._tools.items(): if info["scope"] != scope: continue tools.append( { "type": "function", "function": { "name": name, "description": info["description"], "parameters": info["parameters"], }, } ) return tools注册时有两个细节提醒一下。scope 字段如果不传,默认是default,我建议所有业务注册时都显式传 scope,避免后面策略配置时默认权限搞错。invoke_count目前只是内存计数,生产环境我会丢到 Redis 里做累加,否则重启就清零了。
3.3 路由网关实现
路由网关是请求进入后的守门人。下面这段省略了数据库操作和细节异常处理,保留核心流程:
# gateway/router.py import time import uuid from dataclasses import dataclass @dataclass class ToolContext: agent_id: str scope: str trace_id: str async def route_tool_call(registry, call, context, policies): tool_name = call["name"] args = call.get("arguments", {}) # 1. 权限检查 decision = policies.check(context.agent_id, tool_name, context.scope) if not decision.allowed: audit_log("deny", context, tool_name, {"reason": decision.reason}) raise PermissionError(f"{tool_name} is not allowed for {context.agent_id}") # 2. 参数校验 tool = registry.lookup(tool_name, context.scope) validated_args = validate_with_schema(tool["parameters"], args) # 3. 限流与计数 policies.consume_quota(context.agent_id, tool_name) # 4. 执行并计时 started = time.perf_counter() result = await tool["handler"](**validated_args) elapsed_ms = int((time.perf_counter() - started) * 1000) # 5. 审计 audit_log( "allow", context, tool_name, { "args": validated_args, "result_summary": summarize(result, max_chars=2000), "elapsed_ms": elapsed_ms, }, ) return result这段代码里最有价值的一点是,权限、校验、限流每个环节失败都会走明确的异常路径,并且立刻写审计。实际使用时我还会在每个环节加一个起始时间戳,方便查看一次调用到底卡在权限还是卡在外部接口。
3.4 与 LLM 的循环调用实现
路由网关本身不产生模型调用,真正驱动它的是 Agent 的循环。这里实现了标准的“模型决定工具 -> 路由执行 -> 结果回填 -> 再交给模型”的循环:
# agents/loop.py import json async def run_agent_with_reach( llm, registry, policies, system_prompt: str, user_message: str, scope: str, agent_id: str, max_iterations: int = 8, ): tools = registry.list_for_llm(scope) messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message}, ] for step in range(max_iterations): response = await llm.chat(messages=messages, tools=tools) if not response.tool_calls: return response.content messages.append(response.message) for call in response.tool_calls: context = ToolContext(agent_id=agent_id, scope=scope, trace_id=uuid.uuid4().hex) try: result = await route_tool_call(registry, call.function, context, policies) content = summarize(result, max_chars=2000) except Exception as exc: content = f"__TOOL_ERROR__: {exc}" messages.append( { "role": "tool", "tool_call_id": call.id, "content": json.dumps(content, ensure_ascii=False), } ) raise RuntimeError(f"exceeded max_iterations={max_iterations}")循环里有两个细节很影响稳定性。第一,工具返回内容必须序列化成字符串再放回消息列表,不能用 dict 直接传,否则不同模型 API 的兼容性会出问题。第二,异常信息不能直接透传原始报错,否则数据库密码、内网地址可能被模型看到甚至复述出来。我上面用了__TOOL_ERROR__前缀,并且只放了一条安全的消息摘要。
3.5 单 Agent 场景实测
搭好之后我先跑了一个最朴素的场景:让 Agent 查某个 SKU 的实时库存。注册工具:
# connectors/warehouse.py async def check_stock(sku: str): # 真实项目里这里换成对 ERP/WMS 接口的 HTTP 调用 return {"sku": sku, "available": 128, "updated_at": "2025-01-12 10:23"} registry.register( name="warehouse.check_stock", description="查询某个 SKU 的实时可用库存。当用户问还剩多少、够不够发货时使用。", parameters={ "type": "object", "properties": {"sku": {"type": "string", "description": "商品 SKU 编码"}}, "required": ["sku"], }, handler=check_stock, scope="retail", )用户输入是“SKU 10086 的库存够明天发货吗?”模型会先生成warehouse.check_stock调用,路由器校验权限,执行 handler,把结果回填,模型再基于库存数字给出判断。整个过程里用户无感知,但审计日志里已经完整记录了模型看到了什么、调了什么、每条结果耗时多少。
第一次跑通这个循环时,系统日志里多了十几个trace_id,那种感觉就像给 Agent 装上了一张能看到“手”在动的仪表盘。这也是我后来坚持所有工具必须走统一路由的原因——没有这一步,你永远不知道模型到底干了什么。
4. 进阶场景:多 Agent 协作与权限审批链
4.1 设计思路:Agent 注册为工具
单 Agent 跑通以后,自然想解决更复杂的问题:多个 Agent 之间怎么协作。我在 Agent-Reach 里采用了一个非常朴素的设计——把一个 Agent 的能力也注册成工具。也就是说,Agent 和 Agent 之间不是私聊,而是通过路由网关互相调用。
这个设计听起来很直接,但它避开了很多协作框架里的坑。Agent 之间不直接传 API Key,不直接约定接口地址,一切都走注册发现。下游 Agent 上线新能力,只需要注册一个新的工具;上游 Agent 想用,只需要策略表里加一条允许规则。协作关系从代码耦合变成了配置声明。
以客服场景为例。客服 Agent 被用户要求退款时,它自己不能调支付退款接口,只能调用一个名为agent.approval的工具,这个工具对应的 handler 会唤醒审批 Agent。审批 Agent 看到退款申请,核对订单和金额,然后通过工具调用返回“同意”或“拒绝”。整个链路里每一步都有迹可循。
4.2 场景实现:审批链路
下面是一个简化版的审批链路注册:
# examples/approval_chain.py async def request_refund_approval(order_id: str, amount: float): # 这里通过内部代理再调起审批 Agent return await route_tool_call( registry, { "name": "agent.approval", "arguments": { "target": "refund", "order_id": order_id, "amount": amount, }, }, context=ToolContext( agent_id="customer_service", scope="ops", trace_id="...", ), policies=approval_policies, ) registry.register( name="agent.approval", description="发起一笔退款审批。客服不能直接退款,必须先取得审批通过。", parameters={ "type": "object", "properties": { "target": {"type": "string", "enum": ["refund"], "description": "审批类型"}, "order_id": {"type": "string", "description": "订单号"}, "amount": {"type": "number", "description": "退款金额,单位元"}, }, "required": ["target", "order_id", "amount"], }, handler=request_refund_approval, scope="ops", )配套的策略表长这样:
# policies/default.yaml policies: - agent: "customer_service" allow: - "order.query" - "order.update_address" - "agent.approval" deny: - "payment.refund" quota: 120/min这个策略文件我要特别解释一下。deny规则比allow优先,这是安全设计上的保守选择。哪怕你以后想在allow里用通配符给客服 Agent 开放一大类工具,只要deny里有payment.refund,这条权限依然会被拦下来。
审批 Agent 被调起来之后,它会先查订单信息,再判断退款金额是否在阈值内,最终返回一个结构化结果。因为审批 Agent 的能力也是通过注册中心暴露的,所以审计日志里能看到:用户说“我要退单” -> 客服 Agent 调用审批工具 -> 审批 Agent 查订单 -> 审批 Agent 返回同意 -> 客服 Agent 回复用户。完整凭证链一目了然。
4.3 常见问题速查表
| 症状 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| Agent 反复调用同一工具 | 参数校验失败但模型没拿到提示 | 看审计日志里 deny 记录 | 在工具错误消息里明确返回缺失字段 |
| 返回内容过长,模型抓不住重点 | 没有设置 context_budget | 检查 messages 长度 | 在路由层强制 summarize |
| 工具调用一直超时 | 外部接口慢,超时参数过短 | 看耗时分布 | 调大 tool_call_timeout,同时做上下游超时隔离 |
| 权限拒绝风暴 | scope 或 policy 配置漂移 | 对比不同环境的策略表 | 用配置文件做版本管理,部署前跑策略单测 |
| 多 Agent 链路死循环 | 误开了 allow_reroute | 查 trace_id 调用链 | 关闭自动重路由,改用显式工具调用 |
这张表不是凭空写的,每一条都在我自己的项目里真实出现过。尤其是“参数校验失败但模型没拿到提示”那条,第一次遇到时排查了很久,最后发现是异常信息里没有告诉模型到底缺了什么字段,模型以为参数没问题,就一遍遍重试。后来我把异常信息改成“缺少参数: order_id”,问题立刻消失。
5. 调优与排障实录:三个最值得说的案例
5.1 流量放大问题的解决
上线第一周,我就发现一个反直觉的现象:Agent 调用工具的 QPS 比用户请求量高了将近十倍。审计日志一查,绝大多数都是同一类情况——用户问一句,Agent 连续调三次库存接口。
原因在上下文,模型的上下文窗口里如果只保留了最近一次库存结果,它在做跨商品比较时就会重新调用。这不是模型笨,是工具结果的记忆太短。解决办法不是加上下文,而是给常用工具加缓存。我在注册中心给warehouse.check_stock加了 30 秒的 Redis 缓存,同 SKU 的查询直接命中缓存,QPS 立刻降了下来。
这里有一个关键前提:缓存只适合幂等查询类工具,任何有副作用的写操作都绝对不能缓存。给写操作加缓存,等于让 Agent 在一个错误前提上继续往下走,后果非常难追溯。
5.2 上下文预算的重要性
另一个项目里,Agent 查订单返回了明细列表,我嫌每个字段都有用,就没做截断。结果用户多问几轮,模型开始出现“记忆稀释”——前面的约束条件忘了,工具调用开始频频出错。这个现象的本质是上下文窗口被工具返回占满,留给对话和推理的空间不够了。
后来我把context_budget从默认值改成显式按工具配置。订单查询这个工具返回给模型的内容,只保留订单状态、金额、关键时间点三个字段,其余明细一律省略。模型回答的准确率反而明显提升。这件事让我确认了一个原则:工具返回要的是“够用”,不是“完整”。
5.3 一次线上排查的完整复盘
最难忘的一次排障大约花了四个小时。现象是客服 Agent 偶尔反馈“查不到订单”,但同样的订单号在管理后台明明能看到。
第一反应是看模型有没有正确传参。审计日志一翻,传参没问题,订单接口也返回了正常数据。再看时间,发现一个规律:每次查不到订单都发生在下午三点左右,而且集中在同一个仓库的订单上。
顺着 trace_id 继续查,最后定位到端侧缓存。仓库那边有一个老接口,查询结果缓存一小时,缓存击穿的时候返回了空列表。问题不在 Agent 侧,但 Agent 把空结果当成了“订单不存在”,并非常自信地告诉了用户。
这次排障的最大收获是:Agent 应用的排查不能只盯着模型和路由层,外部的老系统可能是整个链路里最不可控的一环。后来我在 Agent-Reach 里加了一个约定:所有工具返回必须带数据时间戳,并且路由层会把“数据更新于 XX 分钟前”拼进结果摘要。模型看到这个信息后,回答就会带上“根据 X 分钟前的数据”这样的限定语,不会再百分百确定地给结论。
6. 我再聊几句个人体会
6.1 这个项目真正有价值的产出
做 Agent-Reach 这段时间,我越来越明确一件事:真正让项目落地见效的不是花哨的模型配置,而是把工具边界、权限模型和审计机制想清楚。很多团队一开始追求“什么都能干”的大 Agent,结果上线后连基本的可观测性都没有。我这套方案最大的产出,是把“Agent 干了什么”从黑盒变成了白盒——每次工具调用都有记录,每个权限决定都有依据,每个异常都能顺着 trace_id 找到根因。
6.2 给后来者的几条建议
从我的经验来看,如果你想在自己的项目里借鉴 Agent-Reach,不要一上来就铺开做全套。先让一个 Agent 稳定触达两三个高频工具,跑通权限和审计,再逐步加工具、加 Agent。另外,工具描述值得你花时间字斟句酌,它的质量直接决定模型调用工具的准确率。最后,保持默认参数的保守主义,max_iterations和allow_reroute这种参数,宁可一开始限制得紧一点,也不要让模型自由发挥。
这套东西后边还有很多可以扩展的方向,比如把策略表做成可视化配置、把审计日志接进告警体系、给工具调用做成本预估。我目前还在持续迭代,后续有新的收获会再整理出来。