架构设计

从单兵作战到群体智能 单个智能体的能力再强，也有其认知边界。当任务的复杂度超过单个智能体的处理能力时，多智能体协作就成为一种必然选择。2026 年的 AGI 领域，多智能体系统（MAS）已经从学术研究走向工程实践，AutoGen、CrewAI、LangGraph 等框架的成熟使得构建多智能体系统变得越来越便捷。 但"多个智能体一起工作"远比想象中复杂。如何分工？如何通信？如何处理分歧？如何避免死循环？这些问题的答案，构成了多智能体协作模式的核心设计空间。 一、协作模式分类框架 在深入具体模式之前，我们需要一个分类框架来理解多智能体协作的设计空间。我从两个维度来划分： 控制结构维度： 中心化（有明确的领导者） 去中心化（无明确领导者） 通信拓扑维度： 层级式（树状结构） 总线式（共享黑板） 对等式（网状结构） 环式（链式传递） 这两个维度的组合，产生了实践中最常见的几种协作模式。 二、层级式协作：指挥官与士兵 2.1 基本结构 层级式是最直观的多智能体协作模式。一个"编排者"（Orchestrator）智能体位于顶层，负责理解任务、分解子任务、分配给下游智能体，并汇总结果。 [编排者智能体] / | \ [研究智能体] [编码智能体] [测试智能体] 2.2 实现示例 以 AutoGen 为例，一个典型的层级式协作： import autogen # 配置模型 config_list = [{ "model": "gpt-4", "api_key": "your-api-key" }] # 创建编排者 orchestrator = autogen.AssistantAgent( name="编排者", system_message="""你是项目编排者。职责： 1. 理解用户需求 2. 将需求分解为子任务 3. 分配给合适的专家智能体 4. 汇总结果并确保质量 可用的专家： - 研究员：负责信息搜集和调研 - 工程师：负责代码编写 - 测试员：负责代码测试 请用以下格式分配任务： [TO: 专家名] 任务描述""", llm_config={"config_list": config_list} ) # 创建专家智能体 researcher = autogen.AssistantAgent( name="研究员", system_message="你是研究专家。接收编排者的任务，进行信息搜集和分析。", llm_config={"config_list": config_list} ) engineer = autogen.AssistantAgent( name="工程师", system_message="你是资深工程师。接收编排者的任务，编写高质量代码。", llm_config={"config_list": config_list} ) tester = autogen.AssistantAgent( name="测试员", system_message="你是测试工程师。接收编排者的任务，编写和执行测试。", llm_config={"config_list": config_list} ) # 用户代理 user_proxy = autogen.UserProxyAgent( name="用户", human_input_mode="NEVER", max_consecutive_auto_reply=10 ) # 建立群聊 groupchat = autogen.GroupChat( agents=[user_proxy, orchestrator, researcher, engineer, tester], messages=[], max_round=30 ) manager = autogen.GroupChatManager( groupchat=groupchat, llm_config={"config_list": config_list} ) # 启动协作 user_proxy.initiate_chat( manager, message="开发一个 Python 命令行工具，可以统计指定目录下各类文件的数量。" ) 2.3 层级式的优劣 优势： ...

为什么智能体系统需要特殊的负载均衡？ 传统的负载均衡主要处理无状态的 HTTP 请求，关注的是 QPS、延迟和可用性。但 AI 智能体系统的负载特征与传统 Web 服务有本质区别：单个请求的计算开销极大（一次 LLM 推理可能耗时数秒到数十秒）、请求之间可能存在状态依赖（多轮对话上下文）、且不同模型的资源需求差异巨大。 一个典型的 AGI 智能体系统可能同时包含：轻量级的意图分类（小模型，毫秒级响应）、重量级的推理任务（大模型，数十秒）、长周期的 Agent 工作流（多步骤，分钟级甚至小时级）、以及实时向量检索（RAG pipeline）。这些异构负载如果混在一起不加区分地调度，必然导致资源浪费和体验劣化。 智能体负载均衡的核心维度 1. 请求分类与路由 第一步是对进入系统的请求进行智能分类。不同于传统负载均衡器只看 URL 路径，智能体系统需要根据请求的语义特征进行路由： class RequestRouter: def __init__(self): self.routes = { "classification": {"model": "claude-haiku", "timeout": 2000, "pool": "fast"}, "reasoning": {"model": "gpt-4o", "timeout": 30000, "pool": "deep"}, "agent_workflow": {"model": "agent-orchestrator", "timeout": 300000, "pool": "long"}, "embedding": {"model": "text-embedding-3", "timeout": 1000, "pool": "vector"}, } def route(self, request): req_type = self.classify(request) route_config = self.routes[req_type] # 检查对应池的可用容量 if self.pool_available(route_config["pool"]): return self.dispatch(request, route_config) else: # 降级策略 return self.fallback(request, req_type) 关键设计点在于：不同类型的请求进入不同的处理池，彼此隔离，避免慢请求阻塞快请求。这就是所谓的"泳道隔离"模式。 2. 连接池与令牌桶 对于 LLM API 调用，两个核心限制是并发连接数和 token 速率限制。有效的管理需要同时处理这两个维度： 连接池管理：每个模型端点维护独立的连接池，池大小根据模型的并发能力和成本预算动态调整。例如，对于按量计费的 API，可以设置较高的并发上限；对于有 RPM 限制的 API，需要严格控制在限额以内。 令牌桶限流：不仅仅限制请求数量，更要限制 token 消耗速率。一个请求可能消耗 100 token，另一个可能消耗 10000 token。按 token 而非请求数来限流，才能真正做到公平调度： class TokenBucket: def __init__(self, capacity, refill_rate): self.capacity = capacity # 桶容量（最大突发） self.refill_rate = refill_rate # 每秒补充的 token 数 self.tokens = capacity self.last_refill = time.time() def consume(self, tokens_needed): self._refill() if self.tokens >= tokens_needed: self.tokens -= tokens_needed return True return False def _refill(self): now = time.time() elapsed = now - self.last_refill self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate) self.last_refill = now 3. 优先级队列 并非所有请求都同等重要。用户的实时对话请求应该优先于后台的批量处理任务。一个成熟的优先级策略通常包含 3-4 个级别： ...

为什么工具设计决定了智能体的上限 在 AI 智能体的技术栈中，大模型本身提供的是推理和决策能力，而真正让智能体"做事"的，是它可调用的工具集合。一个智能体的能力边界，不取决于它背后的模型有多强大，而取决于它的工具集设计得有多合理。这不是夸张——如果一个拥有顶级推理能力的模型面对的是一群设计糟糕的工具，它的表现会远不如一个中等模型配合精心设计工具的组合。 工具设计是被严重低估的工程 discipline。很多人认为工具设计就是写几个 API 然后包一层 function calling 接口，但实际上，好的工具设计需要在模型认知特性、工程可维护性、安全性和用户体验之间找到精妙的平衡。 核心设计模式 模式一：原子工具模式（Atomic Tool Pattern） 每个工具只做一件事，且做得很彻底。这是最基础也是最重要的模式。 设计原则： 单一职责：一个工具对应一个明确的操作 参数最小化：只要求模型提供必需的参数 输出结构化：返回 JSON 而非自然语言 示例对比： 糟糕的设计——把多个操作塞进一个工具： { "name": "manage_database", "parameters": { "operation": "create|read|update|delete", "table": "string", "data": "object", "condition": "object" } } 好的设计——拆分为原子工具： // 工具1 { "name": "query_records", "parameters": { "table": "string", "filters": "object", "limit": "integer" } } // 工具2 { "name": "create_record", "parameters": { "table": "string", "data": "object" } } 原子工具的优势在于模型更容易理解每个工具的用途，调用错误率显著降低。实测中，将复合工具拆分为原子工具后，工具调用准确率从 72% 提升到 91%。 模式二：工具组合模式（Tool Composition Pattern） 原子工具解决的是"做什么"，但很多任务需要多个工具按特定顺序协作。工具组合模式通过"编排工具"来封装常用的工具调用链。 设计方式： # 底层原子工具 @register_tool def search_product(query: str) -> list[dict]: """搜索商品""" @register_tool def get_product_detail(product_id: str) -> dict: """获取商品详情""" @register_tool def check_inventory(product_id: str, sku: str) -> dict: """检查库存""" # 组合工具 - 封装常见调用链 @register_tool def search_and_check_inventory(query: str) -> list[dict]: """搜索商品并返回含库存信息的结果列表""" products = search_product(query) results = [] for p in products[:5]: # 只检查前5个 inv = check_inventory(p["id"], p["default_sku"]) results.append({**p, "inventory": inv}) return results 组合工具减少了模型的决策负担和调用轮次。但注意不要过度组合——如果一个组合工具嵌套了超过 3 个原子工具，模型的错误率会急剧上升。 ...

引言：智能体编排的挑战 当单个 LLM 调用无法满足复杂任务需求时，智能体工作流编排应运而生。无论是多步推理、工具调用、还是多智能体协作，都需要一个可靠的编排层来管理执行流程。然而，智能体工作流的编排远比传统的数据处理管道复杂——执行路径可能动态变化，中间结果会影响后续步骤，错误处理需要语义级别的理解。 本文将从最基础的 DAG（有向无环图）编排出发，逐步深入到动态执行图、状态机模型，以及 LangGraph 等主流框架的设计哲学与实践方法。 一、静态 DAG：确定性流程的基石 1.1 什么是 DAG 编排 有向无环图（DAG）是最直观的工作流编排模型。节点表示执行单元（如 LLM 调用、工具调用、数据处理），边表示数据流向。DAG 的核心约束是"无环"——数据只能向前流动，不能回头。 一个典型的 RAG DAG 工作流： 用户查询 → 查询改写 → 向量检索 → 重排序 → 上下文构建 → LLM 生成 → 后处理 → 输出 这种线性流程是最简单的 DAG。更复杂的场景可能包含并行分支，例如同时检索多个数据源、同时调用多个工具，然后汇总结果。 1.2 DAG 的优势与局限 优势： 执行路径确定，易于调试和追踪 天然支持并行执行 可以静态分析依赖关系，优化调度策略 错误处理相对简单——只需处理节点级失败 局限： 无法处理需要循环或迭代的场景（如反思-修正循环） 执行路径在编译时确定，无法根据中间结果动态调整 对于条件分支的支持有限，通常只能通过预先定义所有可能路径来实现 二、条件分支与循环：打破 DAG 的限制 2.1 条件路由 实际业务中，执行路径往往需要根据中间结果动态决定。例如： 检索结果置信度低 → 触发补充检索 生成答案包含不确定信息 → 触发事实核查 用户意图分类为"闲聊" → 跳过检索直接回复 条件路由的实现方式通常是在节点输出上定义路由函数，根据输出内容选择下一个执行节点。这虽然打破了纯 DAG 的"无环"约束，但仍保持了流程的可预测性。 ...

1. 为什么 Agent 需要专门的可观测性 传统微服务可观测性关注请求-响应链路。Agent 系统则复杂得多： 多步骤推理：一个用户请求可能触发 10-50 次 LLM 调用 不确定输出：相同输入可能产生不同输出，难以复现问题 工具调用链：Agent → tool1 → Agent → tool2 → … 形成深层调用链 成本高昂：每次 LLM 调用都有实际成本，需要精准归因 幻觉风险：输出质量难以量化监控 普通监控工具（Prometheus/Grafana）只覆盖指标，无法完整追踪 Agent 行为。需要专门的 Agent 可观测性方案。 2. 可观测性三大支柱 ┌─────────────────────────────────────────────────┐ │ Agent Observability │ ├────────────┬──────────────┬────────────────────┤ │ Traces │ Metrics │ Logs │ │ (追踪) │ (指标) │ (日志) │ │ │ │ │ │ ┌────────┐ │ ┌────────┐ │ ┌──────────────┐ │ │ │Span │ │ │Counter │ │ │ Struct- │ │ │ │Tree │ │ │Gauge │ │ │ ured Log │ │ │ │Timeline│ │ │Histog. │ │ │ Event Log │ │ │ └────────┘ │ └────────┘ │ └──────────────┘ │ │ │ │ │ │ "发生了 │ "发生了几次" │ "发生了什么" │ │ 什么" │ "耗时多久" │ "为什么失败" │ └────────────┴──────────────┴────────────────────┘ 统一关联：TraceID + SpanID + LogID 3. 分布式追踪设计 3.1 Agent Span 模型 from dataclasses import dataclass, field from datetime import datetime from enum import Enum from typing import Optional import uuid class SpanKind(str, Enum): LLM_CALL = "llm_call" TOOL_CALL = "tool_call" AGENT_STEP = "agent_step" RETRIEVAL = "retrieval" USER_INPUT = "user_input" SYSTEM = "system" @dataclass class SpanContext: trace_id: str span_id: str parent_span_id: Optional[str] = None @dataclass class SpanEvent: name: str timestamp: datetime attributes: dict = field(default_factory=dict) @dataclass class AgentSpan: """Agent 追踪 Span - 兼容 OpenTelemetry""" trace_id: str span_id: str parent_span_id: Optional[str] kind: SpanKind name: str agent_id: str session_id: str # 时间 start_time: datetime = field(default_factory=datetime.now) end_time: Optional[datetime] = None duration_ms: Optional[float] = None # 输入/输出 input: dict = field(default_factory=dict) output: dict = field(default_factory=dict) # 状态 status: str = "ok" # ok, error, timeout error_message: Optional[str] = None # 成本 & 性能 token_usage: dict = field(default_factory=dict) # {input: N, output: N} cost_usd: float = 0.0 model: Optional[str] = None # 扩展属性 attributes: dict = field(default_factory=dict) events: list[SpanEvent] = field(default_factory=list) child_spans: list[str] = field(default_factory=list) # child span_ids def finish(self, status: str = "ok"): self.end_time = datetime.now() self.duration_ms = (self.end_time - self.start_time).total_seconds() * 1000 self.status = status def add_event(self, name: str, attributes: dict = None): self.events.append(SpanEvent( name=name, timestamp=datetime.now(), attributes=attributes or {} )) 3.2 追踪器实现 class AgentTracer: """Agent 分布式追踪器""" def __init__(self, exporter: "TraceExporter"): self.exporter = exporter self._active_spans: dict[str, AgentSpan] = {} # span_id -> span self._trace_tree: dict[str, list[str]] = {} # trace_id -> [span_ids] def start_trace(self, session_id: str, agent_id: str, name: str = "agent_session") -> tuple[str, str]: """开始一次 Agent 会话追踪""" trace_id = str(uuid.uuid4()) root_span_id = self._new_span_id() span = AgentSpan( trace_id=trace_id, span_id=root_span_id, parent_span_id=None, kind=SpanKind.AGENT_STEP, name=name, agent_id=agent_id, session_id=session_id, ) self._active_spans[root_span_id] = span self._trace_tree[trace_id] = [root_span_id] return trace_id, root_span_id def start_span(self, trace_id: str, parent_span_id: str, kind: SpanKind, name: str, agent_id: str, session_id: str, input_data: dict = None) -> str: span_id = self._new_span_id() span = AgentSpan( trace_id=trace_id, span_id=span_id, parent_span_id=parent_span_id, kind=kind, name=name, agent_id=agent_id, session_id=session_id, input=input_data or {}, ) self._active_spans[span_id] = span self._trace_tree.setdefault(trace_id, []).append(span_id) # 更新父 span 的 child 列表 if parent_span_id in self._active_spans: self._active_spans[parent_span_id].child_spans.append(span_id) return span_id def finish_span(self, span_id: str, output_data: dict = None, status: str = "ok", error: str = None): if span_id not in self._active_spans: return span = self._active_spans[span_id] span.finish(status) if output_data: span.output = output_data if error: span.error_message = error # 异步导出 asyncio.create_task(self.exporter.export_span(span)) def get_trace(self, trace_id: str) -> list[AgentSpan]: span_ids = self._trace_tree.get(trace_id, []) return [self._active_spans[sid] for sid in span_ids if sid in self._active_spans] @staticmethod def _new_span_id() -> str: return str(uuid.uuid4())[:16] # 短 ID，便于展示 3.3 OpenTelemetry 集成 from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter class OTelIntegration: """将 Agent Span 桥接到 OpenTelemetry""" def __init__(self, endpoint: str = "localhost:4317"): self.provider = TracerProvider() self.exporter = OTLPSpanExporter(endpoint=endpoint) self.processor = BatchSpanProcessor(self.exporter) self.provider.add_span_processor(self.processor) trace.set_tracer_provider(self.provider) self.tracer = trace.get_tracer("agent-system") def to_otel_span(self, agent_span: AgentSpan): """将 AgentSpan 转换为 OTel Span""" with self.tracer.start_as_current_span( agent_span.name, context=self._make_context(agent_span), kind=self._map_kind(agent_span.kind), ) as otel_span: # 设置属性 otel_span.set_attribute("agent.id", agent_span.agent_id) otel_span.set_attribute("session.id", agent_span.session_id) otel_span.set_attribute("llm.model", agent_span.model or "") otel_span.set_attribute("cost.usd", agent_span.cost_usd) for k, v in agent_span.token_usage.items(): otel_span.set_attribute(f"token.{k}", v) # 设置状态 if agent_span.status == "error": otel_span.set_status(trace.Status(trace.StatusCode.ERROR, agent_span.error_message)) # 添加事件 for evt in agent_span.events: otel_span.add_event(evt.name, evt.attributes) def _map_kind(self, kind: SpanKind): mapping = { SpanKind.LLM_CALL: trace.SpanKind.CLIENT, SpanKind.TOOL_CALL: trace.SpanKind.INTERNAL, SpanKind.AGENT_STEP: trace.SpanKind.INTERNAL, SpanKind.RETRIEVAL: trace.SpanKind.CLIENT, } return mapping.get(kind, trace.SpanKind.INTERNAL) 4. 指标采集 4.1 关键指标定义 from prometheus_client import Counter, Histogram, Gauge, Summary import time class AgentMetrics: """Agent 系统关键指标""" def __init__(self, namespace: str = "agent"): # LLM 调用指标 self.llm_requests = Counter( f"{namespace}_llm_requests_total", "Total LLM requests", ["model", "agent_id", "status"] ) self.llm_latency = Histogram( f"{namespace}_llm_latency_seconds", "LLM request latency", ["model", "agent_id"], buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0] ) self.llm_tokens = Counter( f"{namespace}_llm_tokens_total", "Total tokens consumed", ["model", "direction"] # direction: input/output ) self.llm_cost = Counter( f"{namespace}_llm_cost_usd_total", "Total LLM cost in USD", ["model", "agent_id"] ) # Agent 行为指标 self.agent_steps = Counter( f"{namespace}_agent_steps_total", "Total agent reasoning steps", ["agent_id", "step_type"] ) self.agent_session_duration = Histogram( f"{namespace}_agent_session_duration_seconds", "Agent session duration", ["agent_id", "outcome"] # outcome: success/failure/timeout ) self.agent_tool_calls = Counter( f"{namespace}_agent_tool_calls_total", "Total tool calls", ["agent_id", "tool_name", "status"] ) # 系统指标 self.active_sessions = Gauge( f"{namespace}_active_sessions", "Current active sessions", ["agent_id"] ) self.queue_depth = Gauge( f"{namespace}_queue_depth", "Current queue depth", ["queue_name"] ) self.error_rate = Summary( f"{namespace}_error_rate", "Error rate over 5m window", ["agent_id", "error_type"] ) def record_llm_call(self, model: str, agent_id: str, duration: float, input_tokens: int, output_tokens: int, cost: float, status: str): self.llm_requests.labels(model=model, agent_id=agent_id, status=status).inc() self.llm_latency.labels(model=model, agent_id=agent_id).observe(duration) self.llm_tokens.labels(model=model, direction="input").inc(input_tokens) self.llm_tokens.labels(model=model, direction="output").inc(output_tokens) self.llm_cost.labels(model=model, agent_id=agent_id).inc(cost) def record_agent_step(self, agent_id: str, step_type: str): self.agent_steps.labels(agent_id=agent_id, step_type=step_type).inc() def record_tool_call(self, agent_id: str, tool_name: str, status: str): self.agent_tool_calls.labels(agent_id=agent_id, tool_name=tool_name, status=status).inc() 4.2 成本归因指标 from collections import defaultdict class CostAttribution: """成本归因：按用户/会话/功能分解成本""" def __init__(self, redis_client): self.redis = redis_client async def record_cost(self, trace_id: str, user_id: str, feature: str, model: str, cost: float, tokens: int): pipe = self.redis.pipeline() # 按用户聚合 pipe.incrbyfloat(f"cost:user:{user_id}:total", cost) pipe.incrby(f"cost:user:{user_id}:tokens", tokens) # 按功能聚合 pipe.incrbyfloat(f"cost:feature:{feature}:total", cost) # 按模型聚合 pipe.incrbyfloat(f"cost:model:{model}:total", cost) # 按 trace 明细 pipe.hset(f"cost:trace:{trace_id}", mapping={ "user_id": user_id, "feature": feature, "model": model, "cost": str(cost), "tokens": str(tokens), "timestamp": str(time.time()) }) await pipe.execute() async def get_user_cost_report(self, user_id: str, days: int = 7) -> dict: return { "total_cost_usd": float(await self.redis.get(f"cost:user:{user_id}:total") or 0), "total_tokens": int(await self.redis.get(f"cost:user:{user_id}:tokens") or 0), "by_feature": await self._get_by_dimension(f"cost:feature", user_id, days), "by_model": await self._get_by_dimension(f"cost:model", user_id, days), } 5. 结构化日志 5.1 日志格式规范 import json import logging import sys from datetime import datetime from typing import Any class StructuredLogger: """结构化日志：JSON 格式，便于 ELK/Loki 检索""" def __init__(self, service_name: str, level: str = "INFO"): self.service = service_name self.logger = logging.getLogger(service_name) self.logger.setLevel(getattr(logging, level)) handler = logging.StreamHandler(sys.stdout) handler.setFormatter(self) self.logger.addHandler(handler) def format(self, record: logging.LogRecord) -> str: log_entry = { "timestamp": datetime.utcnow().isoformat() + "Z", "level": record.levelname, "service": self.service, "message": record.getMessage(), "trace_id": getattr(record, "trace_id", None), "span_id": getattr(record, "span_id", None), "session_id": getattr(record, "session_id", None), "agent_id": getattr(record, "agent_id", None), "user_id": getattr(record, "user_id", None), } # 添加额外字段 if hasattr(record, "extra_fields"): log_entry.update(record.extra_fields) return json.dumps(log_entry, ensure_ascii=False) def bind(self, **kwargs) -> "BoundLogger": return BoundLogger(self, kwargs) class BoundLogger: """绑定上下文的 Logger""" def __init__(self, logger: StructuredLogger, bound: dict): self._logger = logger self._bound = bound def _log(self, level: str, msg: str, **kwargs): extra = {**self._bound, **kwargs} record = self._logger.logger.makeRecord( self._logger.logger.name, getattr(logging, level), "(unknown)", 0, msg, [], None ) record.extra_fields = extra for k, v in extra.items(): setattr(record, k, v) self._logger.logger.handle(record) def info(self, msg: str, **kwargs): self._log("INFO", msg, **kwargs) def warning(self, msg: str, **kwargs): self._log("WARNING", msg, **kwargs) def error(self, msg: str, **kwargs): self._log("ERROR", msg, **kwargs) def debug(self, msg: str, **kwargs): self._log("DEBUG", msg, **kwargs) 5.2 日志事件类型 class AgentLogEvents: """Agent 关键生命周期事件""" @staticmethod def session_started(logger: BoundLogger, session_id: str, user_id: str): logger.info("agent.session.started", session_id=session_id, user_id=user_id, event_type="session_lifecycle") @staticmethod def llm_call_started(logger: BoundLogger, model: str, prompt_length: int): logger.info("llm.call.started", model=model, prompt_length=prompt_length, event_type="llm_call") @staticmethod def llm_call_completed(logger: BoundLogger, model: str, latency_ms: float, input_tokens: int, output_tokens: int, cost: float): logger.info("llm.call.completed", model=model, latency_ms=latency_ms, input_tokens=input_tokens, output_tokens=output_tokens, cost_usd=cost, event_type="llm_call") @staticmethod def tool_call_started(logger: BoundLogger, tool_name: str, args: dict): logger.info("tool.call.started", tool_name=tool_name, tool_args=args, event_type="tool_call") @staticmethod def tool_call_completed(logger: BoundLogger, tool_name: str, success: bool, result_summary: str): logger.info("tool.call.completed", tool_name=tool_name, success=success, result_summary=result_summary[:200], event_type="tool_call") @staticmethod def hallucination_detected(logger: BoundLogger, claim: str, confidence: float): logger.warning("agent.hallucination.detected", claim=claim[:100], confidence=confidence, event_type="quality") @staticmethod def session_completed(logger: BoundLogger, session_id: str, total_steps: int, total_cost: float, outcome: str): logger.info("agent.session.completed", session_id=session_id, total_steps=total_steps, total_cost_usd=total_cost, outcome=outcome, event_type="session_lifecycle") 6. 完整可观测性集成 class ObservableAgent: """集成可观测性的 Agent 基类""" def __init__(self, agent_id: str, tracer: AgentTracer, metrics: AgentMetrics, logger: StructuredLogger): self.agent_id = agent_id self.tracer = tracer self.metrics = metrics self.logger = logger.bind(agent_id=agent_id) self._current_trace_id: str | None = None self._current_span_id: str | None = None async def run(self, user_query: str, user_id: str, session_id: str) -> str: # 开始追踪 trace_id, root_span_id = self.tracer.start_trace(session_id, self.agent_id) self._current_trace_id = trace_id self._current_span_id = root_span_id # 绑定 trace 上下文到日志 self.logger = self.logger.bind(trace_id=trace_id, session_id=session_id, user_id=user_id) session_start = time.time() try: self.logger.info("agent.session.started", query=user_query[:100]) self.metrics.active_sessions.labels(agent_id=self.agent_id).inc() # 执行 Agent 逻辑（带追踪） result = await self._execute_with_tracing(user_query) duration = time.time() - session_start self.metrics.agent_session_duration.labels( agent_id=self.agent_id, outcome="success" ).observe(duration) self.logger.info("agent.session.completed", duration_seconds=duration, outcome="success") self.tracer.finish_span(root_span_id, {"result": result}, "ok") return result except Exception as e: duration = time.time() - session_start self.metrics.agent_session_duration.labels( agent_id=self.agent_id, outcome="failure" ).observe(duration) self.logger.error("agent.session.failed", error=str(e), exc_info=True) self.tracer.finish_span(root_span_id, status="error", error=str(e)) raise finally: self.metrics.active_sessions.labels(agent_id=self.agent_id).dec() async def _execute_with_tracing(self, query: str) -> str: # LLM 调用 llm_span_id = self.tracer.start_span( self._current_trace_id, self._current_span_id, SpanKind.LLM_CALL, "llm_chat", self.agent_id, self._get_session_id(), {"query": query} ) llm_start = time.time() try: self.logger.info("llm.call.started") response = await self._call_llm(query) latency = time.time() - llm_start self.metrics.record_llm_call( model="gpt-4o", agent_id=self.agent_id, duration=latency, input_tokens=100, output_tokens=50, cost=0.003, status="ok" ) self.tracer.finish_span(llm_span_id, {"response": response}, "ok") return response except Exception as e: self.tracer.finish_span(llm_span_id, status="error", error=str(e)) raise def _get_session_id(self) -> str: return self.logger._bound.get("session_id", "") 7. 可观测性数据流水线 Agent 运行时 │ ├── Traces ──→ OTel Collector ──→ Jaeger/Tempo ──→ Grafana │ ├── Metrics ──→ Prometheus ──────────────────────→ Grafana │ └── Logs ────→ Loki/ELK ───────────────────────→ Grafana │ └──→ 告警规则 → AlertManager → Slack/PagerDuty 7.1 OpenTelemetry Collector 配置 # otel-collector-config.yaml receivers: otlp: protocols: grpc: endpoint: 0.0.0.0:4317 http: endpoint: 0.0.0.0:4318 processors: batch: timeout: 5s send_batch_size: 100 memory_limiter: check_interval: 5s limit_mib: 1000 attributes: actions: - key: agent.id action: insert value: "unknown" resource: attributes: - key: deployment.environment value: "production" action: insert exporters: otlp/jaeger: endpoint: jaeger:4317 tls: insecure: true prometheus: endpoint: 0.0.0.0:8889 loki: endpoint: http://loki:3100/loki/api/v1/push service: pipelines: traces: receivers: [otlp] processors: [batch, memory_limiter, attributes] exporters: [otlp/jaeger] metrics: receivers: [otlp] processors: [batch, memory_limiter] exporters: [prometheus] logs: receivers: [otlp] processors: [batch, memory_limiter] exporters: [loki] 8. 告警规则 # prometheus-alerts.yaml groups: - name: agent_alerts rules: - alert: AgentHighErrorRate expr: | sum(rate(agent_agent_steps_total{status="error"}[5m])) by (agent_id) / sum(rate(agent_agent_steps_total[5m])) by (agent_id) > 0.05 for: 2m labels: severity: warning annotations: summary: "Agent {{ $labels.agent_id }} 错误率超过 5%" - alert: AgentHighLatency expr: | histogram_quantile(0.95, sum(rate(agent_llm_latency_seconds_bucket[5m])) by (le, agent_id) ) > 5 for: 2m labels: severity: warning annotations: summary: "Agent {{ $labels.agent_id }} P95 延迟超过 5 秒" - alert: AgentHighCost expr: | rate(agent_llm_cost_usd_total[1h]) > 10 for: 5m labels: severity: critical annotations: summary: "Agent 每小时成本超过 $10" - alert: AgentHallucinationDetected expr: | increase(agent_quality_hallucination_total[10m]) > 5 labels: severity: warning annotations: summary: "检测到 {{ $value }} 次幻觉" 9. 总结 Agent 可观测性的核心在于全链路关联： ...

1. 推理服务的核心挑战 模型训练完成后，推理服务是将模型价值交付给用户的关键环节。与训练不同，推理服务面临独特挑战： 延迟敏感：用户期望秒级甚至毫秒级响应 吞吐与延迟权衡：批处理提高吞吐但增加单请求延迟 GPU 资源昂贵：需最大化利用率 弹性需求：流量峰谷差异可达 10 倍以上 多模型管理：同时服务数十个模型版本 2. 架构演进路线 阶段1: 单机单GPU 阶段2: 单机多GPU 阶段3: 分布式集群 ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │ FastAPI │ │ FastAPI │ │ API Gateway │ │ + HF Model │ → │ + vLLM │ → │ + vLLM Cluster │ │ + 1 GPU │ │ + 4 GPUs │ │ + K8s Autoscale │ └──────────────┘ └──────────────┘ └──────────────────┘ 延迟: 500ms 延迟: 200ms 延迟: 150ms QPS: 5 QPS: 80 QPS: 1000+ 3. 推理引擎选型 3.1 引擎对比 引擎 优势 劣势 适用场景 vLLM PagedAttention，高吞吐 仅支持主流模型 LLM 生产服务 TensorRT-LLM 极致优化，NVIDIA 官方 编译复杂，灵活性低 极致性能场景 TGI HF 官方，生态好 性能略逊 vLLM HF 模型服务 Ollama 部署简单 性能一般 本地/开发环境 SGLang 结构化生成优化 生态较新 结构化输出 3.2 vLLM 部署 # vLLM 服务启动配置 VLLM_CONFIG = { "model": "meta-llama/Llama-3-70B", "tensor_parallel_size": 4, # 4 GPU 张量并行 "gpu_memory_utilization": 0.90, # GPU 内存利用率 "max_model_len": 8192, # 最大上下文长度 "max_num_seqs": 256, # 最大并发序列 "swap_space": 16, # CPU 交换空间 (GB) "enable_prefix_caching": True, # 前缀缓存 "enable_chunked_prefill": True, # 分块预填充 "enforce_eager": False, # 使用 CUDA Graph "dtype": "float16", # 精度 "quantization": "awq", # 量化 "served_model_name": "llama-3-70b", # 对外模型名 } # 启动命令 # python -m vllm.entrypoints.openai.api_server --config config.json 3.3 TensorRT-LLM 优化 import tensorrt as trt import tensorrt_llm from tensorrt_llm.runtime import ModelRunner, GenerationSession class TensorRTEngine: """ TensorRT-LLM 推理引擎 构建 → 序列化 → 反序列化 → 推理 """ def __init__(self, model_dir: str, engine_dir: str): self.model_dir = model_dir self.engine_dir = engine_dir self.runner: ModelRunner = None def build_engine(self, config: dict): """构建 TensorRT 引擎（一次性操作）""" builder = tensorrt_llm.Builder() builder_config = builder.create_builder_config( name="llama", precision=config.get("precision", "fp16"), tensor_parallel_size=config.get("tp_size", 1), pipeline_parallel_size=config.get("pp_size", 1), max_batch_size=config.get("max_batch_size", 256), max_input_len=config.get("max_input_len", 2048), max_output_len=config.get("max_output_len", 512), max_beam_width=config.get("max_beam_width", 1), use_gpt_attention=config.get("use_gpt_attention", True), use_plugin=True, ) # 构建并序列化引擎 engine = builder.build_model(builder_config, self.model_dir) engine.serialize(self.engine_dir) def load_engine(self): """加载已构建的引擎""" self.runner = ModelRunner.from_dir( self.engine_dir, rank=0, is_pp_first=True, is_pp_last=True ) async def generate(self, prompt_ids: list[int], max_new_tokens: int = 256) -> list[int]: session = GenerationSession( model_config=self.runner.model_config, engine=self.runner.engine, input_ids=prompt_ids ) session.decode(max_new_tokens=max_new_tokens) return session.output_ids 4. 批处理与调度 4.1 动态批处理 import asyncio from dataclasses import dataclass, field from collections import deque import time @dataclass class InferenceRequest: request_id: str input_ids: list[int] max_tokens: int temperature: float = 1.0 top_p: float = 0.9 stream: bool = False arrival_time: float = field(default_factory=time.time) future: asyncio.Future = field(default_factory=asyncio.get_event_loop().create_future) class DynamicBatcher: """ 动态批处理器：收集请求到批次中，达到阈值或超时后提交 核心：在延迟和吞吐之间权衡 """ def __init__(self, engine, max_batch_size: int = 32, max_wait_ms: float = 50, max_seq_len: int = 2048): self.engine = engine self.max_batch_size = max_batch_size self.max_wait_ms = max_wait_ms self.max_seq_len = max_seq_len self.queue: deque[InferenceRequest] = deque() self.running = False async def submit(self, request: InferenceRequest) -> str: self.queue.append(request) return await request.future async def start(self): self.running = True while self.running: batch = await self._collect_batch() if batch: await self._process_batch(batch) async def _collect_batch(self) -> list[InferenceRequest]: batch = [] deadline = time.time() + self.max_wait_ms / 1000 while len(batch) < self.max_batch_size and time.time() < deadline: if self.queue: req = self.queue.popleft() batch.append(req) else: await asyncio.sleep(0.001) return batch async def _process_batch(self, batch: list[InferenceRequest]): # Padding 到相同长度 max_len = min(max(len(r.input_ids) for r in batch), self.max_seq_len) padded_inputs = [] attention_masks = [] for req in batch: padding_len = max_len - len(req.input_ids[:max_len]) padded = req.input_ids[:max_len] + [0] * padding_len mask = [1] * min(len(req.input_ids), max_len) + [0] * padding_len padded_inputs.append(padded) attention_masks.append(mask) # 执行批量推理 outputs = await self.engine.generate_batch( input_ids=padded_inputs, attention_mask=attention_masks, max_new_tokens=max(r.max_tokens for r in batch), temperature=max(r.temperature for r in batch), ) # 分发结果 for req, output in zip(batch, outputs): if not req.future.done(): req.future.set_result(output) 4.2 连续批处理（Continuous Batching） class ContinuousBatcher: """ 连续批处理（Iteration-Level Scheduling） 与传统批处理不同：不需要等所有请求完成才接受新请求 每个解码步后都可以加入新请求、移除已完成请求 """ def __init__(self, engine, max_batch_size: int = 256): self.engine = engine self.max_batch_size = max_batch_size self.active_sequences: list[Sequence] = [] self.waiting_queue: deque[Sequence] = deque() @dataclass class Sequence: request_id: str input_ids: list[int] output_ids: list[int] = field(default_factory=list) max_tokens: int = 256 is_finished: bool = False future: asyncio.Future = None async def submit(self, request: InferenceRequest) -> str: seq = self.Sequence( request_id=request.request_id, input_ids=request.input_ids, max_tokens=request.max_tokens, future=asyncio.get_event_loop().create_future() ) self.waiting_queue.append(seq) return await seq.future async def run(self): while True: # 1. 从等待队列添加新序列到活跃批次 while (self.waiting_queue and len(self.active_sequences) < self.max_batch_size): self.active_sequences.append(self.waiting_queue.popleft()) if not self.active_sequences: await asyncio.sleep(0.001) continue # 2. 执行单步解码 step_outputs = await self.engine.step_decode( [s.input_ids + s.output_ids for s in self.active_sequences] ) # 3. 处理输出 finished = [] for seq, output in zip(self.active_sequences, step_outputs): seq.output_ids.append(output.token_id) if output.is_eos or len(seq.output_ids) >= seq.max_tokens: seq.is_finished = True seq.future.set_result(seq.output_ids) finished.append(seq) # 4. 移除已完成序列 self.active_sequences = [s for s in self.active_sequences if not s.is_finished] 5. GPU 资源调度 5.1 多模型 GPU 共享 class GPUScheduler: """GPU 资源调度器：管理多模型在同一 GPU 上的调度""" def __init__(self, gpu_memory_total: int = 80_000): # MB self.total_memory = gpu_memory_total self.allocated: dict[str, int] = {} # model_id -> memory_mb self.model_streams: dict[str, int] = {} # model_id -> cuda_stream def can_load(self, model_memory: int) -> bool: used = sum(self.allocated.values()) return (self.total_memory - used) >= model_memory def allocate(self, model_id: str, memory_mb: int) -> int: if not self.can_load(memory_mb): raise RuntimeError(f"Insufficient GPU memory: need {memory_mb}MB, " f"available {self.total_memory - sum(self.allocated.values())}MB") self.allocated[model_id] = memory_mb return len(self.allocated) # stream id def deallocate(self, model_id: str): self.allocated.pop(model_id, None) self.model_streams.pop(model_id, None) def get_utilization(self) -> float: return sum(self.allocated.values()) / self.total_memory class ModelWarmPool: """模型热池：常用模型常驻 GPU，不常用模型按需加载""" def __init__(self, scheduler: GPUScheduler, storage_path: str): self.scheduler = scheduler self.storage = storage_path self.loaded: dict[str, Any] = {} # model_id -> model instance self.loading_locks: dict[str, asyncio.Lock] = {} self.access_times: dict[str, float] = {} self.idle_threshold = 600 # 10分钟未使用则卸载 async def get_model(self, model_id: str, model_config: dict): if model_id in self.loaded: self.access_times[model_id] = time.time() return self.loaded[model_id] # 防止并发加载同一模型 if model_id not in self.loading_locks: self.loading_locks[model_id] = asyncio.Lock() async with self.loading_locks[model_id]: if model_id in self.loaded: return self.loaded[model_id] # 检查 GPU 内存是否足够 required_mem = model_config["memory_mb"] if not self.scheduler.can_load(required_mem): await self._evict_idle_models(required_mem) # 加载模型 model = await self._load_model(model_id, model_config) self.scheduler.allocate(model_id, required_mem) self.loaded[model_id] = model self.access_times[model_id] = time.time() return model async def _evict_idle_models(self, required_mem: int): """LRU 淘汰空闲模型""" sorted_models = sorted(self.access_times.items(), key=lambda x: x[1]) for model_id, last_access in sorted_models: if time.time() - last_access > self.idle_threshold: await self._unload_model(model_id) if self.scheduler.can_load(required_mem): break async def _unload_model(self, model_id: str): if model_id in self.loaded: del self.loaded[model_id] self.scheduler.deallocate(model_id) self.access_times.pop(model_id, None) import torch torch.cuda.empty_cache() 6. 弹性伸缩 6.1 自动扩缩容控制器 class Autoscaler: """基于指标的自定扩缩容""" def __init__(self, min_replicas: int = 1, max_replicas: int = 10, target_gpu_util: float = 0.7, target_latency_ms: float = 500): self.min_replicas = min_replicas self.max_replicas = max_replicas self.target_gpu_util = target_gpu_util self.target_latency = target_latency_ms self.current_replicas = min_replicas def decide(self, metrics: dict) -> dict: gpu_util = metrics.get("gpu_utilization", 0) avg_latency = metrics.get("avg_latency_ms", 0) queue_depth = metrics.get("queue_depth", 0) qps = metrics.get("qps", 0) # 扩容信号 scale_up_score = 0 if gpu_util > self.target_gpu_util * 1.2: scale_up_score += 2 if avg_latency > self.target_latency * 1.5: scale_up_score += 2 if queue_depth > 50: scale_up_score += 1 # 缩容信号 scale_down_score = 0 if gpu_util < self.target_gpu_util * 0.3: scale_down_score += 2 if avg_latency < self.target_latency * 0.3 and queue_depth < 5: scale_down_score += 1 if qps < 1 and self.current_replicas > 1: scale_down_score += 1 if scale_up_score >= 2 and self.current_replicas < self.max_replicas: return {"action": "scale_up", "replicas": self.current_replicas + 1} elif scale_down_score >= 2 and self.current_replicas > self.min_replicas: return {"action": "scale_down", "replicas": self.current_replicas - 1} else: return {"action": "none", "replicas": self.current_replicas} 6.2 Kubernetes 部署配置 # vllm-deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: vllm-llama-70b spec: replicas: 2 selector: matchLabels: app: vllm-llama-70b template: metadata: labels: app: vllm-llama-70b spec: containers: - name: vllm image: vllm/vllm-openai:latest args: - --model=meta-llama/Llama-3-70B - --tensor-parallel-size=4 - --gpu-memory-utilization=0.90 - --max-model-len=8192 resources: limits: nvidia.com/gpu: 4 memory: 128Gi requests: nvidia.com/gpu: 4 memory: 64Gi readinessProbe: httpGet: path: /health port: 8000 initialDelaySeconds: 120 periodSeconds: 10 --- apiVersion: autoscaling/v2 kind: HorizontalPodAutoscaler metadata: name: vllm-hpa spec: scaleTargetRef: apiVersion: apps/v1 kind: Deployment name: vllm-llama-70b minReplicas: 2 maxReplicas: 10 metrics: - type: Pods pods: metric: name: gpu_utilization target: type: AverageValue averageValue: "70" - type: Resource resource: name: memory target: type: Utilization averageUtilization: 80 7. 负载均衡 class InferenceLoadBalancer: """推理负载均衡器：感知 GPU 负载和模型位置""" def __init__(self): self.backends: dict[str, BackendInfo] = {} @dataclass class BackendInfo: url: str model_id: str gpu_util: float = 0.0 active_requests: int = 0 queue_depth: int = 0 avg_latency_ms: float = 0.0 healthy: bool = True def select(self, model_id: str) -> str: candidates = [ (url, info) for url, info in self.backends.items() if info.model_id == model_id and info.healthy ] if not candidates: raise RuntimeError(f"No healthy backend for model {model_id}") # 最少连接 + 最低延迟加权 def score(info: BackendInfo) -> float: load_score = info.active_requests * 0.5 + info.queue_depth * 0.3 latency_score = info.avg_latency_ms / 1000 * 0.2 return load_score + latency_score return min(candidates, key=lambda x: score(x[1]))[0] def update_health(self, url: str, metrics: dict): if url in self.backends: info = self.backends[url] info.gpu_util = metrics.get("gpu_util", info.gpu_util) info.active_requests = metrics.get("active_requests", info.active_requests) info.avg_latency_ms = metrics.get("avg_latency_ms", info.avg_latency_ms) info.healthy = metrics.get("healthy", True) 8. 性能优化总结 优化手段 吞吐提升 延迟降低 实现复杂度 动态批处理 3-5x -10% ⭐⭐ 连续批处理 5-10x -20% ⭐⭐⭐ TensorRT 优化 2-3x -40% ⭐⭐⭐⭐ 前缀缓存 2x -50% ⭐⭐ AWQ/GPTQ 量化 2x -30% ⭐⭐ Speculative Decoding 1.5x -40% ⭐⭐⭐⭐ 多 GPU 并行 线性 持平 ⭐⭐⭐ 9. 总结 AI 推理服务的架构设计需要平衡延迟、吞吐和成本： ...

1. 从传统数据库到 AI 原生数据库 传统关系数据库为结构化数据设计，以行/列为基本单位，通过 B+ 树索引加速点查询和范围查询。AI 时代新增了一类核心需求：向量检索——在高维空间中寻找与查询向量最相似的 Top-K 条记录。 AI 原生数据库（AI-Native Database）将向量检索与结构化查询深度融合，支持： SELECT * FROM articles WHERE category = 'AI' ORDER BY embedding <-> query_vector LIMIT 10 语义相似性过滤、混合排序、元数据 + 向量联合查询 代表系统：Pinecone、Weaviate、Chroma、Qdrant、Milvus、pgvector（PostgreSQL 扩展）、StarRocks（向量化）、ClickHouse（向量化）。 2. 存储引擎架构 2.1 整体架构 ┌──────────────────────────────────────────────────┐ │ SQL / GraphQL / REST API │ ├──────────────────────────────────────────────────┤ │ Query Optimizer & Planner │ │ ┌─────────────┐ ┌─────────────┐ ┌────────┐ │ │ │Scalar Filter│ │ Vector Index│ │Sort/Max│ │ │ │ Planner │ │ Planner │ │ Planner│ │ │ └─────────────┘ └─────────────┘ └────────┘ │ ├──────────────────────────────────────────────────┤ │ Execution Engine (Vectorized) │ ├──────────────────────────────────────────────────┤ │ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │ │ Vector │ │ Row │ │ Column │ │ │ │ Index │ │ Store │ │ Store │ │ │ │(HNSW/IVF)│ │ (Pilosa) │ │(Arrow/Parquet)│ │ │ └──────────┘ └──────────┘ └──────────────┘ │ ├──────────────────────────────────────────────────┤ │ Distributed Storage Layer │ │ ┌────────────────────────────────────────────┐ │ │ │ S3 / HDFS / Local SSD / Remote Memory │ │ │ └────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────┘ 2.2 向量存储格式 from dataclasses import dataclass, field from typing import Optional import numpy as np @dataclass class VectorRecord: id: str vector: np.ndarray # 原始向量（float32/float16) metadata: dict = field(default_factory=dict) version: int = 0 # MVCC 版本号 deleted: bool = False def to_bytes(self) -> bytes: """序列化：4字节维度 + 原始向量 + 元数据JSON + 版本""" dim = len(self.vector) vec_bytes = self.vector.astype(np.float32).tobytes() meta_bytes = json.dumps(self.metadata).encode("utf-8") header = struct.pack("<i", dim) version_bytes = struct.pack("<i", self.version) deleted_byte = b"\x01" if self.deleted else b"\x00" return header + version_bytes + deleted_byte + vec_bytes + meta_bytes @classmethod def from_bytes(cls, data: bytes) -> "VectorRecord": dim, version = struct.unpack("<ii", data[:8]) deleted = data[8] == 1 vec_bytes = data[9:9 + dim * 4] meta_bytes = data[9 + dim * 4:].decode("utf-8") return cls( id="", # id 需要从索引层获取 vector=np.frombuffer(vec_bytes, dtype=np.float32), metadata=json.loads(meta_bytes), version=version, deleted=deleted ) class VectorPage: """向量分页存储单元""" def __init__(self, page_size: int = 4096): self.page_size = page_size self.records: dict[str, bytes] = {} # id -> serialized bytes self.current_size = 0 def add(self, record: VectorRecord) -> bool: record_bytes = record.to_bytes() if self.current_size + len(record_bytes) > self.page_size: return False self.records[record.id] = record_bytes self.current_size += len(record_bytes) return True 3. 向量索引算法 3.1 HNSW（Hierarchical Navigable Small World） HNSW 是当前最流行的向量索引算法，在召回率和延迟之间取得优秀平衡。 ...

1. 为什么 LLM 需要缓存 LLM 调用成本高、延迟高。以 GPT-4o 为例： 延迟：单次请求 500ms - 3000ms 成本：输入 $2.5/1M tokens，输出 $10/1M tokens 并发限制：RPM（每分钟请求数）和 TPM（每分钟 tokens）受限 缓存可以显著降低延迟和成本。但 LLM 缓存不同于传统缓存：用户问题"今天天气"和"天气怎么样"语义相近，应该命中同一缓存条目——这就是语义缓存的核心价值。 2. 语义缓存原理 2.1 相似度计算 from dataclasses import dataclass import numpy as np @dataclass class CacheEntry: query: str query_embedding: np.ndarray response: str model: str tokens_used: int timestamp: float hit_count: int = 0 ttl_seconds: int = 3600 class SemanticCache: """ 语义缓存：基于向量相似度的缓存命中 核心思想：如果新查询与已缓存查询的语义距离 < 阈值，则复用缓存结果 """ def __init__(self, embedding_model, similarity_threshold: float = 0.95): self.embedding_model = embedding_model self.threshold = similarity_threshold self.cache: list[CacheEntry] = [] self.index = None # 可选：FAISS/Annoy 加速 async def get(self, query: str) -> tuple[bool, str | None]: query_embedding = await self.embedding_model.embed(query) # 线性搜索（生产环境用向量索引） for entry in self.cache: similarity = self._cosine_similarity(query_embedding, entry.query_embedding) if similarity >= self.threshold: entry.hit_count += 1 return True, entry.response return False, None async def set(self, query: str, response: str, model: str, tokens: int): embedding = await self.embedding_model.embed(query) entry = CacheEntry( query=query, query_embedding=embedding, response=response, model=model, tokens_used=tokens, timestamp=time.time() ) self.cache.append(entry) def _cosine_similarity(self, v1: np.ndarray, v2: np.ndarray) -> float: return float(np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))) 2.2 阈值调优 import matplotlib.pyplot as plt class ThresholdOptimizer: """语义缓存阈值优化器""" def __init__(self, cache: SemanticCache, test_data: list[tuple[str, str, bool]]): """ test_data: [(query1, cached_query, should_hit), ...] """ self.cache = cache self.test_data = test_data def evaluate(self, threshold: float) -> dict: self.cache.threshold = threshold true_positives = 0 false_positives = 0 false_negatives = 0 true_negatives = 0 for query, cached, should_hit in self.test_data: # 手动计算相似度 e1 = await self.cache.embedding_model.embed(query) e2 = await self.cache.embedding_model.embed(cached) sim = self.cache._cosine_similarity(e1, e2) is_hit = sim >= threshold if should_hit and is_hit: true_positives += 1 elif not should_hit and is_hit: false_positives += 1 elif should_hit and not is_hit: false_negatives += 1 else: true_negatives += 1 precision = true_positives / (true_positives + false_positives) if (true_positives + false_positives) > 0 else 0 recall = true_positives / (true_positives + false_negatives) if (true_positives + false_negatives) > 0 else 0 f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0 return {"threshold": threshold, "precision": precision, "recall": recall, "f1": f1} def find_optimal_threshold(self) -> float: thresholds = np.arange(0.80, 1.0, 0.01) results = [self.evaluate(t) for t in thresholds] best = max(results, key=lambda x: x["f1"]) return best["threshold"] 推荐阈值： ...

1. 为什么需要 LLM 网关 当企业接入多个 LLM 供应商（OpenAI、Anthropic、Google、本地模型等），直接调用会面临： 供应商锁定：切换模型需要修改所有调用方代码 成本失控：无法统一监控和控制 token 消耗 可靠性差：单供应商宕机即服务不可用 安全风险：API Key 分散在各处，难以统一管理 LLM 网关作为统一接入层，解决以上所有问题。 2. 整体架构 ┌─────────────────────────────────────────────────┐ │ LLM Gateway │ │ │ │ ┌─────────┐ ┌──────────┐ ┌───────────────┐ │ │ │ Router │ │ Load │ │ Rate │ │ │ │ Engine │→ │ Balancer │→ │ Limiter │ │ │ └─────────┘ └──────────┘ └───────────────┘ │ │ │ │ │ ┌────┴────────────────────────────────────┐ │ │ │ Model Adapter Pool │ │ │ │ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ │ │ │ │ │OpenAI│ │Claude│ │Gemini│ │Local │ │ │ │ │ └──────┘ └──────┘ └──────┘ └──────┘ │ │ │ └─────────────────────────────────────────┘ │ │ │ │ ┌──────────┐ ┌──────────┐ ┌────────────┐ │ │ │ Cache │ │ Metrics │ │ Audit │ │ │ │ Layer │ │ Collector│ │ Logger │ │ │ └──────────┘ └──────────┘ └────────────┘ │ └─────────────────────────────────────────────────┘ 3. 多模型路由引擎 3.1 路由策略 from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Optional import hashlib import time @dataclass class LLMRequest: model: str messages: list[dict] temperature: float = 0.7 max_tokens: int = 2048 stream: bool = False metadata: dict = field(default_factory=dict) @dataclass class ModelEndpoint: provider: str # openai, anthropic, local model_id: str # gpt-4o, claude-3.5-sonnet endpoint_url: str api_key: str max_concurrent: int = 10 cost_per_1k_input: float = 0.0 cost_per_1k_output: float = 0.0 avg_latency_ms: float = 500 availability: float = 99.9 current_load: int = 0 class RoutingStrategy(ABC): @abstractmethod def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint: pass class CostOptimizedRouting(RoutingStrategy): """最低成本优先路由""" def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint: available = [c for c in candidates if c.current_load < c.max_concurrent] if not available: raise RuntimeError("No available endpoints") return min(available, key=lambda c: c.cost_per_1k_input) class LatencyOptimizedRouting(RoutingStrategy): """最低延迟优先路由""" def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint: available = [c for c in candidates if c.current_load < c.max_concurrent] if not available: raise RuntimeError("No available endpoints") return min(available, key=lambda c: c.avg_latency_ms) class WeightedRoundRobinRouting(RoutingStrategy): """加权轮询路由""" def __init__(self): self._index = 0 self._weights: dict[str, int] = {} def set_weight(self, model_id: str, weight: int): self._weights[model_id] = weight def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint: available = [c for c in candidates if c.current_load < c.max_concurrent] if not available: raise RuntimeError("No available endpoints") # 构建加权列表 weighted = [] for c in available: w = self._weights.get(c.model_id, 1) weighted.extend([c] * w) selected = weighted[self._index % len(weighted)] self._index += 1 return selected class CapabilityBasedRouting(RoutingStrategy): """基于任务能力的智能路由""" CAPABILITY_MAP = { "code_generation": ["gpt-4o", "claude-3.5-sonnet", "deepseek-coder"], "long_context": ["gemini-1.5-pro", "claude-3.5-sonnet"], "vision": ["gpt-4o", "gemini-1.5-pro"], "low_cost": ["gpt-4o-mini", "claude-3-haiku", "gemini-1.5-flash"], } def select(self, request: LLMRequest, candidates: list[ModelEndpoint]) -> ModelEndpoint: required_cap = request.metadata.get("capability") if not required_cap: return candidates[0] preferred_models = self.CAPABILITY_MAP.get(required_cap, []) for model_id in preferred_models: for c in candidates: if c.model_id == model_id and c.current_load < c.max_concurrent: return c # 降级：选任意可用的 return next((c for c in candidates if c.current_load < c.max_concurrent), candidates[0]) 3.2 路由策略对比 策略 优势 劣势 适用场景 成本优先 最低开销 可能高延迟 批量处理 延迟优先 响应快 成本高 实时对话 加权轮询 负载均衡 不考虑成本 通用场景 能力路由 任务匹配最优 需维护映射 混合任务 4. 限流与降级 4.1 多维度限流 from collections import defaultdict, deque import time class MultiDimensionRateLimiter: def __init__(self): # 按用户限流 self.user_limits: dict[str, tuple[int, float]] = {} # user -> (max_rpm, window_s) self.user_counters: dict[str, deque] = defaultdict(deque) # 按模型限流 self.model_limits: dict[str, tuple[int, float]] = {} self.model_counters: dict[str, deque] = defaultdict(deque) # 全局限流 self.global_limit = (1000, 60) # 1000 rpm self.global_counter: deque = deque() def check(self, user_id: str, model: str) -> bool: now = time.time() if not self._check_counter(self.global_counter, self.global_limit, now): return False if user_id in self.user_limits: if not self._check_counter(self.user_counters[user_id], self.user_limits[user_id], now): return False if model in self.model_limits: if not self._check_counter(self.model_counters[model], self.model_limits[model], now): return False return True def _check_counter(self, counter: deque, limit: tuple, now: float) -> bool: max_count, window = limit while counter and counter[0] < now - window: counter.popleft() if len(counter) >= max_count: return False counter.append(now) return True 4.2 熔断降级链 class FallbackChain: """模型降级链：主模型失败时自动切换到备选模型""" def __init__(self): self.chains: dict[str, list[str]] = { "gpt-4o": ["claude-3.5-sonnet", "gemini-1.5-pro", "gpt-4o-mini"], "claude-3.5-sonnet": ["gpt-4o", "gemini-1.5-pro"], } async def execute_with_fallback(self, gateway, request: LLMRequest) -> dict: primary = request.model chain = [primary] + self.chains.get(primary, []) last_error = None for model_id in chain: try: request.model = model_id result = await gateway.forward(request) return result except Exception as e: last_error = e continue raise last_error 5. 统一适配器层 class ModelAdapter(ABC): @abstractmethod async def chat(self, request: LLMRequest) -> dict: pass @abstractmethod async def stream_chat(self, request: LLMRequest): pass @abstractmethod def count_tokens(self, messages: list[dict]) -> int: pass class OpenAIAdapter(ModelAdapter): def __init__(self, endpoint: ModelEndpoint): self.endpoint = endpoint self.client = httpx.AsyncClient( base_url=endpoint.endpoint_url, headers={"Authorization": f"Bearer {endpoint.api_key}"}, timeout=120 ) async def chat(self, request: LLMRequest) -> dict: resp = await self.client.post("/v1/chat/completions", json={ "model": request.model, "messages": request.messages, "temperature": request.temperature, "max_tokens": request.max_tokens, }) resp.raise_for_status() data = resp.json() return { "content": data["choices"][0]["message"]["content"], "model": data["model"], "usage": data["usage"], } async def stream_chat(self, request: LLMRequest): async with self.client.stream("POST", "/v1/chat/completions", json={ "model": request.model, "messages": request.messages, "stream": True, }) as resp: async for line in resp.aiter_lines(): if line.startswith("data: "): yield line[6:] def count_tokens(self, messages: list[dict]) -> int: # tiktoken 计算 import tiktoken enc = tiktoken.encoding_for_model("gpt-4o") total = sum(len(enc.encode(m["content"])) for m in messages) return total class AnthropicAdapter(ModelAdapter): async def chat(self, request: LLMRequest) -> dict: # 适配 Anthropic API 格式 system = next((m["content"] for m in request.messages if m["role"] == "system"), "") user_messages = [m for m in request.messages if m["role"] != "system"] resp = await self.client.post("/v1/messages", json={ "model": request.model, "system": system, "messages": user_messages, "max_tokens": request.max_tokens, }) # 转换为统一格式 ... class LocalModelAdapter(ModelAdapter): """本地 vLLM/Ollama 适配器""" async def chat(self, request: LLMRequest) -> dict: resp = await self.client.post("/v1/chat/completions", json={ "model": request.model, "messages": request.messages, "temperature": request.temperature, }) ... 6. 成本控制与计量 @dataclass class UsageRecord: user_id: str model: str input_tokens: int output_tokens: int cost: float timestamp: float request_id: str class CostTracker: def __init__(self, redis_client): self.redis = redis_client async def record(self, record: UsageRecord): pipe = self.redis.pipeline() # 按用户累计 pipe.hincrby(f"cost:user:{record.user_id}:daily:{date.today()}", "total", int(record.cost * 10000)) pipe.hincrby(f"cost:user:{record.user_id}:monthly:{date.today().strftime('%Y-%m')}", "total", int(record.cost * 10000)) # 按模型累计 pipe.hincrby(f"cost:model:{record.model}:daily:{date.today()}", "total", int(record.cost * 10000)) # 预算检查 pipe.hget(f"budget:user:{record.user_id}:monthly", "limit") results = await pipe.execute() budget_limit = results[-1] if budget_limit and results[1] > int(budget_limit): raise BudgetExceededError(f"User {record.user_id} exceeded monthly budget") async def get_usage_report(self, user_id: str, period: str = "daily") -> dict: key = f"cost:user:{user_id}:{period}:{date.today()}" data = await self.redis.hgetall(key) return {"user": user_id, "period": period, "cost_cents": int(data.get("total", 0)) / 10000} 7. 可观测性 class GatewayMetrics: def __init__(self): self.request_count = Counter("gateway_requests_total", ["model", "status"]) self.latency = Histogram("gateway_latency_seconds", ["model"]) self.token_usage = Counter("gateway_tokens_total", ["model", "direction"]) self.cost = Counter("gateway_cost_total", ["model"]) self.active_connections = Gauge("gateway_active_connections") def record_request(self, model: str, status: str, duration: float, input_tokens: int, output_tokens: int, cost: float): self.request_count.labels(model=model, status=status).inc() self.latency.labels(model=model).observe(duration) self.token_usage.labels(model=model, direction="input").inc(input_tokens) self.token_usage.labels(model=model, direction="output").inc(output_tokens) self.cost.labels(model=model).inc(cost) 8. 部署架构 ┌─────────┐ │ CDN │ └────┬────┘ │ ┌──────────┼──────────┐ │ │ │ ┌────┴───┐ ┌───┴────┐ ┌──┴─────┐ │GW #1 │ │GW #2 │ │GW #3 │ │(active)│ │(active)│ │(active)│ └────┬───┘ └───┬────┘ └──┬─────┘ │ │ │ ┌────┴─────────┴─────────┴────┐ │ Redis Cluster │ │ (rate limits, cache, cost) │ └──────────────────────────────┘ │ │ ┌────┴────┐ ┌────┴────┐ │ Model A │ │ Model B │ │ Provider│ │ Provider│ └─────────┘ └─────────┘ 9. 总结 LLM 网关是企业级 AI 应用的必备基础设施。核心设计要点： ...

1. 引言：为什么需要多智能体编排 单个 LLM Agent 在复杂任务中面临上下文窗口限制、角色混淆、推理链断裂等问题。多智能体架构通过任务分解、角色专精和协作机制，将复杂问题分配给多个专业化 Agent 协同完成。然而，如何编排这些 Agent——谁来调度、如何通信、何时同步——是工程落地的核心挑战。 2. 三种核心编排模式 2.1 中心化编排（Orchestrator Pattern） 一个中央编排器（Orchestrator）负责任务分配、状态管理和结果汇总。所有 Agent 只与编排器通信，互不直接交互。 ┌──────────────────────────────────┐ │ Orchestrator │ │ ┌──────┐ ┌──────┐ ┌──────┐ │ │ │Worker│ │Worker│ │Worker│ │ │ │ A │ │ B │ │ C │ │ │ └──────┘ └──────┘ └──────┘ │ └──────────────────────────────────┘ 核心代码实现： from abc import ABC, abstractmethod from typing import Any, Optional import asyncio class Agent(ABC): def __init__(self, name: str, system_prompt: str): self.name = name self.system_prompt = system_prompt self.message_history: list[dict] = [] @abstractmethod async def execute(self, task: str, context: dict) -> str: pass class Orchestrator: def __init__(self): self.agents: dict[str, Agent] = {} self.task_queue: list[dict] = [] self.results: dict[str, Any] = {} def register(self, agent: Agent): self.agents[agent.name] = agent async def dispatch(self, agent_name: str, task: str, context: dict = None) -> str: agent = self.agents[agent_name] result = await agent.execute(task, context or {}) self.results[f"{agent_name}:{task[:20]}"] = result return result async def run_pipeline(self, plan: list[dict]) -> dict: """按计划顺序执行任务，支持依赖传递""" for step in plan: agent_name = step["agent"] task = step["task"] deps = step.get("depends_on", []) merged_context = {d: self.results.get(d) for d in deps} await self.dispatch(agent_name, task, merged_context) return self.results 适用场景： 工作流明确的任务（如代码审查流水线、文档生成管线） ...