企业官网建设流程全解析

如果你最近在折腾AI Agent，应该已经体会到了：单个Agent跑通一个流程并不难，难的是让多个Agent真正协作起来。我写下 Agent-Reach 这个轻量级中间层，就是被多Agent协作里的连接问题逼出来的。简单说，它就是给每个Agent发了一张“门牌号”，让彼此能通过一个统一触达层互相发现、调用和鉴权，而不是在代码里写死一堆IP和端口。

这个项目适合两类人看。一类是已经开始做多Agent系统，但被服务发现、权限控制、链路定位搞得焦头烂额的工程师；另一类是刚接触Agent工程化，想理解Agent之间到底怎么通信的开发者。我会从设计思路、核心代码、部署方式，一路讲到我踩过的坑，尽量说人话，不讲虚的。

1. 为什么我会自己写一个 Agent-Reach

1.1 多Agent协作里的“连接”问题

最开始我做多Agent系统时，根本没想那么复杂。模型负责推理，工具负责执行，Agent A需要Agent B帮忙时，直接写一个HTTP请求就完了。业务不复杂的时候，这样最快，真的，两次请求就能跑通一条链路。

但Agent数量从两三个涨到六七个之后，问题就全冒出来了。配置文件里散落着各个Agent的地址，今天这个Agent换个端口，明天那个Agent升级一下接口，调用方要么手动改配置，要么等着线上报错。更麻烦的是权限边界不清，有的Agent能力属于敏感操作，比如退款、删除数据，我不能让任意一个Agent都能调它。还有排障问题，一次跨Agent调用失败了，到底是调用方问题、网络问题，还是目标Agent内部出错，日志分散在各个服务里，抓瞎抓了半天。

我也考虑过引入现成的服务网格或者消息总线。但说实话，在Agent这个场景，很多方案都太重了。Agent数量通常不会像微服务那样动辄几百上千，我却要为它搭一套完整的服务发现、负载均衡体系，维护成本比业务本身还高。我需要一个更贴合Agent语义的轻量方案。

1.2 Agent-Reach 到底解决什么问题

Agent-Reach 解决的问题可以归纳成四件事：服务发现、统一路由、访问控制，和可观测性。它就像电话交换机，每个Agent都是接入交换机的一部电话。以前我要给每部电话都记一份“哪个分机在哪”的通讯录，现在交换机统一维护分机表，谁要联系谁，拨个号就行。

服务发现解决的是“Agent有哪些能力，在哪个地址”。每个Agent启动时把自己能提供的技能列表和实际端点注册上来，调用方按能力名搜索，就能知道该找谁。

统一路由解决的是“调用方不需要直接访问目标Agent”。所有请求先经过Agent-Reach，由它根据注册表找到目标端点并转发，调用方只需要知道Agent-Reach一个地址就够了。这样目标Agent可以随时换端口、换宿主、扩多个实例，调用方完全无感知。

访问控制和可观测性是我在实现过程中发现的两个隐藏刚需。没有统一入口，就没有统一的鉴权点；没有鉴权点，就只能靠各个Agent自己设白名单。而链路追踪更不用说了，多Agent调用必须有trace_id贯穿整条链路，不然出了问题根本定位不到是哪一段耗时。

1.3 它的边界在哪里

Agent-Reach 不适合解决所有问题。比如Agent之间如果只是发布事件、不需要立刻拿结果，那应该走消息队列，而不是RPC网关。Agent-Reach 做的是同步请求/响应，调用方发出请求后会等目标Agent返回结果，这和异步事件驱动完全是两种模式，不能混着用。

如果你的团队已经有Istio这类相当成熟的服务网格，也没必要再造一个Agent-Reach，直接复用已有的东西就行。Agent-Reach 更适合的阶段是：你已经开始做多Agent，但还没有统一的基础设施，又不想为这个问题引入一套重型框架。说白了，它追求的是用最小成本把Agent之间的连接打通。

2. Agent-Reach 的整体架构与关键设计

2.1 注册中心：给每个Agent发一张“门牌号”

Agent-Reach 的核心是注册中心。我选了Redis来做这张“门牌号登记表”，原因很直接：Agent数量不大，Redis足够扛住；它的EXPIRE和有序集合天然适合做TTL过期和清理，我可以少写很多管理代码。

每个Agent启动时，会往注册中心写入一条记录，包含几个关键字段：agent_id是全局唯一标识，name是给人看的名字，capabilities是这个Agent能提供的能力列表，endpoint是实际可访问的HTTP地址。写入同时设置一个TTL，比如30秒，Agent需要定期发心跳续期，否则记录会自动过期。

这里我要特别强调一个容易踩的坑：注册的endpoint必须是Agent-Reach网关能访问到的地址，不是Agent自己觉得能访问的地址。很多人会把本机地址写成localhost:9001，在单机演示时没问题，一旦Agent跑在容器里或者另一台机器上，网关转发过去就是连接拒绝。所以实际项目中，我通常要求注册时传入在局域网内可达的IP或服务名，这个约束一定要写进接入文档。

能力发现我用了一个最直白的方式：在Redis的有序集合里维护所有存活Agent的ID和过期时间，需要查找某个能力时，遍历存活的Agent，逐个从Hash里取出capabilities列表做精确匹配。Agent数量在几十个以内时，这个方案性能完全没问题。如果以后规模真的上去了，可以换成单独的技能索引甚至接入全文搜索，但那不是当前阶段要考虑的事。

2.2 一次带链路追踪的远程调用

Agent-Reach 处理一次调用的完整流程，可以拆成五步。调用方先构造一个JSON-RPC请求，发给Agent-Reach；网关校验调用方有没有权限，这个动作必须有；然后根据target或者method从注册表查出目标Agent的真实端点；接着网关生成一个trace_id，并设置整体超时时间，把请求转发给目标Agent；最后等到目标Agent返回结果，网关再原样包装返回给调用方。

在协议选择上，我用了JSON-RPC 2.0而不是自定义一套REST接口。原因是多Agent协作本质上是能力调用，用RPC语义更顺。JSON-RPC 2.0规定了jsonrpc、id、method、params、result和error这些标准字段，所有Agent都能按同一套规则处理请求，不用每个服务都自己定义一套URL规范和参数结构。HTTP传输层还是保留，因为网关要做鉴权、路由、超时控制，HTTP最成熟也最好调试。

一次请求的报文长这样：

{ "jsonrpc": "2.0", "id": "req_20250101_001", "method": "order.query", "params": { "order_id": "20250101001" }, "trace_id": "ar_8f3a2c4e9b", "meta": { "target": "order-agent", "deadline_ms": 3000 } }

很多人在设计Agent通信时忽略了超时这件事。一次跨Agent调用，目标Agent背后可能还接着模型推理、数据库查询，这些操作动辄几秒甚至几十秒。如果不在网关层统一设置deadline，调用方很容易自己先超时重试，导致目标Agent被重复请求打爆。Agent-Reach 里我要求每次调用显式带上deadline_ms，网关用它控制转发等待时间，超时就返回504，干净利落。

2.3 安全与鉴权：先内网互信，再外网隔离

安全设计我建议分两步走。第一版只要解决内网环境下的基础信任问题，不需要一上来就上mTLS、证书体系，否则项目还没跑起来就被复杂度拖死了。第二步才是针对外部网络、跨团队调用的加固。

广域网边界上，我在Agent-Reach网关前加一道Token校验。调用方请求时必须带上经过签发的访问令牌，否则直接拒绝。这个Token只负责“你是否有资格调用Agent-Reach”，不负责细粒度权限。

细粒度权限放在能力级别。我在网关侧维护了一个简单的访问控制表：某个能力允许哪些Agent调用，例如order.refund只允许finance-agent和admin-agent调用。这个表第一版直接用YAML配置文件就行，我见过太多人一上来就引入新的权限系统，最后权限系统比业务还复杂，完全没必要。

网络层面，我没有开放Agent-Reach给公网直接访问，而是要求它部署在内网网关后面。如果一定要跨网络访问，建议在负载均衡层加IP白名单，后续再考虑双向mTLS。Agent之间走局域网通信，风险面小很多，这符合先解决核心问题、再逐步加固的思路。

3. 最小可用版本怎么落地

3.1 技术栈选型与工程目录

整个项目的技术栈非常克制。API框架用FastAPI，它天然支持异步，类型提示和自动文档省了我不少事。注册中心用Redis，前面说过，TTL特性就是为这种场景准备的。HTTP转发用httpx的异步客户端，超时控制写起来很顺手。部署用Docker Compose，本地起一整套环境只要一条命令。

工程目录也很简单：

agent-reach/ ├─ app/ │ ├─ main.py # FastAPI 入口，注册路由接口 │ ├─ registry.py # 注册中心逻辑，Redis 操作 │ ├─ router.py # 转发核心逻辑 │ └─ security.py # Token 与签名校验 ├─ requirements.txt └─ docker-compose.yml

我特意没有引入Nacos、Consul这类注册中心，不是说它们不好，而是多Agent场景的服务数量通常不大，几十个已经很多了。用Redis做注册表，运维只需要一个Redis实例，心智负担小得多。框架选型这事，永远要匹配场景复杂度，杀鸡不用牛刀。

3.2 注册与发现的代码实现

注册模块的核心逻辑其实不复杂，就是用Redis的Hash存Agent详情，用有序集合维护存活状态。我先把registry.py的关键代码贴出来：

import json import time import redis.asyncio as aioredis class Registry: def __init__(self, redis_url: str): self.redis = aioredis.from_url(redis_url, decode_responses=True) async def register(self, agent: dict, ttl: int = 30) -> None: agent_id = agent["agent_id"] key = f"agent:{agent_id}" await self.redis.hset(key, mapping={ "name": agent.get("name", ""), "endpoint": agent["endpoint"], "capabilities": json.dumps(agent.get("capabilities", [])), "updated_at": str(time.time()), }) # 设置整体过期时间，同时把 agent 放入存活集合 await self.redis.expire(key, ttl) await self.redis.zadd("agents:ttl", {agent_id: time.time() + ttl}) async def heartbeat(self, agent_id: str, ttl: int = 30) -> bool: key = f"agent:{agent_id}" if not await self.redis.exists(key): return False await self.redis.expire(key, ttl) await self.redis.zadd("agents:ttl", {agent_id: time.time() + ttl}) return True async def unregister(self, agent_id: str) -> None: await self.redis.delete(f"agent:{agent_id}") await self.redis.zrem("agents:ttl", agent_id) async def find_by_capability(self, capability: str) -> list[dict]: agents = [] for agent_id in await self.redis.zrange("agents:ttl", 0, -1): data = await self.redis.hgetall(f"agent:{agent_id}") if not data: # 如果 Hash 已经过期，顺手清理 zset 里的残留项 await self.redis.zrem("agents:ttl", agent_id) continue caps = json.loads(data.get("capabilities", "[]")) if capability in caps: data["agent_id"] = agent_id agents.append(data) return agents

注意heartbeat只续期TTL，不做业务健康检查。这算是一个经验：心跳只能说明进程还活着，不能说明Agent背后依赖的数据库、模型服务还正常。所以我在每个Agent内部都单独提供/healthz接口，Agent-Reach网关可以做定时健康探测，但健康探测不应该影响注册表的实时性。两个机制分开，别搅在一起。

3.3 路由转发与能力调用的代码实现

转发模块是整个网关最核心的部分。调用方发来一个POST /v1/call请求，里面带目标Agent标识、方法名、参数，以及调用方自己的身份和签名。网关要做的事情是：校验身份、查注册表、设置trace_id、转发、等待结果。

核心代码如下：

import uuid import httpx from fastapi import HTTPException async def forward_call(payload: dict, agent: dict) -> dict: trace_id = f"ar_{uuid.uuid4().hex[:12]}" body = { "jsonrpc": "2.0", "id": payload["request_id"], "method": payload["method"], "params": payload.get("params", {}), "trace_id": trace_id, } endpoint = agent["endpoint"] timeout_ms = payload.get("deadline_ms", 3000) try: async with httpx.AsyncClient(timeout=timeout_ms / 1000) as client: resp = await client.post(endpoint, json=body) resp.raise_for_status() return resp.json() except httpx.ReadTimeout as exc: raise HTTPException(status_code=504, detail="target agent timeout") from exc except httpx.ConnectError as exc: raise HTTPException(status_code=502, detail="target agent unavailable") from exc

这里有一个设计取舍值得聊。网关为什么一定要同步等待目标Agent返回结果，而不是告诉调用方“请求已受理，之后回调你”？因为大部分Agent调用场景，调用方都需要立刻拿到结果来继续决策。客服Agent查订单状态，就是要等订单Agent返回“已支付”才能回答用户，异步反而增加复杂度。只有那些真正耗时长、不需要实时结果的任务，才应该退化成消息队列模式。

3.4 用 Docker Compose 一键启动

为了让整个项目能快速跑起来，我把依赖和服务打包成了一套Docker Compose。配置如下：

version: "3.8" services: redis: image: redis:7-alpine ports: - "6379:6379" reach: build: . environment: - REDIS_URL=redis://redis:6379/0 - REACH_TOKEN=${REACH_TOKEN:-dev-token} ports: - "8000:8000" depends_on: - redis

Dockerfile 也写得很简单：

FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY app ./app CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

这样本地部署就是一条命令的事：

docker compose up -d

我特别提醒一句：这个配置只是开发环境，生产环境里Redis绝对不要对外暴露端口，也不要让Agent-Reach直接监听公网。Redis端口一旦暴露，别人可以直接往你的注册表里塞假数据，或者更糟，利用云厂商的配置漏洞控制整个Agent网络。安全上宁严勿松。

4. 实操现场：让两个Agent互相触达

4.1 场景定义：客服Agent调用订单Agent

我拿一个真实场景来演示。假设系统里有两个Agent：order-agent负责订单查询和退款，它暴露了一个RPC端点来处理能力调用；cs-agent是客服助理，它自己不会直接访问订单数据库，需要通过Agent-Reach找到order-agent来获取订单状态。

订单Agent启动时，会向Agent-Reach发起注册请求。开发环境下我用curl模拟这个过程：

curl -X POST http://localhost:8000/v1/register \ -H "Content-Type: application/json" \ -H "Authorization: Bearer dev-token" \ -d '{ "agent_id": "order-agent", "name": "订单处理Agent", "endpoint": "http://order-agent:9001/rpc", "capabilities": ["order.query", "order.refund"] }'

注意这里的endpoint我写的是http://order-agent:9001/rpc，因为订单Agent的容器名是order-agent。在Docker Compose网络里，服务名就是可路由的地址，不需要知道具体IP。这也是为什么我说不要写localhost，容器环境里写localhost指向的是当前容器自己，必挂。

订单Agent本身是个简单的FastAPI服务，它只需要能处理JSON-RPC格式的POST请求，代码如下：

from fastapi import FastAPI, Request app = FastAPI() @app.post("/rpc") async def handle_rpc(request: Request): body = await request.json() method = body["method"] if method == "order.query": return { "jsonrpc": "2.0", "id": body["id"], "result": { "status": "paid", "updated_at": 1735689600 }, } return { "jsonrpc": "2.0", "id": body["id"], "error": {"code": -32601, "message": "method not found"} }

目标Agent不一定要完整实现JSON-RPC规范，只要能识别jsonrpc、id、method这几个字段，按固定结构返回，就足够和Agent-Reach配合了。

4.2 端到端调用与结果

现在客服Agent需要查一笔订单。它不需要知道订单Agent的地址，只需要知道能力的名字，然后请求Agent-Reach做转发：

curl -X POST http://localhost:8000/v1/call \ -H "Content-Type: application/json" \ -H "Authorization: Bearer dev-token" \ -d '{ "caller": "cs-agent", "target": "order-agent", "method": "order.query", "params": {"order_id": "20250101001"}, "request_id": "req_20250101_001", "deadline_ms": 3000 }'

返回结果：

{ "jsonrpc": "2.0", "id": "req_20250101_001", "result": { "status": "paid", "updated_at": 1735689600 } }

在这条看似简单的调用链背后，Agent-Reach实际做了四件事：从Redis中查出order-agent对应的端点；生成trace_id并写入日志；把JSON-RPC请求转发给订单Agent；等待响应并原样返回。整个过程不到几毫秒，如果算上订单Agent的查询时间，通常也能保持在几十毫秒内。

这个演示虽然简单，但它证明了核心链路是通的：能力注册、服务发现、鉴权校验、请求转发、结果回传，全部跑通。有了这个骨架，后面加能力路由、灰度权重、MCP适配，都只是往里面填策略的问题。

4.3 可以很自然扩展出来的能力

实际操作中我很快就发现，有了Agent-Reach这一层，很多功能扩展起来特别顺手。一个典型的例子是灰度发布。旧版订单Agent和新版订单Agent可以同时注册，网关在发现到多个候选时，按标签或者权重选择目标。比如按labels.version过滤出v2版本的Agent，让客服Agent只走新版逻辑，流量慢慢放量。

另一个我觉得很有价值的方向，是把外部工具也注册成“伪Agent”。工具本身不是Agent，但它可以声明自己有什么能力，提供一个标准的RPC端点，然后Agent-Reach把它和真正的Agent平等对待。这样外部API、内部服务、甚至另一个团队的服务，都能用同一套机制接入Agent网络。

更远的扩展是兼容MCP。Agent-Reach可以把自己暴露成一个Model Context Protocol服务，让任意支持MCP的客户端通过它调用整个Agent集群的能力。也可以反过来，把MCP服务器包装成Agent注册进来。这样，Agent-Reach就成了一个标准的“能力总线”，而不是一个自成一派的封闭协议。这个扩展工作量不大，价值却很直接。

5. 常见问题与排查实录

5.1 Agent注册成功，但调用一直超时

这是最常遇到的问题，而且多半是自己坑自己。现象是注册返回200，心跳也正常，但一发起调用就502或者504。我第一反应通常是怀疑网关转发逻辑，但排查到最后，九成都是注册时填的endpoint有问题。

最常见的两种场景。第一种，Agent和网关在同一台机器，注册时写localhost:9001，单机跑没问题，一旦搬到另一台机器或者容器，网关去访问localhost:9001就会被拒绝。第二种，Agent在Docker容器里，注册时错误地写了容器自身的IP，而不是网关能访问的宿主机IP或服务名。

排查步骤很简单，直接在网关所在的容器或主机上执行curl http://<agent_endpoint>/rpc，看能不能通。如果这个curl都失败，那就别查网关日志了，先去把endpoint改成正确可达的地址。

5.2 心跳断了，注册表里却有残留

这个问题出现的原因是TTL设置太长，或者Agent进程被强杀之前没有主动调用注销接口。比如心跳间隔5秒，TTL却设为10分钟，那么Agent已经挂掉很久，注册表里还认为它存活，调用请求会一直往死掉的端点打。

解决办法我说过多次：TTL设置为心跳间隔的3倍，这样最多错过两次心跳后记录就会过期；同时要求Agent优雅下线时调用/v1/unregister，主动从注册表移除。另外我还发现一个容易忽略的点：Redis的Hash过期后，有序集合里对应的member并不会自动删除。所以我在find_by_capability里遍历时做了兜底判断，发现Hash不存在就顺手执行一次zrem，避免残留数据一直占着集合。

5.3 请求体太大导致网关直接断开

Agent之间传输的内容不总是中小型JSON，有时候一个Agent要给另一个Agent传一段很长的文本片段或者分析报告，报文可能就有好几兆。此时网关或者前面的Nginx如果限制过请求体大小，就会出现连接被重置的诡异现象。

排查经验是这样的：如果调用小请求一切正常，只有传大文本时报错，先检查Nginx的client_max_body_size，再检查Agent-Reach服务的body大小限制。Python的Uvicorn默认不限制请求体大小，但FastAPI在解析JSON时会受内存约束，超大JSON解析也可能爆内存。

我最终的取舍是：RPC接口只承担结构化参数传输，不要拿来传大文件。需要传输大文本时，调用方先把内容存储在对象存储或临时文件服务里，参数里只传一个引用ID，目标Agent再去拉取。这样既避开了网关的压力，也让日志里的参数不再是一大坨冗余内容。

5.4 重复提交与消息乱序

跨Agent调用时，调用方很容易因为网络波动而超时重试。问题是，第一次请求可能已经到达目标Agent并在处理中，重试就会导致同样的事情被执行两次。这在查询操作上无所谓，但如果是创建订单、扣减库存这类写入操作，后果就很严重。

我的做法是引入幂等键。调用方在request_id里生成全局唯一的ID，并保存一段时间；目标Agent在处理请求前检查最近完成过的ID，如果处理过就直接返回上一次结果，不再重复执行。这个去重表可以放在Agent自己的Redis里，保留最近半小时的记录就够用。

至于消息乱序，我目前不建议在Agent-Reach层面做全局排序，成本和复杂度都不低。多Agent场景下，真正需要严格有序的往往是同一个会话内的连续动作，这应该由Agent内部的会话管理来保证，而不是依赖网关。排序责任下沉，网关保持无状态，这是更合理的分工。

5.5 关于定位边界，我的一些实在看法

用了Agent-Reach一段时间后，我最大的体会是：中间层这种组件，一定要克制。它的职责就是把连接打通，把流量带过去，把结果带回来，别什么都往里塞。一旦你开始往里面加工作流编排、状态管理、消息持久化，它就会变成一个越来越重的平台，最后反而拖累业务迭代。

我更推荐的做法是让Agent-Reach保持薄、保持简单。每个Agent仍然是独立的服务，网关只负责发现和转发。复杂的能力比如多步任务编排，应该放在更高层的编排Agent里，而不是下沉到通信层。通信层越轻，整个系统的演进速度越快。

最后说一个实打实的小技巧：给每个接入Agent-Reach的服务都打上app_name和version标签，日志里也加上这两个字段。刚开始可能觉得多余，等你的Agent数量上了两位数，排查问题时你就知道这两个字段有多值钱了。