企业官网建设流程全解析

1. 项目缘起：当两个AI Agent“军团”需要协同作战时

最近在折腾AI Agent的落地应用，很多团队都面临一个现实问题：手头可能同时运行着不止一套Agent系统。比如，我们团队内部就同时维护着基于OpenClaw和基于Hermes的两套Agent集群。OpenClaw那边，我们用它来处理一些需要深度集成内部知识库和复杂工作流的任务，比如自动化的代码审查和文档生成；而Hermes这边，则更擅长处理一些需要快速响应、基于对话的客服或信息查询场景。两套系统各有侧重，也各自积累了不少业务逻辑和状态数据。

最初的想法很简单，能不能搞个“大一统”，把其中一套迁移到另一套上，或者干脆开发一个超级Agent来包揽所有事？但实际操作起来，发现这想法太天真了。首先，**“不卸载”是底线。两套系统都承载着线上业务，贸然停服或迁移，风险不可控，业务方第一个不答应。其次，“不迁移”是现实。代码、数据、模型权重、运行状态，迁移成本高得吓人，而且两套框架的设计哲学和API接口差异巨大，强行融合等于重写。最后，我们还需要“可灰度”**的能力，新来的请求，是给OpenClaw还是给Hermes？能不能根据流量、业务类型或者用户身份做动态路由？甚至，未来引入第三套、第四套Agent时，这个架构还能不能平滑扩展？

于是，“用一个Router把两套Agent‘编成一队’”的想法就诞生了。这不是要做一个能理解所有Agent内部逻辑的“大脑”，而是做一个高效的“调度中心”或“交通警察”。它的核心目标非常明确：对外提供一个统一的入口，对内根据一套清晰的规则，将用户请求智能地分发到后端的OpenClaw或Hermes集群，并且整个过程对后端Agent透明，无需它们做任何改造。听起来是不是有点像微服务里的API网关？没错，思想是相通的，但我们要面对的是更具“个性”的AI Agent——它们的输入输出格式可能不同，会话状态管理方式各异，甚至错误处理机制都千差万别。

接下来，我就结合我们团队的实践，从头到尾拆解一下这个“混合集群Router”的设计与实现。你会发现，它不涉及高深的AI算法，更多的是对工程架构、通信协议和运维思维的考验。

2. 架构核心：Router的定位与关键设计决策

在深入代码之前，我们必须先想清楚这个Router到底要干什么，以及为什么这么干。这决定了整个项目的技术选型和复杂度。

2.1 Router的四大核心职责

我们的Router被设计成一个独立的服务，它承担了以下关键角色：

1. 统一入口与协议适配器：对外暴露一个统一的API端点（比如/v1/chat/completions），接收标准格式的请求（例如OpenAI兼容格式）。然后，它需要将这份请求“翻译”成后端OpenClaw或Hermes Agent能够理解的格式。反之，将后端Agent的响应再“翻译”回统一的格式返回给客户端。这是实现“不迁移”的基础。
2. 智能路由决策器：这是Router的“大脑”。它需要根据预设的规则，决定当前请求应该发给哪个Agent集群。规则可以非常简单（如基于请求路径中的某个标识），也可以非常复杂（如基于请求内容进行意图识别、基于负载情况做动态权重分配）。
3. 流量管控与灰度发布器：为了实现“可灰度”，Router必须支持精细化的流量切分。例如，可以通过用户ID哈希、请求头中的特定字段（如x-tenant-id）或者一个百分比，将流量按比例分发给不同的后端。新上线的Agent版本可以先接收1%的流量进行验证。
4. 容错与降级卫士：当某个后端Agent集群出现故障、响应超时或返回错误时，Router不能跟着一起挂掉。它需要具备重试机制（可能重试到另一个健康的实例）、故障转移（Failover）到备用集群，以及最终的服务降级策略（例如返回一个友好的错误提示，或者将请求转发给一个功能简化的保底Agent）。

2.2 为什么选择独立Router，而不是修改Agent？

这是一个根本性的设计抉择。我们评估过几种方案：

• 方案A：魔改Agent，让它们互相认识。在OpenClaw里写调用Hermes的代码，反之亦然。这很快被否决，因为它破坏了系统的解耦性，让两个本应独立演进的系统产生了强耦合，未来维护将是噩梦。
• 方案B：在客户端做路由。让每个调用方自己决定调用哪个服务。这增加了客户端的复杂性，且无法集中管理路由策略和灰度规则，不利于统一运维。
• 方案C：独立的Router服务。这正是我们选择的路径。它的优势非常明显：
  • 对后端透明：OpenClaw和Hermes无需任何修改，照常运行。
  • 关注点分离：路由逻辑、协议转换、流量治理等横切关注点被集中到一个服务中，易于维护和升级。
  • 灵活扩展：未来新增第三个Agent（比如Claude Agent），只需要在Router中增加相应的适配器和路由规则即可。
  • 统一管控面：所有流量都经过Router，便于我们做统一的监控、日志收集、限流和审计。

基于这些考虑，一个独立的、轻量级但功能强大的Router服务成为了不二之选。

2.3 技术栈选型：平衡性能、生态与开发效率

技术选型需要围绕Router的职责展开。我们主要考虑了以下几点：

• 高性能与高并发：Router作为所有流量的入口，不能成为性能瓶颈。需要支持异步非阻塞I/O来处理大量并发请求。
• 丰富的HTTP/WebSocket生态：AI Agent的交互通常基于HTTP（同步）或WebSocket（异步流式输出），框架需要对此有良好支持。
• 配置化与动态更新：路由规则、后端服务地址等最好能通过配置文件或配置中心管理，并支持热更新，避免频繁重启服务。
• 成熟的中间件与插件生态：便于快速集成认证、限流、日志、指标收集等功能。

结合团队技术背景和社区活跃度，我们最终选择了FastAPI作为Web框架。它基于Starlette（异步），性能出色，自动生成OpenAPI文档，而且编写API非常简单直观。对于需要更底层控制或追求极致性能的场景，Go + Gin/Echo或Rust + Axum也是绝佳选择，但FastAPI在开发速度和Python生态集成上对我们更有吸引力。

对于配置管理，我们使用了Pydantic Settings来管理环境变量和配置文件，并计划未来集成Consul或Nacos来实现后端服务地址的动态发现与健康检查。

3. 实现详解：从零搭建一个生产可用的Router

理论说再多，不如一行代码。下面我们就来一步步实现这个Router的核心模块。我会省略掉项目初始化、依赖安装（fastapi,httpx,pydantic-settings等）这些基础步骤，直接切入核心逻辑。

3.1 定义统一的数据模型与协议

首先，我们要定义Router对外和对内通信的数据格式。对外，我们选择兼容OpenAI API格式，这几乎是业界的“通用语”。

# schemas.py from pydantic import BaseModel, Field from typing import List, Optional, Union, Dict, Any class UnifiedChatMessage(BaseModel): role: str # "system", "user", "assistant" content: str class UnifiedChatRequest(BaseModel): """Router对外暴露的统一请求格式""" model: str = Field(default="gpt-3.5-turbo") # 这里可用来传递路由线索，如"openclaw-v1" messages: List[UnifiedChatMessage] stream: bool = False temperature: Optional[float] = None max_tokens: Optional[int] = None # ... 其他OpenAI兼容参数 router_hint: Optional[str] = None # 自定义字段，用于强制指定路由目标 class UnifiedChatResponse(BaseModel): """Router对外返回的统一响应格式（非流式）""" id: str object: str = "chat.completion" created: int model: str choices: List[Dict[str, Any]] usage: Dict[str, int] # 流式响应的数据块模型（此处简化，实际需遵循OpenAI流式协议） class UnifiedChatStreamResponseChunk(BaseModel): # ... 流式响应格式定义

同时，我们需要定义后端Agent的配置模型。这里假设OpenClaw和Hermes都有其特定的HTTP API端点。

# config.py from pydantic_settings import BaseSettings from typing import List class AgentBackendConfig(BaseSettings): name: str # "openclaw", "hermes" api_base: str # 后端服务的Base URL，如 "http://openclaw-svc:8000" api_path: str # 具体的聊天端点，如 "/v1/chat/completions" health_check_path: str = "/health" # 健康检查端点 timeout: int = 30 # 请求超时时间（秒） max_retries: int = 1 # 失败重试次数 weight: int = 50 # 负载均衡权重 enabled: bool = True # 是否启用 # 该后端特有的请求头或认证信息 headers: Dict[str, str] = {} # 用于协议转换的适配器类名 adapter: str = "default" class RouterConfig(BaseSettings): backends: List[AgentBackendConfig] default_backend: str = "openclaw" # 默认路由的后端 # 路由策略配置 routing_strategy: str = "rule_based" # "rule_based", "load_balance", "content_aware" rule_based_config: Optional[Dict[str, str]] = None # 规则映射，如 {"model:openclaw-*": "openclaw"} # 灰度发布配置 canary_config: Optional[Dict[str, Any]] = None # 服务降级配置 fallback_backend: Optional[str] = "hermes" # 主后端失败时的降级目标 class Config: env_prefix = "ROUTER_"

3.2 构建协议适配器层

这是Router中最“脏”但也最核心的部分。因为OpenClaw和Hermes的API很可能不一致。

# adapters.py from abc import ABC, abstractmethod import httpx from schemas import UnifiedChatRequest, UnifiedChatResponse from config import AgentBackendConfig class BaseAdapter(ABC): """所有协议适配器的基类""" def __init__(self, backend_config: AgentBackendConfig): self.config = backend_config self.client = httpx.AsyncClient( base_url=self.config.api_base, timeout=self.config.timeout, headers=self.config.headers ) @abstractmethod async def transform_request(self, unified_request: UnifiedChatRequest) -> Dict[str, Any]: """将统一请求转换为后端特定格式""" pass @abstractmethod def transform_response(self, backend_raw_response: httpx.Response) -> UnifiedChatResponse: """将后端原始响应转换为统一响应格式""" pass async def health_check(self) -> bool: """执行健康检查""" try: resp = await self.client.get(self.config.health_check_path) return resp.status_code == 200 except Exception: return False async def close(self): await self.client.aclose() class OpenClawAdapter(BaseAdapter): """OpenClaw后端适配器（假设其API与OpenAI略有不同）""" async def transform_request(self, unified_request: UnifiedChatRequest) -> Dict[str, Any]: # 示例：OpenClaw可能需要一个不同的字段名或结构 transformed = { "conversation": [{"role": msg.role, "text": msg.content} for msg in unified_request.messages], "stream": unified_request.stream, "parameters": { "temperature": unified_request.temperature, "max_new_tokens": unified_request.max_tokens, } } # 可能还需要处理OpenClaw特有的参数 return transformed def transform_response(self, backend_raw_response: httpx.Response) -> UnifiedChatResponse: resp_data = backend_raw_response.json() # 将OpenClaw的响应格式转换为UnifiedChatResponse # 这里需要根据OpenClaw的实际响应结构进行映射 unified_choice = { "index": 0, "message": { "role": "assistant", "content": resp_data.get("reply", ""), }, "finish_reason": "stop", } return UnifiedChatResponse( id=f"chatcmpl-{backend_raw_response.headers.get('x-request-id', '')}", created=int(time.time()), model="openclaw", choices=[unified_choice], usage={"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0} # 实际应从后端获取 ) class HermesAdapter(BaseAdapter): """Hermes后端适配器（假设其API完全兼容OpenAI）""" async def transform_request(self, unified_request: UnifiedChatRequest) -> Dict[str, Any]: # Hermes兼容OpenAI，直接返回字典即可 return unified_request.dict(exclude_none=True) def transform_response(self, backend_raw_response: httpx.Response) -> UnifiedChatResponse: # Hermes返回的就是OpenAI格式，直接解析 resp_data = backend_raw_response.json() return UnifiedChatResponse(**resp_data) # 适配器工厂 class AdapterFactory: _adapters = { "openclaw": OpenClawAdapter, "hermes": HermesAdapter, "default": HermesAdapter, # 默认使用Hermes（OpenAI兼容）格式 } @classmethod def get_adapter(cls, backend_config: AgentBackendConfig) -> BaseAdapter: adapter_class = cls._adapters.get(backend_config.adapter, cls._adapters["default"]) return adapter_class(backend_config)

注意：实际的协议转换可能比示例复杂得多，特别是错误码映射、流式响应处理等。这里的关键是抽象出BaseAdapter，让每种后端的差异被隔离在各自的适配器中，便于维护和扩展。

3.3 实现路由决策引擎

路由逻辑是Router的“大脑”。我们实现了多种策略，并通过配置驱动。

# router_engine.py import hashlib import random from typing import Optional, Dict, Any from schemas import UnifiedChatRequest from config import RouterConfig, AgentBackendConfig class RoutingEngine: def __init__(self, config: RouterConfig, backends: Dict[str, AgentBackendConfig]): self.config = config self.backends = backends self._strategy_map = { "rule_based": self._route_by_rule, "load_balance": self._route_by_load_balance, "content_aware": self._route_by_content, # 简单示例，实际可能用模型判断 } async def decide_backend(self, request: UnifiedChatRequest, **kwargs) -> Optional[str]: """决定请求应该发送到哪个后端""" # 1. 最高优先级：请求中明确指定的路由提示 if request.router_hint and request.router_hint in self.backends: return request.router_hint # 2. 根据配置的路由策略决策 strategy_func = self._strategy_map.get(self.config.routing_strategy, self._route_by_rule) backend_name = await strategy_func(request, **kwargs) # 3. 灰度发布逻辑（Canary Release） if backend_name and self.config.canary_config: backend_name = self._apply_canary_routing(request, backend_name) # 4. 检查后端是否可用 if backend_name and self.backends[backend_name].enabled: return backend_name # 5. 返回默认后端或None return self.config.default_backend if self.config.default_backend in self.backends else None def _route_by_rule(self, request: UnifiedChatRequest, **kwargs) -> Optional[str]: """基于规则的静态路由""" if not self.config.rule_based_config: return None # 示例规则：根据请求中的model字段匹配 model = request.model for pattern, backend in self.config.rule_based_config.items(): if pattern.endswith('*') and model.startswith(pattern[:-1]): return backend elif pattern == model: return backend return None def _route_by_load_balance(self, request: UnifiedChatRequest, **kwargs) -> Optional[str]: """基于权重的负载均衡路由""" enabled_backends = [b for b in self.backends.values() if b.enabled] if not enabled_backends: return None # 简单的权重随机选择 total_weight = sum(b.weight for b in enabled_backends) r = random.uniform(0, total_weight) upto = 0 for backend in enabled_backends: upto += backend.weight if upto >= r: return backend.name return enabled_backends[0].name def _route_by_content(self, request: UnifiedChatRequest, **kwargs) -> Optional[str]: """基于内容的路由（示例：根据用户问题关键词）""" # 这是一个非常简单的示例，生产环境可能需要用更复杂的NLP模型 user_content = "" for msg in request.messages: if msg.role == "user": user_content = msg.content.lower() break if any(keyword in user_content for keyword in ["代码", "编程", "bug"]): return "openclaw" # 假设OpenClaw擅长代码相关 elif any(keyword in user_content for keyword in ["客服", "帮助", "怎么"]): return "hermes" # 假设Hermes擅长客服对话 return None def _apply_canary_routing(self, request: UnifiedChatRequest, candidate_backend: str) -> str: """应用灰度发布规则""" canary_conf = self.config.canary_config # 示例1：基于用户ID的百分比灰度 user_id = kwargs.get('user_id') or hash(request.messages[0].content) % 100 canary_percentage = canary_conf.get('percentage', 0) canary_backend = canary_conf.get('backend') if canary_backend and canary_backend in self.backends and user_id < canary_percentage: return canary_backend return candidate_backend

3.4 组装主服务与API端点

最后，我们将所有组件在FastAPI应用中组装起来。

# main.py from fastapi import FastAPI, HTTPException, Request, status from fastapi.responses import StreamingResponse import asyncio import time import json from contextlib import asynccontextmanager from config import RouterConfig, AgentBackendConfig from adapters import AdapterFactory from router_engine import RoutingEngine from schemas import UnifiedChatRequest, UnifiedChatResponse # 全局变量（实际生产环境应使用更好的状态管理，如依赖注入） router_config = RouterConfig() backends_map = {b.name: b for b in router_config.backends} adapters_map = {} routing_engine = RoutingEngine(router_config, backends_map) @asynccontextmanager async def lifespan(app: FastAPI): # 启动时初始化所有适配器 for name, backend_conf in backends_map.items(): adapters_map[name] = AdapterFactory.get_adapter(backend_conf) yield # 关闭时清理资源 for adapter in adapters_map.values(): await adapter.close() app = FastAPI(lifespan=lifespan, title="AI Agent混合集群路由器") @app.post("/v1/chat/completions") async def chat_completion(request: UnifiedChatRequest, fastapi_req: Request): """ 统一的聊天补全端点。 1. 接收标准格式请求。 2. 路由决策。 3. 协议转换并转发。 4. 返回统一格式响应。 """ # 1. 提取可能用于路由的上下文（如用户ID） user_id = fastapi_req.headers.get("x-user-id") # 2. 路由决策 target_backend_name = await routing_engine.decide_backend(request, user_id=user_id) if not target_backend_name: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="No available backend service found." ) target_adapter = adapters_map.get(target_backend_name) if not target_adapter: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Adapter for backend '{target_backend_name}' not initialized." ) # 3. 健康检查（可选，可缓存结果定期检查） # is_healthy = await target_adapter.health_check() # if not is_healthy: # # 触发故障转移逻辑 # target_backend_name = router_config.fallback_backend # target_adapter = adapters_map.get(target_backend_name) # 4. 协议转换并转发请求 try: backend_specific_payload = await target_adapter.transform_request(request) async with target_adapter.client as client: if request.stream: # 处理流式响应 async def response_stream_generator(): backend_resp_stream = client.stream( "POST", target_adapter.config.api_path, json=backend_specific_payload, headers={"Accept": "text/event-stream"} # 根据后端要求调整 ) async with backend_resp_stream as response: async for chunk in response.aiter_bytes(): # 这里需要根据后端流式协议进行可能的格式转换 # 简单起见，假设后端流式格式与OpenAI兼容 yield chunk return StreamingResponse( response_stream_generator(), media_type="text/event-stream" ) else: # 处理非流式响应 backend_resp = await client.post( target_adapter.config.api_path, json=backend_specific_payload, timeout=target_adapter.config.timeout ) backend_resp.raise_for_status() # 5. 转换响应格式并返回 unified_response = target_adapter.transform_response(backend_resp) return unified_response except httpx.TimeoutException: # 超时，尝试故障转移到降级后端 if router_config.fallback_backend and router_config.fallback_backend != target_backend_name: # 记录日志，重试到降级后端（代码略） pass raise HTTPException(status_code=504, detail="Backend service timeout.") except httpx.HTTPStatusError as e: # 将后端错误码和消息适当转换后返回 raise HTTPException(status_code=e.response.status_code, detail=f"Backend error: {e.response.text}") except Exception as e: # 记录详细日志 raise HTTPException(status_code=500, detail=f"Internal router error: {str(e)}") @app.get("/health") async def health_check(): """Router自身的健康检查端点，可聚合后端健康状态""" overall_healthy = True details = {} for name, adapter in adapters_map.items(): is_healthy = await adapter.health_check() details[name] = "healthy" if is_healthy else "unhealthy" if not is_healthy: overall_healthy = False status_code = 200 if overall_healthy else 503 return {"status": "healthy" if overall_healthy else "degraded", "details": details}, status_code

至此，一个具备基本路由、协议转换、灰度发布和容错能力的混合Agent集群Router就搭建完成了。你可以通过uvicorn main:app --reload命令启动它。

4. 生产环境进阶：稳定性、可观测性与动态配置

一个能在实验室跑通的Demo和一個能扛住生产流量的服务之间，隔着十万八千里。下面分享几个我们在实际部署中踩过的坑和积累的经验。

4.1 稳定性保障：超时、重试与熔断

直接使用httpx的简单客户端是不够的。生产环境必须考虑网络抖动、后端瞬时压力等问题。

• 连接池与超时设置：为每个后端的httpx.AsyncClient配置合理的连接池限制(limits)和超时时间。全局超时和读写超时要分开设置。

  client = httpx.AsyncClient( base_url=config.api_base, timeout=httpx.Timeout(connect=5.0, read=60.0, write=60.0, pool=5.0), limits=httpx.Limits(max_keepalive_connections=10, max_connections=100), transport=httpx.AsyncHTTPTransport(retries=2) # 内置简单重试 )

• 智能重试策略：不要对所有错误都重试。5xx错误（服务器内部错误）和网络超时可以重试，4xx错误（客户端错误）重试通常无效。可以使用tenacity库实现更灵活的重试策略（如指数退避）。
• 熔断器模式：当某个后端连续失败多次后，应快速将其“熔断”，在一段时间内直接拒绝发往该后端的请求，给后端恢复的时间。可以使用circuitbreaker库。在Adapter的调用处包裹熔断器逻辑。

  from circuitbreaker import circuit class OpenClawAdapter(BaseAdapter): @circuit(failure_threshold=5, recovery_timeout=60) async def _call_backend(self, payload): # ... 实际调用代码

• 优雅降级：当主后端（如OpenClaw）熔断或不可用时，Router应能自动、无缝地将流量切换到降级后端（如Hermes，或一个功能简化的保底服务）。这需要在路由决策和请求转发逻辑中显式处理。

4.2 可观测性：监控、日志与链路追踪

“看不见”的系统是危险的。你必须知道流量去哪了，性能如何，哪里出错了。

• 结构化日志：使用structlog或json-logging记录每一条请求的完整上下文：request_id、user_id、target_backend、request_duration、http_status、error_message。这便于后续用ELK或Loki进行聚合分析。
• 关键指标暴露：使用prometheus_client在Router中暴露Metrics。
  • 请求量：router_requests_total{backend, status_code}
  • 延迟分布：router_request_duration_seconds_bucket{backend, le}
  • 后端健康状态：router_backend_up{backend}
  • 路由决策：router_routing_decisions_total{strategy, backend}这些指标可以通过Prometheus采集，并在Grafana中绘制成Dashboard，实时监控流量分布、响应时间和错误率。
• 分布式链路追踪：集成OpenTelemetry。为每个进入Router的请求生成一个唯一的trace_id，并透传给后端Agent。这样，你可以在Jaeger或Zipkin中看到一个用户请求从进入Router，到被转发至OpenClaw/Hermes，再到最终响应的完整调用链，对于排查复杂问题至关重要。

4.3 动态化与配置管理

将后端地址、路由规则、灰度比例等硬编码在配置文件里，每次变更都需要重启服务，这在生产环境是不可接受的。

• 后端服务发现：将Router与你的服务注册中心（如Consul、Nacos、K8s Service）集成。AgentBackendConfig中的api_base可以是一个服务名，Router启动时或定期从注册中心拉取健康的实例地址列表，并更新到客户端的负载均衡器中。这实现了后端实例的动态扩缩容和故障实例的自动剔除。
• 动态配置中心：将routing_strategy、rule_based_config、canary_config等路由规则存储在Apollo、Nacos（配置管理功能）或etcd中。Router监听配置变更，并在内存中热更新路由引擎的配置，实现秒级的规则生效，无需重启。
• 配置版本化与回滚：任何路由规则的变更都应该有版本记录和快速回滚的能力。在配置中心管理时，这通常是内置功能。

4.4 安全与治理

• 认证与鉴权：Router可以作为统一的认证关口。在请求转发前，验证API Key、JWT Token等。可以将鉴权逻辑抽象为中间件，避免每个后端重复实现。
• 限流与配额：在Router层实施全局限流（如使用slowapi），防止恶意流量打垮后端Agent。也可以根据用户或租户实施细粒度的配额管理。
• 请求/响应改写：你可以在适配器中加入全局性的请求/响应改写逻辑。例如，为所有发出的请求添加一个x-request-source: router的头；或者，在返回的响应中统一添加一些监控信息。

5. 踩坑实录与性能调优

理论很美好，现实很骨感。下面是我们上线前后遇到的一些典型问题。

5.1 流式响应（Server-Sent Events）的兼容性陷阱

这是我们遇到的第一个大坑。OpenClaw和Hermes的流式输出协议可能不同。OpenClaw可能用自定义的data: {...}格式，而Hermes完全遵循OpenAI的流式格式。我们的Router如果只是简单透传字节流，客户端可能会因为格式不统一而解析失败。

解决方案：在适配器的流式处理部分，不能简单yield chunk。需要根据后端类型，对数据块进行实时解析和格式转换。我们为每个适配器实现了transform_stream_chunk方法，将后端的数据块转换为标准的OpenAI流式数据块格式（data: {"id":"...","object":"...","created":...,"model":"...","choices":[...]}\n\n），然后再yield出去。这保证了无论后端是谁，客户端收到的流式数据格式都是统一的。

5.2 长上下文请求的超时与内存压力

当用户发送一个包含数万tokens的长文档进行总结时，请求处理时间可能超过30秒。简单的固定超时设置会导致大量超时错误。

解决方案：

1. 动态超时：根据请求的messages长度或预估的token数，动态调整转发给后端的超时时间。可以设置一个基线（如60秒），并随token数增加而延长。
2. 异步任务与轮询：对于极长耗时的任务，不适合用同步HTTP请求。可以修改协议，让Router将任务提交到后端后立即返回一个task_id，并提供另一个GET /tasks/{task_id}的端点供客户端轮询结果。这需要后端Agent也支持异步任务接口。
3. 内存监控：Router本身在转换大请求/响应时也可能消耗大量内存。需要使用tracemalloc等工具监控内存使用，并设置合理的请求体大小限制（client = httpx.AsyncClient(..., limits=httpx.Limits(max_connections=100, max_keepalive_connections=20, max_content_width=10_000_000))）。

5.3 会话（Session）状态管理的挑战

有些Agent是有状态的，一次对话的后续请求依赖于之前的上下文（保存在Agent的会话内存中）。当Router将同一个用户的多次请求负载均衡到不同后端实例时，状态就丢失了。

解决方案：

1. 会话粘滞（Session Affinity）：在路由决策时，对于同一会话（可通过客户端传入的session_id或根据user_id哈希），总是路由到同一个后端实例。这可以通过在路由引擎中维护一个简单的session_id -> backend映射来实现（注意分布式环境下的共享问题），或者使用一致性哈希算法。
2. 外部化状态：更优雅但更复杂的方式是要求所有Agent将会话状态存储到外部数据库（如Redis）中。这样Router就可以无状态地路由，任何实例都能处理任何请求。但这需要对Agent本身进行改造，违反了“不迁移”的原则，可作为远期架构目标。

5.4 性能基准测试与优化

在流量上来之前，我们做了一次压力测试。发现当并发数超过500时，Router的响应时间急剧上升。

排查过程：

1. 使用py-spy进行性能剖析：发现大量时间花在json.dumps和json.loads上，尤其是在协议转换层。
2. 优化：对于已知固定的转换逻辑（如OpenClaw适配器），我们不再使用通用的字典操作，而是预先构建好模板，只替换变量部分，减少了序列化/反序列化的开销。对于大型响应，考虑使用orjson替代标准库json，它有显著的性能提升。
3. 连接池调优：发现默认的连接池大小成为瓶颈。根据压测结果，我们调整了每个后端AsyncClient的max_keepalive_connections和max_connections参数，使其与后端服务的实际承受能力匹配。
4. 异步代码审查：检查了整个调用链，确保没有在异步函数中调用阻塞的IO操作（如错误的文件读写、同步的Redis调用）。将所有阻塞调用改为异步版本。

经过这些优化，Router的99分位响应时间（P99）在1000并发下保持了稳定。

构建这样一个混合Agent集群的Router，更像是在铺设一条智能的“高速公路系统”，而不是造一辆更快的车。它的价值不在于自身有多强的AI能力，而在于其连接、调度与治理的能力。通过将OpenClaw、Hermes乃至未来的新Agent无缝“编成一队”，我们实现了技术栈的自主选择、系统的平滑演进和业务风险的精细控制。这个过程中积累的关于协议转换、流量治理、可观测性的经验，其价值甚至超过了项目本身。如果你也在面临多套AI系统共存的烦恼，不妨从设计一个简单的Router开始，它会为你打开一扇通往更灵活、更健壮的AI基础设施的大门。