拓冰网站优化 
最近在尝试将大语言模型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的“快车”模型。能力异构不同智能体擅长不同任务如一个负责代码生成一个负责文本摘要一个负责逻辑推理。位置异构智能体可能部署在不同的机器、不同的区域甚至混合了云端和本地部署。那么为什么我们需要这样一个系统它解决了什么痛点性能与成本的平衡将简单、高频的请求路由到轻量、快速的模型降低成本提高吞吐将复杂、低频的请求路由到强大但缓慢的模型保证质量。提高系统可靠性与弹性单个模型服务可能宕机或过载。多智能体系统可以通过健康检查和负载均衡将请求故障转移到其他可用智能体实现服务降级。实现能力组合与增强一个任务可能需要多个步骤每个步骤由最擅长的智能体处理。例如用户提问“分析这段代码并给出优化建议”系统可以先后调用“代码理解智能体”和“代码优化智能体”。降低延迟Latency这是“Chimera”等前沿研究关注的核心。通过对请求进行预测性分析如复杂度评估并在多个智能体间进行智能调度甚至让它们并行处理子任务可以显著降低端到端响应时间实现**延迟与性能感知Latency- and Performance-Aware**的服务。接下来我们将动手构建一个具备基础能力的多智能体服务系统。2. 环境准备与版本说明我们的演示系统将使用 Python 作为主要开发语言并利用一些成熟的网络框架和模型客户端库。请确保你的环境满足以下要求操作系统Linux (Ubuntu 20.04/22.04) 或 macOSWindows 建议使用 WSL2。Python 版本 3.8 推荐 3.9 或 3.10。核心依赖fastapi用于构建高性能的 API 服务。uvicornASGI 服务器用于运行 FastAPI。pydantic用于数据验证和设置管理。httpx异步 HTTP 客户端用于向不同的模型后端发送请求。redis可选用于实现请求队列、缓存或分布式锁在进阶架构中很有用。模型后端为了模拟异构环境我们需要至少两个不同的模型服务端点。你可以使用本地部署Ollama (运行 LLaMA2、CodeLlama 等)、vLLM、Xinference。云服务 APIOpenAI API、 Anthropic Claude API、 国内各大平台的 API。我们将使用Ollama运行两个不同大小的模型来模拟因为它易于本地设置。版本说明 本文示例代码基于以下常见版本但重点在于展示架构和思路实际版本请根据你的项目需求调整。fastapi0.104.1 uvicorn[standard]0.24.0 pydantic2.5.0 httpx0.25.1 redis5.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.md3. 核心架构与组件拆解一个典型的多智能体服务系统包含以下几个核心组件API 网关API Gateway接收所有客户端请求是系统的唯一入口。路由决策器Router / Orchestrator核心大脑。分析请求内容根据预定义的策略如基于内容、基于性能、基于负载决定将请求分发给哪个或哪几个智能体。智能体池Agent Pool管理多个智能体实例。负责智能体的注册、发现、健康状态监控和负载统计。智能体Agent实际执行任务的工作单元。它封装了与特定 LLM 后端的通信逻辑、上下文管理和工具调用。结果聚合器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(timeout5.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_factorydict) 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: float4. 完整实战案例构建多智能体服务现在我们开始搭建完整的服务。假设我们有两个本地运行的 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文件fastapi0.104.1 uvicorn[standard]0.24.0 pydantic2.5.0 httpx0.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_idagent_id, nameFast-7B-Agent, description快速响应智能体适用于简单问答和对话。, endpointhttp://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(timeout30.0) as client: resp await client.post( f{self.endpoint}/api/chat, jsonollama_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( contentcontent, agent_idself.agent_id, model_usedself.MODEL_NAME, latency_mslatency_ms, finish_reasonresult.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 - 1Expert 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_idagent_id, nameExpert-13B-Agent, description高精度专家智能体适用于复杂推理和创作。, endpointhttp://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(timeout60.0) as client: # 专家模型超时设长 resp await client.post( f{self.endpoint}/api/chat, jsonollama_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( contentresult[message][content], agent_idself.agent_id, model_usedself.MODEL_NAME, latency_mslatency_ms, finish_reasonresult.get(done_reason) ) except Exception as e: self.is_healthy False raise e finally: self.current_load - 14.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, keylambda a: a.current_load) # 否则从所有健康智能体中选负载最低的默认负载均衡 else: return min(healthy_agents, keylambda a: a.current_load) def convert_to_agent_request(self, user_request: UserRequest) - AgentRequest: 将用户请求转换为智能体请求 messages [Message(roleuser, contentuser_request.query)] return AgentRequest(messagesmessages)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(titleHeterogeneous LLMs Multi-Agent Serving, lifespanlifespan) 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_modelOrchestratorResponse) async def chat(user_request: UserRequest): 主聊天接口。 1. 接收用户请求。 2. 路由器选择智能体。 3. 调用智能体。 4. 返回结果。 if not router: raise HTTPException(status_code503, detailRouter not initialized) # 1. 选择智能体 selected_agent await router.select_agent(user_request) if not selected_agent: raise HTTPException(status_code503, detailNo 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_code500, detailfAgent invocation failed: {str(e)}) # 4. 构造返回 all_metadata [agent.get_metadata() for agent in agents] return OrchestratorResponse( responseagent_response.content, chosen_agentselected_agent.agent_id, all_metadataall_metadata, total_latency_msagent_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/status2. 发送一个请求使用 Fast Agentcurl -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d { query: 你好请简单介绍一下Python。, user_hint: 快速回答 }预期响应中chosen_agent字段应为fast_01。3. 发送一个复杂请求使用 Expert Agentcurl -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. 常见问题与排查思路在搭建和运行多智能体服务时你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案服务启动失败端口被占用端口 8000 或其他指定端口已被其他进程使用。1. 使用lsof -i :8000或netstat -tulnp | grep 8000查看占用进程。2. 终止占用进程或修改应用启动端口uvicorn ... --port 8001。调用/chat接口返回503: No healthy agent available所有智能体健康检查失败。1. 检查agents/status端点查看各个智能体的healthy状态。2. 确认 Ollama 或其他模型后端服务是否正在运行 (ps aux | grep ollama)。3. 检查网络连通性确保能从服务主机访问到模型后端的地址和端口 (curl http://localhost:11434/api/tags)。4. 检查模型名称是否正确是否已在后端加载。请求响应非常慢甚至超时。1. 模型后端本身推理慢。2. 网络延迟高。3. 智能体负载过高请求排队。1. 直接调用模型后端API测试其原生响应速度。2. 检查avg_latency和current_load监控指标。3. 考虑增加智能体实例水平扩展或实现更细粒度的请求队列和超时控制。路由决策不符合预期如复杂问题仍被分给 Fast Agent。路由规则过于简单或关键词匹配不准确。1. 检查user_hint和query的提取与匹配逻辑。2. 考虑引入更复杂的路由策略如基于请求长度、嵌入向量相似度、或预测模型。流式响应streamTrue不工作。我们的示例代码未完整实现流式传输的透传。1. 需要在Agent.invoke方法中处理stream参数并将模型后端返回的流式数据块实时转发给客户端。2. FastAPI 需要使用StreamingResponse。这是一个进阶功能需要调整接口设计。在高并发下current_load计数不准确。current_load的增减不是原子操作存在并发问题。使用线程安全的计数器如asyncio.Lock或threading.Lock来保护current_load的修改或使用atomic操作。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: 602. 智能体注册与发现在微服务架构中智能体可能动态扩缩容。需要实现一个注册中心如使用 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 的协同应用提供了强大的框架。本文提供的实战案例是一个起点你可以在此基础上根据具体的业务需求、性能指标和成本约束不断迭代和优化你的智能体生态系统最终构建出稳定、高效、智能的下一代大模型应用服务。 构建异构大语言模型多智能体服务：从架构设计到实战部署
最近在尝试将大语言模型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的“快车”模型。能力异构不同智能体擅长不同任务如一个负责代码生成一个负责文本摘要一个负责逻辑推理。位置异构智能体可能部署在不同的机器、不同的区域甚至混合了云端和本地部署。那么为什么我们需要这样一个系统它解决了什么痛点性能与成本的平衡将简单、高频的请求路由到轻量、快速的模型降低成本提高吞吐将复杂、低频的请求路由到强大但缓慢的模型保证质量。提高系统可靠性与弹性单个模型服务可能宕机或过载。多智能体系统可以通过健康检查和负载均衡将请求故障转移到其他可用智能体实现服务降级。实现能力组合与增强一个任务可能需要多个步骤每个步骤由最擅长的智能体处理。例如用户提问“分析这段代码并给出优化建议”系统可以先后调用“代码理解智能体”和“代码优化智能体”。降低延迟Latency这是“Chimera”等前沿研究关注的核心。通过对请求进行预测性分析如复杂度评估并在多个智能体间进行智能调度甚至让它们并行处理子任务可以显著降低端到端响应时间实现**延迟与性能感知Latency- and Performance-Aware**的服务。接下来我们将动手构建一个具备基础能力的多智能体服务系统。2. 环境准备与版本说明我们的演示系统将使用 Python 作为主要开发语言并利用一些成熟的网络框架和模型客户端库。请确保你的环境满足以下要求操作系统Linux (Ubuntu 20.04/22.04) 或 macOSWindows 建议使用 WSL2。Python 版本 3.8 推荐 3.9 或 3.10。核心依赖fastapi用于构建高性能的 API 服务。uvicornASGI 服务器用于运行 FastAPI。pydantic用于数据验证和设置管理。httpx异步 HTTP 客户端用于向不同的模型后端发送请求。redis可选用于实现请求队列、缓存或分布式锁在进阶架构中很有用。模型后端为了模拟异构环境我们需要至少两个不同的模型服务端点。你可以使用本地部署Ollama (运行 LLaMA2、CodeLlama 等)、vLLM、Xinference。云服务 APIOpenAI API、 Anthropic Claude API、 国内各大平台的 API。我们将使用Ollama运行两个不同大小的模型来模拟因为它易于本地设置。版本说明 本文示例代码基于以下常见版本但重点在于展示架构和思路实际版本请根据你的项目需求调整。fastapi0.104.1 uvicorn[standard]0.24.0 pydantic2.5.0 httpx0.25.1 redis5.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.md3. 核心架构与组件拆解一个典型的多智能体服务系统包含以下几个核心组件API 网关API Gateway接收所有客户端请求是系统的唯一入口。路由决策器Router / Orchestrator核心大脑。分析请求内容根据预定义的策略如基于内容、基于性能、基于负载决定将请求分发给哪个或哪几个智能体。智能体池Agent Pool管理多个智能体实例。负责智能体的注册、发现、健康状态监控和负载统计。智能体Agent实际执行任务的工作单元。它封装了与特定 LLM 后端的通信逻辑、上下文管理和工具调用。结果聚合器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(timeout5.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_factorydict) 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: float4. 完整实战案例构建多智能体服务现在我们开始搭建完整的服务。假设我们有两个本地运行的 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文件fastapi0.104.1 uvicorn[standard]0.24.0 pydantic2.5.0 httpx0.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_idagent_id, nameFast-7B-Agent, description快速响应智能体适用于简单问答和对话。, endpointhttp://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(timeout30.0) as client: resp await client.post( f{self.endpoint}/api/chat, jsonollama_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( contentcontent, agent_idself.agent_id, model_usedself.MODEL_NAME, latency_mslatency_ms, finish_reasonresult.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 - 1Expert 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_idagent_id, nameExpert-13B-Agent, description高精度专家智能体适用于复杂推理和创作。, endpointhttp://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(timeout60.0) as client: # 专家模型超时设长 resp await client.post( f{self.endpoint}/api/chat, jsonollama_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( contentresult[message][content], agent_idself.agent_id, model_usedself.MODEL_NAME, latency_mslatency_ms, finish_reasonresult.get(done_reason) ) except Exception as e: self.is_healthy False raise e finally: self.current_load - 14.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, keylambda a: a.current_load) # 否则从所有健康智能体中选负载最低的默认负载均衡 else: return min(healthy_agents, keylambda a: a.current_load) def convert_to_agent_request(self, user_request: UserRequest) - AgentRequest: 将用户请求转换为智能体请求 messages [Message(roleuser, contentuser_request.query)] return AgentRequest(messagesmessages)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(titleHeterogeneous LLMs Multi-Agent Serving, lifespanlifespan) 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_modelOrchestratorResponse) async def chat(user_request: UserRequest): 主聊天接口。 1. 接收用户请求。 2. 路由器选择智能体。 3. 调用智能体。 4. 返回结果。 if not router: raise HTTPException(status_code503, detailRouter not initialized) # 1. 选择智能体 selected_agent await router.select_agent(user_request) if not selected_agent: raise HTTPException(status_code503, detailNo 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_code500, detailfAgent invocation failed: {str(e)}) # 4. 构造返回 all_metadata [agent.get_metadata() for agent in agents] return OrchestratorResponse( responseagent_response.content, chosen_agentselected_agent.agent_id, all_metadataall_metadata, total_latency_msagent_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/status2. 发送一个请求使用 Fast Agentcurl -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d { query: 你好请简单介绍一下Python。, user_hint: 快速回答 }预期响应中chosen_agent字段应为fast_01。3. 发送一个复杂请求使用 Expert Agentcurl -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. 常见问题与排查思路在搭建和运行多智能体服务时你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案服务启动失败端口被占用端口 8000 或其他指定端口已被其他进程使用。1. 使用lsof -i :8000或netstat -tulnp | grep 8000查看占用进程。2. 终止占用进程或修改应用启动端口uvicorn ... --port 8001。调用/chat接口返回503: No healthy agent available所有智能体健康检查失败。1. 检查agents/status端点查看各个智能体的healthy状态。2. 确认 Ollama 或其他模型后端服务是否正在运行 (ps aux | grep ollama)。3. 检查网络连通性确保能从服务主机访问到模型后端的地址和端口 (curl http://localhost:11434/api/tags)。4. 检查模型名称是否正确是否已在后端加载。请求响应非常慢甚至超时。1. 模型后端本身推理慢。2. 网络延迟高。3. 智能体负载过高请求排队。1. 直接调用模型后端API测试其原生响应速度。2. 检查avg_latency和current_load监控指标。3. 考虑增加智能体实例水平扩展或实现更细粒度的请求队列和超时控制。路由决策不符合预期如复杂问题仍被分给 Fast Agent。路由规则过于简单或关键词匹配不准确。1. 检查user_hint和query的提取与匹配逻辑。2. 考虑引入更复杂的路由策略如基于请求长度、嵌入向量相似度、或预测模型。流式响应streamTrue不工作。我们的示例代码未完整实现流式传输的透传。1. 需要在Agent.invoke方法中处理stream参数并将模型后端返回的流式数据块实时转发给客户端。2. FastAPI 需要使用StreamingResponse。这是一个进阶功能需要调整接口设计。在高并发下current_load计数不准确。current_load的增减不是原子操作存在并发问题。使用线程安全的计数器如asyncio.Lock或threading.Lock来保护current_load的修改或使用atomic操作。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: 602. 智能体注册与发现在微服务架构中智能体可能动态扩缩容。需要实现一个注册中心如使用 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 的协同应用提供了强大的框架。本文提供的实战案例是一个起点你可以在此基础上根据具体的业务需求、性能指标和成本约束不断迭代和优化你的智能体生态系统最终构建出稳定、高效、智能的下一代大模型应用服务。