企业官网建设流程全解析

1. 为什么需要 Agent-Reach：多智能体协作的触达难题

三五个 Agent 凑在一起干活的时候，问题往往不在单个 Agent 的模型能力有多强，而在一个特别基础但又特别要命的地方：它们根本没有办法稳定地找到对方、把话说清楚。

我自己最早做多 Agent 系统的时候，犯过一个挺典型的错。当时用一个大框架把三个 Agent 编排在一起，A 负责理解用户意图，B 负责查订单数据，C 负责生成回复。流程图画得很漂亮，但一跑起来就露馅了：A 调 B 的时候硬编码了一个服务地址，B 一重启换了端口，A 直接 502；C 要的订单上下文放在 B 的局部变量里，B 跑完任务就释放了，C 拿不到数据只能胡说；更别说几十分钟的会话里消息来回穿插，有时候 Agent 回复完却发现找不着这个消息属于哪个用户。

这些问题的根源都指向一件事：我们设计多 Agent 协作的时候，把大部分精力放在“流程怎么编排”上，却忽略了“消息怎么触达”。Agent-Reach 这个名字对应的正是这个“触达层”的建设。

简单说，Agent-Reach 解决的三个核心问题是：Agent 怎么被发现（能力注册）、消息怎么投递（路由与信封）、协作状态怎么共享（会话上下文）。它适合正在做 Multi-Agent 系统的工程师、负责 AI 平台架构的技术负责人，以及被一堆 Agent 互相调用搞得焦头烂额的实践者。对于刚入门的朋友，这篇文章也能帮你理解 Agent 协作中最容易被忽略的地基。

需要说明的是，Agent-Reach 本身并不是某个特定的开源项目或商业产品，我在这里是基于多年的多 Agent 工程实践，把它作为一个“智能体触达层”的最小实现思路来讲。整套方案你可以直接用 Python 复刻，也可以迁移到 Java、Go 等任何你熟悉的技术栈上。它的价值在于协议设计和架构思想，而不在于某一个特定的实现。

2. 核心架构拆解：触达层到底管什么

2.1 三个经典痛点反映的设计缺口

先说痛点，后面才好理解设计。多 Agent 系统在实际跑的时候，最常见的三个坑分别是：

第一，Agent 之间没有统一的能力描述。A 要调 B 的某个功能，要么写死 B 的 HTTP 接口路径，要么在代码里硬编码 B 的类名。一旦 B 的接口升级、路径变化，A 就得跟着改。更麻烦的是，系统里 Agent 一多，你很难搞清楚“到底哪个 Agent 能处理某种请求”，只能靠人工记忆去连线。

第二，消息没有“信封”。消息就是一把裸数据丢过去，没有目标标识、没有过期时间、没有优先级、没有请求唯一 ID。看起来无所谓，实际上在异步场景里非常致命。比如一个 Agent 处理任务耗时 30 秒，调用方早就超时放弃，但 Agent 还在继续跑；再比如两个请求同时发给同一个 Agent，返回结果根本对不上号。

第三，会话上下文是割裂的。多 Agent 协作的链路往往很长，A 收集的信息要传给 B，B 处理后传给 C。如果每个 Agent 都各自维护一份会话状态，两三个节点还行，超过五个节点基本就乱了；特别是 Agent 崩溃重启之后，自己的内存状态全没了，整个会话就只能从头再来。

Agent-Reach 的设计思路就是把这些散落在各处的“杂活”收拢到一个统一层。它不做业务逻辑，不做流程编排，只做四件事：注册、寻址、投递、上下文共享。你可以把它想象成公司里的前台：你不需要知道财务部在哪一层哪个工位，只需要告诉前台“我要报销”，前台负责把人带到正确的位置。

2.2 核心概念与信息模型

Agent-Reach 的模型一共五个核心要素，理解这五个就理解了整套设计。

首先是 Agent 注册信息，叫做 AgentManifest。每个 Agent 启动后必须向注册中心提交一份自己的“身份和技能说明”，包含名称、版本、所属团队、能力列表、联系方式、标签等。能力列表是关键，它决定了这个 Agent 能处理什么类型的请求。

其次是能力描述。每一项能力需要说明它的输入和输出是什么，我会用 JSON Schema 来描述，这样调用方不依赖具体编程语言也能知道该传什么参数。比如一个“查询订单”能力，输入是订单号和用户 ID，输出是订单状态和金额，结构一目了然。

第三是消息信封 Envelope。所有在 Agent 之间传递的消息都包在一个信封里。信封上写着：这个消息从哪来、要到哪去、是什么类型（请求还是响应还是事件）、属于哪个会话、什么时候过期、优先级多高、幂等键是什么。收件 Agent 不关心信封是怎么送到手上的，只需要拆开处理里面的内容。

第四是路由策略。当一条请求到达 Agent-Reach 时，由路由模块根据信封里写明的目标，结合 AgentManifest 里的能力描述，决定投递给哪个 Agent 实例。路由有三种模式：能力路由、标签路由、上下文路由，后面详细说。

第五是会话上下文池。所有跨 Agent 的共享数据都按 session_id 存在上下文池里，任何 Agent 需要的时候都可以去读取和更新，不需要彼此之间反复传递整份数据。上下文池带生存时间，会话结束后自动清理。

2.3 为什么不用现成的消息队列或 RPC 框架

你可能会问，这些问题 RabbitMQ、Kafka、gRPC 不都能解决吗，为什么要自己搞一套？

这个问题我在选型的时候仔细想过。消息队列解决的是异步消息的可靠传递，但它的路由模型是“主题订阅”，Agent 之间的触达往往是“按能力查找”，这两者并不完全等价。你用 Kafka 发消息，消费者得自己去订阅 topic，但多 Agent 场景里你常常不知道“哪个 Agent 有能力处理这个请求”，需要的是能力匹配而不是 topic 分组。当然，你可以用 MQ 作为底层传输，这是没有问题的，Agent-Reach 在传输层并不排斥 MQ，它额外增加的正是能力注册和路由决策这一层业务语义。

gRPC 和 HTTP 这类同步 RPC 也有问题：服务发现、负载均衡、超时重试都要自己搭一套，而且接口契约是编译期固定的，Agent 能力一变，所有调用方都要重新生成代码。Agent 场景的特点是变化快、能力多、语义复杂，需要一个运行时可查可改的注册与路由层。

所以 Agent-Reach 的定位是：在编排框架和 Agent 业务逻辑之间，补一层“触达管道”。编排层管“流程怎么走”，触达层管“消息怎么到”。两者配合，而不是互相替代。

3. 从零搭建 Agent-Reach：最小可用系统实现

3.1 设计思路与技术选型

为了让你能快速跑起来，我选了最简技术栈：Python 3.10 以上版本，FastAPI 做 HTTP 接口，Redis 做分布式注册中心和会话上下文存储（单机演示时可以用内存版，生产环境必须换 Redis）。

说实话，Redis 在这里是性价比非常高的选择，它天然支持键值过期，正好用来做注册信息的 TTL 续约；同时又支持简单的哈希结构，适合存会话上下文。如果你不想引入 Redis，也可以先用 Python 的字典模拟注册中心，但那就没法做多实例部署了。

整个系统分两部分：Agent-Reach Core，也就是触达层服务本身，负责接收请求、路由转发；Agent SDK，每个 Agent 启动时调用 SDK 完成注册和消息收发。这样分开的好处是，Agent 只需要关心自己的业务逻辑，触达的复杂度都被封装了。

3.2 注册中心实现：Agent 的能力名片

注册中心的核心逻辑很简单：接收 Agent 的注册请求，把 AgentManifest 存起来，提供按能力查询的接口。下面这个是注册中心的简化实现。

# registry.py import time import threading from typing import Dict, List, Optional class AgentRegistry: """注册中心：管理 Agent 的能力名片与健康状态""" def __init__(self, ttl: int = 30): self.agents: Dict[str, dict] = {} self.ttl = ttl self.lock = threading.Lock() def register(self, manifest: dict) -> bool: """注册或续约 Agent 信息""" agent_id = manifest.get("agent_id") if not agent_id: return False manifest["last_heartbeat"] = time.time() manifest["expires_at"] = time.time() + self.ttl with self.lock: self.agents[agent_id] = manifest return True def unregister(self, agent_id: str) -> bool: with self.lock: if agent_id in self.agents: del self.agents[agent_id] return True return False def find_by_capability(self, capability: str) -> Optional[dict]: """按能力名查找在线 Agent，返回最新注册的一个""" now = time.time() candidates = [] with self.lock: for agent in self.agents.values(): if agent["expires_at"] < now: continue caps = agent.get("capabilities", []) if any(c.get("name") == capability for c in caps): candidates.append(agent) if not candidates: return None # 简单负载策略：取心跳时间最新的 candidates.sort(key=lambda a: a["last_heartbeat"], reverse=True) return candidates[0] def _cleanup_expired(self): """定时清理过期 Agent""" while True: time.sleep(5) now = time.time() with self.lock: expired = [k for k, v in self.agents.items() if v["expires_at"] < now] for k in expired: del self.agents[k]

这份代码最关键的细节是 TTL 机制。每个 Agent 注册后默认 30 秒无人续约就会被清理。为什么要做过期？因为 Agent 可能会崩溃、被 kill、或网络分区，不能靠 Agent 主动注销，必须有一个“被动健康检测”。那些做过服务注册的朋友应该很熟悉，这就是微服务里心跳续约的思路。

在实际生产中，Agent 应该每隔 10 秒调一次注册接口续约。如果整个服务重启，它必须先主动注销旧的注册信息，再重新注册，否则会有短暂的双实例存在。

3.3 消息信封：触达的通用语言

信封是整套协议里最不能省的部分。我见过很多 Agent 系统跑着跑着消息就对不上号了，原因就是没有信封。当你只传 payload 的时候，消息的“元信息”就丢了，这些元信息恰恰是路由判断的基础。

# envelope.py import uuid from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Optional @dataclass class Envelope: """Agent 之间的消息信封""" kind: str # request / response / event from_agent: str to_agent: Optional[str] # 逻辑地址或能力名 session_id: str payload: Any expires_at: Optional[float] = None priority: int = 5 # 1-10，数值越高越优先 id: str = field(default_factory=lambda: uuid.uuid4().hex) trace_id: str = field(default_factory=lambda: uuid.uuid4().hex) idempotency_key: Optional[str] = None def is_expired(self) -> bool: if self.expires_at is None: return False return time.time() > self.expires_at def build_envelope(from_agent: str, to_target: str, session_id: str, payload: Any, kind: str = "request", ttl_seconds: int = 60) -> Envelope: """构建一个带默认过期时间的信封""" return Envelope( kind=kind, from_agent=from_agent, to_agent=to_target, session_id=session_id, payload=payload, expires_at=time.time() + ttl_seconds )

信封里要特别注意两个字段。第一个是 trace_id，它贯穿整条调用链。排查问题的时候，你只需要拿着 trace_id 去日志系统里搜，就能看到这个请求在哪些 Agent 之间流转过，谁处理的、用了多久、在哪一步出了错。第二个是 idempotency_key，用于幂等重试。如果业务允许，调用方重试同一个请求时带上同一个 idempotency_key，接收方就能根据这个键判断是否已经处理过，避免重复下单、重复扣款这类脏数据。

3.4 路由引擎：三种策略按需选择

路由引擎接收一个 Envelope，根据 to_agent 字段的性质来选择策略。

# router.py import re from typing import Optional class Router: def __init__(self, registry: AgentRegistry): self.registry = registry def route(self, envelope: Envelope) -> Optional[dict]: """根据信封目标进行路由，返回目标 Agent（若不存在则返回 None）""" target = envelope.to_agent if not target: return None # 策略1：精确能力名路由，如 capability:query_order if target.startswith("capability:"): cap_name = target.split(":", 1)[1] return self.registry.find_by_capability(cap_name) # 策略2：标签路由，如 tag:finance if target.startswith("tag:"): tag = target.split(":", 1)[1] return self._find_by_tag(tag) # 策略3：精确 Agent ID 路由 return self.registry.get(target) def _find_by_tag(self, tag: str) -> Optional[dict]: candidates = [] for agent in self.registry.agents.values(): agent_tags = agent.get("tags", []) if tag in agent_tags: candidates.append(agent) if not candidates: return None candidates.sort(key=lambda a: a["last_heartbeat"], reverse=True) return candidates[0]

三种策略各有适用的场景。能力路由是主力，调用方不关心谁来处理，只关心“有这个能力的 Agent 是谁”，解耦效果最彻底。标签路由适合灰度发布和分组隔离，比如只给“finance”组的 Agent 发财务类任务。精确路由适合明确的点对点通讯，比如一个 Agent 必须由指定实例处理时。

需要说明的是，这里的路由策略和网络路由不是一回事，它实际上是更上层的“业务路由”，解决的是“谁有能力做这件事”的语义寻址，而不是 IP 层的包转发。生产环境里两者是配合关系，Agent-Reach 完成业务语义寻址之后，实际的数据传输仍然走 HTTP、gRPC 这些常规协议。

3.5 接入真实 Agent：用 LLM 实现一个订单客服

路由问题解决了，接下来要看 Agent 是怎么接入触达层的。我以目前最常见的 LLM Agent 为例，写一个“订单客服 Agent”，它能查询订单状态、处理退货请求。下面这段代码演示了 Agent 启动时如何注册、以及如何收到请求后调用 LLM 处理。

# agent_http_server.py import os import time import requests from fastapi import FastAPI, HTTPException LLM_API_KEY = os.getenv("LLM_API_KEY") LLM_URL = os.getenv("LLM_URL", "https://api.openai.com/v1/chat/completions") REGISTRY_URL = "http://localhost:8000/register" agent_id = "order_agent_v1" app = FastAPI() # 1. 启动后自动注册 manifest = { "agent_id": agent_id, "name": "order_agent", "version": "1.0.0", "capabilities": [ {"name": "query_order", "input": "order_id: str, user_id: str", "output": "order_status: str"}, {"name": "return_order", "input": "order_id: str, reason: str", "output": "return_id: str"} ], "tags": ["ecommerce", "order"], "endpoint": "http://agent-host:9001/process" } requests.post(REGISTRY_URL, json=manifest) @app.post("/process") def handle_request(envelope: dict): if envelope["to_agent"] == "capacity:query_order": result = call_llm( "你是订单客服，请根据订单信息回答客户的问题。", envelope["payload"] ) else: result = call_llm( "你是退货专员，请礼貌确认退货原因并生成退货单号。", envelope["payload"] ) return {"ok": True, "result": result} def call_llm(system_msg, user_msg, trace_id=""): """调用大模型生成回复，设置 trace_id 便于链路追踪""" headers = { "Authorization": f"Bearer {LLM_API_KEY}", "Content-Type": "application/json" } body = { "model": "gpt-4o-mini", "messages": [ {"role": "system", "content": system_msg}, {"role": "user", "content": str(user_msg)} ], "temperature": 0.2, "metadata": {"trace_id": trace_id} } resp = requests.post(LLM_URL, headers=headers, json=body, timeout=30) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"]

这段代码里有一个很关键的点：Agent 的 endpoint 是注册到注册中心的，调用方永远不需要知道这个地址，它只会往 Agent-Reach 发一条“我要查订单”的消息，剩下的都由触达层解决。

3.6 触达层入口：请求转发完整链路

最后是触达层自身的 HTTP 入口，它负责接收外部请求、做路由、转发给目标 Agent，并且处理超时。

# agent_reach_core.py import httpx from fastapi import FastAPI, Request app = FastAPI() router = Router(AgentRegistry()) @app.post("/invoke") async def invoke(envelope: dict): """外部系统通过这个接口发起一次 Agent 调用""" # 1. 检查信封是否过期 if envelope["expires_at"] and time.time() > envelope["expires_at"]: return {"error": "envelope expired"} # 2. 路由查找目标 Agent target_agent = router.route(envelope) if target_agent is None: return {"error": "no available agent for capability", "envelope": envelope} # 3. 转发请求到目标 Agent async with httpx.AsyncClient(timeout=30) as client: response = await client.post(target_agent["endpoint"], json=envelope) return {"result": response.json()}

到这里，一个最小可用的 Agent-Reach 已经能跑通了。一个请求从进入到返回，经过了“信封校验 → 能力路由 → 目标转发”三步，Agent 本身不需要感知调用方是谁、在哪、怎么发起的。

4. 生产环境实战：把 Agent-Reach 跑稳的关键细节

4.1 注册性能优化：本地缓存与变更推送

演示版直接查 Redis 没有问题，但生产环境里每一次路由都穿透到 Redis 就会成为性能瓶颈。注册信息的变更频率其实非常低（Agent 启动、崩溃、缩容时才变），所以适合做本地缓存。

我们实测下来，单机 Agent-Reach 每秒要处理几百次路由请求，如果每次都去 Redis GET，延迟会多出 0.5~2ms。这个量级看起来不大，但并发一高，再加上网络抖动，就很容易触发超时。优化方案是在 Agent-Reach 内存中维护一份注册表副本，每隔 15 秒从 Redis 全量同步一次，同时监听 Key 过期事件做增量更新。注意，缓存时间不能太长，否则 Agent 已经崩溃 30 秒了，本地还在往它那里发请求。

另外，多实例部署 Agent-Reach 时，要确保所有实例的缓存最终一致。最简单的方式是依赖 Redis 的通知通道，Agent 注册或注销时推送变更消息到指定频道，所有 Agent-Reach 实例订阅这个频道后更新本地缓存。

4.2 投递可靠性：重试、熔断与降级

Agent 是软件程序，总会出错。生产环境里必须先假设 Agent 会失败，再来设计投递机制。我的建议是三个等级：

第一级是超时重试。一个请求发给 Agent 后，等响应的时间限制要足够宽裕——普通 Agent 可以给 15 秒，但那种要调用大模型的 Agent 建议给到 60 秒以上。如果你只给了 5 秒，LLM 还在推理，重试就已经开始跑了，反而加重系统负担。重试策略要避免同时发多个副本，应该用指数退避，比如第一次重试间隔 1 秒，第二次 2 秒，第三次 4 秒，最多重试 3 次。

第二级是熔断。当某个 Agent 连续失败次数超过阈值（比如 7 次），就要把它从注册中心临时摘除，停止向它转发新流量，同时进入冷却期。冷却期结束后放行少量流量试探恢复情况，如果成功了就恢复正常权重，继续失败就继续冷却。熔断是为了保护下游，避免 Agent 已经故障还一直向它发送一大堆请求，把它彻底打挂。

第三级是降级。当找不到合适的 Agent 时，Agent-Reach 不应该直接抛错，而应该返回一个标准化的“no agent available”消息。调用方收到这个错误后可以走自己的降级逻辑，比如展示“客服繁忙，请留言”而不是直接报 500。这对线上用户体验非常重要。

4.3 安全与访问控制：不能让 Agent 裸奔

触达层是所有 Agent 之间通讯的十字路口，安全设计必须从第一天开始做。

认证与鉴权是必须的。Agent-Reach 在处理 /invoke 请求时，先校验调用方提供的 API Key 是否有效，同时检查调用的能力是否在调用方的授权范围内。实现上可以为每个调用方分配一个 Key，并且绑定一份允许调用的能力清单。这样即使某个调用方的代码被攻击者拿到，它也只能调用被允许的那几个能力，没办法轻易横向移动。

敏感能力的管控值得单独讲。比如“转账”“删除数据”“管理权限”这类高危能力，应该额外开启二次确认。合理做法是给能力清单增加一个 sensitive: true 属性，Agent-Reach 发现请求的是敏感能力时，强制要求调用方在信封里附带额外的审批凭证，比如上级系统签发的短期 token。这样就算调用方的 API Key 被盗，对方也调不了高危能力。

消息内容脱敏也不能忽视。在日志中记录请求 payload 时，要自动遮蔽身份证号、手机号、支付账号等敏感字段。实现在写入日志前跑一个脱敏模板，把匹配到的敏感模式替换成***。这个坑我踩过，一旦生产日志里泄露了用户信息，整改成本会非常被动，所以越早做越安全。

5. 常见故障与排查实录

5.1 Agent 明明注册成功，但触达层总是找不到它

这是最常见的问题。现象是你在注册中心能看到 Agent 的信息，但路由结果始终是 None。我们遇到过的原因有三种：

第一种是能力名大小写不一致。注册的时候写的是queryOrder，调用方请求的是query_order，匹配不上。建议统一改成下划线命名规范，同时在注册时把能力名自动转小写。

第二种是 TTL 过期没续约。如果 Agent 注册后忘了做定时心跳续约，注册信息会在 30 秒后被清理掉。你查的时候正好过了有效期，自然找不到。排查方法是看这段时间里 Agent 有没有日志报错“500 on heartbeat”。

第三种是 Agent 多实例部署时，新的实例还在启动过程中，但旧的实例已经将注册信息续约成了自己的地址，而注册中心里保留了旧实例的过期缓存。这属于并发注册的脏数据问题，解决办法是注册时带上版本号，路由时优先选版本号最新的。

5.2 多个会话的上下文串场，Agent 回答错乱

这个问题出现的时候会让整个系统看起来像“失控”。现象是用户 A 查订单，Agent 回答里却包含用户 B 的订单信息。

几乎每次排查下来，根因都是 session_id 被重复使用。比如你从业务系统传入 session_id 时用了用户的订单号，同一个用户的两个不同会话就共享了一个上下文池，消息全部搅在一起。正确做法是内部统一生成 UUID 作为 session_id，只在入口处记录业务系统原始会话 ID 和内部 session_id 的映射关系。

还有一个隐蔽的原因：上下文池的过期时间设得太长。设计上是会话结束 10 分钟后自动清理，但如果设成 24 小时，同一个 Agent 复用时就会读到上一次会话的数据。排查这个问题的第一步永远是把上下文池里的数据打印出来，检查 session_id 是否一致。

5.3 Agent 响应超时，但 Agent 本身没挂

刚上线的时候我们遇到过一种诡异情况：Agent 日志显示处理只要 2 秒，但调用方总报 5 秒超时。后来发现是 LLM 推理的同步阻塞问题。

当 Agent 内部调用大模型时，如果用的是阻塞式 HTTP 请求，并且 Agent 没有启用线程池或异步处理，它只能同时处理一个请求。其他请求全都阻塞在队列里排队，自然就超时了。解决办法是给 Agent 加并发能力，比如用 FastAPI 的异步接口配合 httpx.AsyncClient 并发调用 LLM，或者引入 Worker Pool 模式。

另外，如果 Agent 和 LLM 之间的网络链接积压了大量 TIME_WAIT 连接，也会导致新的请求建立连接变慢。这个属于基础网络调优，调整内核参数可以把 TIME_WAIT 复用打开，也可以改用连接池。

5.4 生产环境重启后，整个触达层“失忆”

注册中心的信息全部存在 Redis 里，看起来不会丢。但有一种情况：Redis 本身做了持久化、数据没丢，而 Agent-Reach 重启后，之前所有在线 Agent 的注册信息仍然有效，可 Agent 自己的进程却没了。等于注册信息是好的，但 Agent 实际已经不存在了。

这个问题要让 Agent 的优雅退出机制来解决。Agent 进程接受到终止信号 SIGTERM 时，先主动调用 unregister 接口把自己从注册中心摘掉，再处理完手头的请求退出。同时，Agent 启动脚本里要等待注册成功、以及确保不再持有旧实例的过期信息之后，再开始接收流量。

6. 另一个容易被忽略的核心：会话上下文的存活设计

前面提过上下文池，这里专门展开讲。因为很多 Agent 系统里，崩溃的不是路由，而是状态。

我在设计 Agent-Reach 的上下文模型时，参考了微服务架构里“状态外部化”的思路。Agent 本身应该是尽量无状态的，它需要的所有业务数据都从上下文池拉，处理完再把结果写回去。这样 Agent 实例就算崩溃重启，只要 session 还在，它就能从上下文池里恢复工作状态，用户几乎无感知。

上下文池的实现建议用 Redis Hash，key 是 session_id，field 是上下文中的各个数据片段名称，value 是数据内容。比如一个电商客服会话的上下文可以是：

session_abc123 = { "user_id": "u_8888", "cart_items": ["item_1", "item_2"], "selected_address": "浙江省杭州市...", "intent": "checkout", "step": "confirm_address" }

每个 Agent 在处理请求时，先 GET 取它需要的字段，处理完后 SET 更新字段。这里要小心并发写的问题：两个 Agent 同时在更新同一个 session 的不同字段，如果用正常 Redis 写入，后面的写入可能覆盖前面的。解决办法是让 Agent 写入时采用 hash set，而不是整体替换 session 的 JSON。

上下文的生命周期管理同样重要。所有 session 都应该有 TTL，延迟清理是可接受的，但绝不能留下垃圾数据。我一般给默认会话设置 30 分钟空闲过期，每次有请求访问就续期。比如用户买完东西 20 分钟后又回来查看进度，会话还能接上；但如果他三天之后才回来，早就该开新会话了。

7. 写在最后：触达层是 Agent 系统的水电煤

做多 Agent 做了这么长时间，我的体会是：大家总是在追逐更高的智能、更复杂的编排，但真正让系统“能干活”的，反而是一堆不起眼的基础设施。

Agent-Reach 就是这套基础设施中的一个关键拼图。它不生产智能，不编排流程，只做一件事：让 Agent 之间可以像人与人一样，通过前台转接而不是私加微信的方式协作。业务上，这意味着你可以随时上线一个新的 Agent，替换一个老 Agent，所有调用方都不需要改代码；出现问题时，你可以拿着 trace_id 快速定位到底是哪个环节掉了链子。

根据我个人的经验，搭建这套触达层的时候最需要注意的就是“保持薄”。很多人做着做着就会往里面塞业务逻辑、塞流程控制、塞数据清洗，结果触达层变成了一个大泥潭。记住它的职责边界：注册、路由、投递、共享上下文，仅此而已。超出这个范围的，都该放在 Agent 业务层或编排层。

如果你正准备从两三个 Agent 的玩具项目走向真正的生产系统，建议你从今天开始把“触达”当作一个正式模块来设计。等你试过 Agent 数量超过十个之后，再回头看就会明白，早一点建这层水电煤，是最省钱的决定。