最近在尝试将大语言模型（LLMs）应用到一些对延迟和性能有严苛要求的业务场景时，比如实时对话、智能客服或代码补全，我们常常会遇到一个核心矛盾：单个模型的能力上限与响应速度难以兼得。一个强大的模型可能推理缓慢，而一个轻量模型又可能无法满足复杂任务的需求。这时，一个自然而然的思路是：能否让多个模型协同工作，取长补短？这正是多智能体服务（Multi-Agent Serving）要解决的问题，而近期社区热议的“Chimera”等概念，更是将“异构LLMs的低延迟、高性能协同服务”推向了前台。本文将围绕如何构建一个面向异构大语言模型的多智能体服务系统展开，从核心概念、架构设计到实战部署，为你提供一套完整的解决方案。

本文适合有一定大模型应用基础的开发者，无论是希望优化现有服务的性能，还是探索模型协同的新范式，都能从中获得可直接复用的代码、配置与避坑指南。我们将从零开始，搭建一个简易但功能完整的多智能体服务框架，并深入探讨其背后的调度策略与性能优化。

1. 背景与核心概念：为什么需要多智能体服务？

在深入代码之前，我们首先要厘清几个关键概念：LLMs、智能体（Agent）以及多智能体服务（Multi-Agent Serving）。

大语言模型（LLMs）是我们熟知的基础，如 GPT、LLaMA、ChatGLM 等，它们能够理解和生成人类语言。然而，不同的模型在规模、能力、推理速度和资源消耗上差异巨大。

智能体（Agent）在此语境下，可以简单理解为一个封装了特定LLM实例的服务单元。它不仅仅是一个模型调用接口，更包含了与模型交互的上下文管理、工具调用（Tool Calling）、思维链（Chain-of-Thought）等逻辑。一个智能体代表了一种特定的问题解决能力。

多智能体服务（Multi-Agent Serving）则是一个更高层的协调系统。它的核心目标是高效、智能地调度和管理多个异构的智能体（背后是异构的LLMs），共同完成用户请求。这里的“异构”体现在多个维度：

• 模型异构：使用不同架构、不同大小的模型（如一个70B的“专家”模型和一個7B的“快车”模型）。
• 能力异构：不同智能体擅长不同任务（如一个负责代码生成，一个负责文本摘要，一个负责逻辑推理）。
• 位置异构：智能体可能部署在不同的机器、不同的区域，甚至混合了云端和本地部署。

那么，为什么我们需要这样一个系统？它解决了什么痛点？

1. 性能与成本的平衡：将简单、高频的请求路由到轻量、快速的模型（降低成本，提高吞吐），将复杂、低频的请求路由到强大但缓慢的模型（保证质量）。
2. 提高系统可靠性与弹性：单个模型服务可能宕机或过载。多智能体系统可以通过健康检查和负载均衡，将请求故障转移到其他可用智能体，实现服务降级。
3. 实现能力组合与增强：一个任务可能需要多个步骤，每个步骤由最擅长的智能体处理。例如，用户提问“分析这段代码并给出优化建议”，系统可以先后调用“代码理解智能体”和“代码优化智能体”。
4. 降低延迟（Latency）：这是“Chimera”等前沿研究关注的核心。通过对请求进行预测性分析（如复杂度评估），并在多个智能体间进行智能调度，甚至让它们并行处理子任务，可以显著降低端到端响应时间，实现**延迟与性能感知（Latency- and Performance-Aware）**的服务。

接下来，我们将动手构建一个具备基础能力的多智能体服务系统。

2. 环境准备与版本说明

我们的演示系统将使用 Python 作为主要开发语言，并利用一些成熟的网络框架和模型客户端库。请确保你的环境满足以下要求：

• 操作系统：Linux (Ubuntu 20.04/22.04) 或 macOS，Windows 建议使用 WSL2。
• Python 版本：>= 3.8， 推荐 3.9 或 3.10。
• 核心依赖：
  • fastapi：用于构建高性能的 API 服务。
  • uvicorn：ASGI 服务器，用于运行 FastAPI。
  • pydantic：用于数据验证和设置管理。
  • httpx：异步 HTTP 客户端，用于向不同的模型后端发送请求。
  • redis：可选，用于实现请求队列、缓存或分布式锁，在进阶架构中很有用。
• 模型后端：为了模拟异构环境，我们需要至少两个不同的模型服务端点。你可以使用：
  • 本地部署：Ollama (运行 LLaMA2、CodeLlama 等)、vLLM、Xinference。
  • 云服务 API：OpenAI API、 Anthropic Claude API、 国内各大平台的 API。
  • 我们将使用Ollama运行两个不同大小的模型来模拟，因为它易于本地设置。

版本说明： 本文示例代码基于以下常见版本，但重点在于展示架构和思路，实际版本请根据你的项目需求调整。

fastapi==0.104.1 uvicorn[standard]==0.24.0 pydantic==2.5.0 httpx==0.25.1 redis==5.0.1

项目结构预览： 在开始前，我们先规划一下项目目录。

multi_agent_serving/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口 │ ├── config.py # 配置文件 │ ├── agents/ # 智能体模块 │ │ ├── __init__.py │ │ ├── base.py # 智能体基类 │ │ ├── fast_agent.py # 轻量快速智能体 │ │ ├── expert_agent.py # 重量专家智能体 │ │ └── router.py # 路由决策器 │ ├── models/ # 数据模型 │ │ ├── __init__.py │ │ └── schemas.py # Pydantic 模型定义 │ └── utils/ # 工具函数 │ ├── __init__.py │ └── health_check.py ├── requirements.txt └── README.md

3. 核心架构与组件拆解

一个典型的多智能体服务系统包含以下几个核心组件：

1. API 网关（API Gateway）：接收所有客户端请求，是系统的唯一入口。
2. 路由决策器（Router / Orchestrator）：核心大脑。分析请求内容，根据预定义的策略（如基于内容、基于性能、基于负载）决定将请求分发给哪个或哪几个智能体。
3. 智能体池（Agent Pool）：管理多个智能体实例。负责智能体的注册、发现、健康状态监控和负载统计。
4. 智能体（Agent）：实际执行任务的工作单元。它封装了与特定 LLM 后端的通信逻辑、上下文管理和工具调用。
5. 结果聚合器（Aggregator）：如果一个请求被拆分并由多个智能体并行或串行处理，则需要此组件来合并最终结果。

我们的实战将重点实现前四个组件。

3.1 智能体基类设计

所有智能体都应遵循统一的接口，便于路由器和池化管理。我们在app/agents/base.py中定义基类。

# app/agents/base.py from abc import ABC, abstractmethod from typing import Any, Dict, List, Optional from app.models.schemas import AgentRequest, AgentResponse class BaseAgent(ABC): """智能体抽象基类""" def __init__(self, agent_id: str, name: str, description: str, endpoint: str): self.agent_id = agent_id self.name = name self.description = description self.endpoint = endpoint # 模型后端的API地址 self.is_healthy = True self.current_load = 0 # 当前负载，可简单用正在处理的请求数表示 self.avg_latency = 0.0 # 平均响应延迟（毫秒） @abstractmethod async def invoke(self, request: AgentRequest) -> AgentResponse: """ 调用智能体的核心方法。 参数: AgentRequest 对象，包含用户输入、会话历史等。 返回: AgentResponse 对象，包含模型输出、元数据等。 """ pass async def health_check(self) -> bool: """检查智能体后端服务是否健康""" # 简化实现：发送一个简单的 ping 请求 import httpx try: async with httpx.AsyncClient(timeout=5.0) as client: # 这里假设后端有一个 /health 端点 resp = await client.get(f"{self.endpoint}/health") self.is_healthy = resp.status_code == 200 return self.is_healthy except Exception: self.is_healthy = False return False def get_metadata(self) -> Dict[str, Any]: """获取智能体元数据，用于路由决策""" return { "agent_id": self.agent_id, "name": self.name, "healthy": self.is_healthy, "current_load": self.current_load, "avg_latency": self.avg_latency, "description": self.description, }

3.2 数据模型定义

在app/models/schemas.py中定义请求和响应的数据结构。

# app/models/schemas.py from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any class Message(BaseModel): role: str # "user", "assistant", "system" content: str class AgentRequest(BaseModel): """发送给单个智能体的请求""" messages: List[Message] stream: bool = False # 是否流式输出 max_tokens: Optional[int] = None temperature: Optional[float] = 0.7 # 可以添加其他模型特定参数 extra_params: Dict[str, Any] = Field(default_factory=dict) class AgentResponse(BaseModel): """从单个智能体返回的响应""" content: str agent_id: str model_used: str latency_ms: float # 本次调用的延迟 finish_reason: Optional[str] = None token_usage: Optional[Dict[str, int]] = None # 输入/输出token数 class UserRequest(BaseModel): """用户发送到网关的原始请求""" query: str session_id: Optional[str] = None # 用于会话保持 # 可选的用户提示，用于影响路由，如“需要详细解答”、“请快速回答” user_hint: Optional[str] = None class OrchestratorResponse(BaseModel): """网关返回给用户的最终响应""" response: str chosen_agent: str all_metadata: List[Dict[str, Any]] # 所有参与智能体的元数据（用于调试） total_latency_ms: float

4. 完整实战案例：构建多智能体服务

现在，我们开始搭建完整的服务。假设我们有两个本地运行的 Ollama 模型：

• Fast Agent: 使用llama2:7b模型，部署在http://localhost:11434，特点是响应快。
• Expert Agent: 使用llama2:13b模型，部署在http://localhost:11435，特点是能力更强但稍慢。

4.1 创建项目结构与依赖

首先，创建项目目录并安装依赖。

mkdir multi_agent_serving && cd multi_agent_serving python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate

创建requirements.txt文件：

fastapi==0.104.1 uvicorn[standard]==0.24.0 pydantic==2.5.0 httpx==0.25.1

安装依赖：

pip install -r requirements.txt

按照之前规划的目录结构创建文件和文件夹。

4.2 实现具体的智能体

Fast Agent (app/agents/fast_agent.py):

# app/agents/fast_agent.py import httpx import time from typing import Dict, Any from app.agents.base import BaseAgent from app.models.schemas import AgentRequest, AgentResponse class FastAgent(BaseAgent): """轻量快速智能体，对接小型/快速模型""" MODEL_NAME = "llama2:7b" def __init__(self, agent_id: str = "fast_01"): super().__init__( agent_id=agent_id, name="Fast-7B-Agent", description="快速响应智能体，适用于简单问答和对话。", endpoint="http://localhost:11434" # Ollama 默认端口 ) async def invoke(self, request: AgentRequest) -> AgentResponse: start_time = time.perf_counter() self.current_load += 1 try: # 构造 Ollama 兼容的请求体 ollama_payload = { "model": self.MODEL_NAME, "messages": [msg.dict() for msg in request.messages], "stream": request.stream, "options": { "temperature": request.temperature, "num_predict": request.max_tokens, } } # 移除 None 值 ollama_payload["options"] = {k: v for k, v in ollama_payload["options"].items() if v is not None} async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.post( f"{self.endpoint}/api/chat", json=ollama_payload, headers={"Content-Type": "application/json"} ) resp.raise_for_status() result = resp.json() content = result["message"]["content"] latency_ms = (time.perf_counter() - start_time) * 1000 # 简单更新平均延迟（可优化为滑动平均） self.avg_latency = (self.avg_latency + latency_ms) / 2 if self.avg_latency else latency_ms return AgentResponse( content=content, agent_id=self.agent_id, model_used=self.MODEL_NAME, latency_ms=latency_ms, finish_reason=result.get("done_reason"), token_usage={"prompt_tokens": -1, "completion_tokens": -1} # Ollama 默认不返回，需配置 ) except Exception as e: # 记录错误，更新健康状态 self.is_healthy = False raise e finally: self.current_load -= 1

Expert Agent (app/agents/expert_agent.py): 其结构与 FastAgent 类似，主要区别在于MODEL_NAME和endpoint。我们可以通过继承来减少重复代码，但为了清晰，这里展示独立实现。

# app/agents/expert_agent.py import httpx import time from app.agents.base import BaseAgent from app.models.schemas import AgentRequest, AgentResponse class ExpertAgent(BaseAgent): """专家智能体，对接更大更强的模型""" MODEL_NAME = "llama2:13b" def __init__(self, agent_id: str = "expert_01"): super().__init__( agent_id=agent_id, name="Expert-13B-Agent", description="高精度专家智能体，适用于复杂推理和创作。", endpoint="http://localhost:11435" # 假设第二个 Ollama 实例运行在不同端口 ) async def invoke(self, request: AgentRequest) -> AgentResponse: start_time = time.perf_counter() self.current_load += 1 try: ollama_payload = { "model": self.MODEL_NAME, "messages": [msg.dict() for msg in request.messages], "stream": request.stream, "options": {"temperature": request.temperature} } if request.max_tokens: ollama_payload["options"]["num_predict"] = request.max_tokens async with httpx.AsyncClient(timeout=60.0) as client: # 专家模型超时设长 resp = await client.post( f"{self.endpoint}/api/chat", json=ollama_payload ) resp.raise_for_status() result = resp.json() latency_ms = (time.perf_counter() - start_time) * 1000 self.avg_latency = (self.avg_latency + latency_ms) / 2 if self.avg_latency else latency_ms return AgentResponse( content=result["message"]["content"], agent_id=self.agent_id, model_used=self.MODEL_NAME, latency_ms=latency_ms, finish_reason=result.get("done_reason") ) except Exception as e: self.is_healthy = False raise e finally: self.current_load -= 1

4.3 实现路由决策器

路由决策是系统的核心。我们实现一个简单的基于规则的路由器 (app/agents/router.py)。更复杂的系统可以使用机器学习模型来预测。

# app/agents/router.py from typing import List, Optional from app.agents.base import BaseAgent from app.models.schemas import UserRequest, AgentRequest, Message class SimpleRouter: """简单的基于规则和负载的路由器""" def __init__(self, agents: List[BaseAgent]): self.agents = {agent.agent_id: agent for agent in agents} async def select_agent(self, user_request: UserRequest) -> Optional[BaseAgent]: """ 根据请求内容选择最合适的智能体。 策略： 1. 只选择健康的智能体。 2. 如果用户提示包含‘快速’、‘简单’，优先选 FastAgent。 3. 如果用户提示包含‘详细’、‘深入’、‘复杂’，优先选 ExpertAgent。 4. 否则，选择当前负载最低的智能体。 """ healthy_agents = [agent for agent in self.agents.values() if agent.is_healthy] if not healthy_agents: return None hint = (user_request.user_hint or "").lower() query = user_request.query.lower() # 规则匹配 fast_keywords = ["快速", "简单", "快", "brief", "quick", "simple"] expert_keywords = ["详细", "深入", "复杂", "分析", "解释", "detailed", "complex", "analyze"] candidate_agents = [] for agent in healthy_agents: if "fast" in agent.name.lower() and any(kw in hint or kw in query for kw in fast_keywords): candidate_agents.append(agent) elif "expert" in agent.name.lower() and any(kw in hint or kw in query for kw in expert_keywords): candidate_agents.append(agent) # 如果规则匹配到了，从匹配的里面选负载最低的 if candidate_agents: return min(candidate_agents, key=lambda a: a.current_load) # 否则，从所有健康智能体中选负载最低的（默认负载均衡） else: return min(healthy_agents, key=lambda a: a.current_load) def convert_to_agent_request(self, user_request: UserRequest) -> AgentRequest: """将用户请求转换为智能体请求""" messages = [Message(role="user", content=user_request.query)] return AgentRequest(messages=messages)

4.4 构建 FastAPI 主应用与智能体池

现在，我们在app/main.py中整合所有组件。

# app/main.py from fastapi import FastAPI, HTTPException from contextlib import asynccontextmanager import asyncio from app.models.schemas import UserRequest, OrchestratorResponse from app.agents.fast_agent import FastAgent from app.agents.expert_agent import ExpertAgent from app.agents.router import SimpleRouter # 全局变量存储智能体和路由器 agents = [] router = None @asynccontextmanager async def lifespan(app: FastAPI): """生命周期管理：启动时初始化，关闭时清理""" # 启动 global agents, router print("Initializing agents...") fast_agent = FastAgent() expert_agent = ExpertAgent() agents = [fast_agent, expert_agent] router = SimpleRouter(agents) # 启动后台健康检查任务 asyncio.create_task(periodic_health_check()) yield # 关闭 print("Shutting down...") # 可以在这里添加清理逻辑 async def periodic_health_check(interval: int = 30): """定期检查所有智能体的健康状态""" while True: for agent in agents: await agent.health_check() await asyncio.sleep(interval) app = FastAPI(title="Heterogeneous LLMs Multi-Agent Serving", lifespan=lifespan) @app.get("/") async def root(): return {"message": "Heterogeneous LLMs Multi-Agent Serving System is running."} @app.get("/agents/status") async def get_agents_status(): """获取所有智能体的状态（用于监控）""" return {agent.agent_id: agent.get_metadata() for agent in agents} @app.post("/chat", response_model=OrchestratorResponse) async def chat(user_request: UserRequest): """ 主聊天接口。 1. 接收用户请求。 2. 路由器选择智能体。 3. 调用智能体。 4. 返回结果。 """ if not router: raise HTTPException(status_code=503, detail="Router not initialized") # 1. 选择智能体 selected_agent = await router.select_agent(user_request) if not selected_agent: raise HTTPException(status_code=503, detail="No healthy agent available") # 2. 转换请求格式 agent_request = router.convert_to_agent_request(user_request) # 3. 调用智能体 try: agent_response = await selected_agent.invoke(agent_request) except Exception as e: raise HTTPException(status_code=500, detail=f"Agent invocation failed: {str(e)}") # 4. 构造返回 all_metadata = [agent.get_metadata() for agent in agents] return OrchestratorResponse( response=agent_response.content, chosen_agent=selected_agent.agent_id, all_metadata=all_metadata, total_latency_ms=agent_response.latency_ms )

4.5 运行与验证

首先，确保你的 Ollama 服务已经启动并加载了相应模型。假设你已经在不同端口运行了两个实例（需要 Ollama 支持多实例，可通过环境变量OLLAMA_HOST或--port参数实现）。

然后，启动我们的多智能体服务：

cd multi_agent_serving uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

服务启动后，你可以通过以下方式进行测试：

1. 查看智能体状态：

curl http://localhost:8000/agents/status

2. 发送一个请求（使用 Fast Agent）：

curl -X POST http://localhost:8000/chat \ -H "Content-Type: application/json" \ -d '{ "query": "你好，请简单介绍一下Python。", "user_hint": "快速回答" }'

预期响应中，chosen_agent字段应为fast_01。

3. 发送一个复杂请求（使用 Expert Agent）：

curl -X POST http://localhost:8000/chat \ -H "Content-Type: application/json" \ -d '{ "query": "请详细解释Transformer架构中的注意力机制，并给出数学公式。", "user_hint": "需要深入分析" }'

预期响应中，chosen_agent字段应为expert_01。

4. 不提供提示，由系统负载均衡：

curl -X POST http://localhost:8000/chat \ -H "Content-Type: application/json" \ -d '{ "query": "今天的天气怎么样？" }'

系统将选择当前负载最低的健康智能体。

5. 常见问题与排查思路

在搭建和运行多智能体服务时，你可能会遇到以下典型问题：

6. 最佳实践与工程建议

将多智能体服务投入生产环境，需要考虑更多工程化细节。

1. 配置化管理将所有智能体的端点、模型名称、超时时间、路由规则等抽取到配置文件（如 YAML）或配置中心（如 Apollo）。避免硬编码。

# config/agents.yaml agents: - id: fast_01 name: Fast-7B-Agent type: fast endpoint: ${FAST_MODEL_ENDPOINT:http://localhost:11434} model: llama2:7b health_check_path: /api/tags timeout: 30 - id: expert_01 name: Expert-13B-Agent type: expert endpoint: ${EXPERT_MODEL_ENDPOINT:http://localhost:11435} model: llama2:13b health_check_path: /api/tags timeout: 60

2. 智能体注册与发现在微服务架构中，智能体可能动态扩缩容。需要实现一个注册中心（如使用 Redis、Etcd 或 Consul）。智能体启动时向注册中心注册自身信息（端点、能力、负载），路由器从注册中心拉取可用智能体列表。

3. 高级路由策略

• 基于预测的路由：训练一个轻量级分类器，根据请求的嵌入向量或特征，预测其复杂度和所需模型类型。
• 基于代价的路由：综合考虑延迟、成本（token 费用）、准确率，做出最优决策。
• 组合路由（Chimera 思想）：将复杂请求拆分为子任务，让 Fast Agent 处理简单部分，Expert Agent 处理困难部分，并行执行后聚合结果，这是降低整体延迟的关键。

4. 监控与可观测性

• 指标收集：记录每个请求的路径、所选智能体、响应延迟、token 使用量、错误率。集成 Prometheus 和 Grafana。
• 链路追踪：使用 OpenTelemetry 对请求在多智能体间的流转进行追踪，便于排查性能瓶颈。
• 日志聚合：结构化日志，集中收集到 ELK 或 Loki 中。

5. 容错与降级

• 重试机制：对智能体调用失败进行有限次数的重试，可更换智能体重试。
• 熔断器：对频繁失败的智能体实施熔断，暂时将其从可用列表中剔除，定期探测恢复。
• 服务降级：当所有 Expert Agent 都不可用时，系统应能自动将所有请求降级到 Fast Agent，保证基本服务可用。

6. 安全与权限

• API 认证：为网关接口添加 API Key 或 JWT 认证。
• 智能体间认证：确保只有网关可以调用智能体服务。
• 输入输出过滤：防止 Prompt 注入攻击，对用户输入和模型输出进行必要的清洗和过滤。

7. 性能优化

• 连接池：使用httpx.AsyncClient的连接池功能，避免为每个请求创建新连接。
• 请求批处理：如果多个请求适合同一个模型，可以考虑在智能体端进行批处理以提高吞吐。
• 缓存：对常见、确定的查询结果进行缓存（如使用 Redis），直接返回，绕过模型推理。

从简单的规则路由到复杂的性能感知调度，多智能体服务为异构 LLMs 的协同应用提供了强大的框架。本文提供的实战案例是一个起点，你可以在此基础上，根据具体的业务需求、性能指标和成本约束，不断迭代和优化你的智能体生态系统，最终构建出稳定、高效、智能的下一代大模型应用服务。