实践指南

为什么需要 Prompt 工程化 在原型阶段，Prompt 通常是一个写在代码里的字符串常量。但当应用走向生产，问题开始浮现： 改了一个词，线上效果突然变差，却不知道回退到哪个版本 A/B 测试靠手动切换环境变量，数据散落在日志文件里 新来的同事改了 Prompt，破坏了之前精心设计的 Few-shot 格式 同一个功能有 5 个 Prompt 变体，没人知道哪个在跑 Prompt 工程化的核心目标：让 Prompt 成为可追踪、可测试、可回滚的一等公民。 Prompt 版本管理 目录结构设计 prompts/ ├── config.yaml # 全局配置 ├── chatbot/ # 功能模块 │ ├── meta.yaml # 模块元数据 │ ├── v1.0.0/ # 语义化版本 │ │ ├── system.txt # 系统提示 │ │ ├── few_shot.jsonl # Few-shot 示例 │ │ └── config.yaml # 模型参数 │ ├── v1.1.0/ │ │ ├── system.txt │ │ ├── few_shot.jsonl │ │ └── config.yaml │ └── v2.0.0/ # 大版本变更 │ ├── system.txt │ ├── few_shot.jsonl │ └── config.yaml └── classifier/ └── ... Prompt 注册中心实现 import yaml import json from pathlib import Path from dataclasses import dataclass, field from typing import Any @dataclass class PromptVersion: """单个 Prompt 版本""" name: str version: str system: str few_shot: list[dict] = field(default_factory=list) model: str = "gpt-4o-mini" temperature: float = 0.7 max_tokens: int = 1024 status: str = "active" # draft | testing | active | archived def render(self, user_input: str) -> list[dict]: """渲染为 API 消息格式""" messages = [{"role": "system", "content": self.system}] for example in self.few_shot: messages.append({"role": example["role"], "content": example["content"]}) messages.append({"role": "user", "content": user_input}) return messages class PromptRegistry: """Prompt 版本注册中心""" def __init__(self, base_dir: str = "prompts"): self.base_dir = Path(base_dir) self._cache: dict[str, PromptVersion] = {} def load(self, name: str, version: str = "latest") -> PromptVersion: """加载指定版本的 Prompt""" if version == "latest": version = self._get_latest_version(name) cache_key = f"{name}@{version}" if cache_key in self._cache: return self._cache[cache_key] version_dir = self.base_dir / name / f"v{version}" # 加载系统提示 system = (version_dir / "system.txt").read_text(encoding="utf-8") # 加载 Few-shot few_shot = [] few_shot_path = version_dir / "few_shot.jsonl" if few_shot_path.exists(): for line in few_shot_path.read_text(encoding="utf-8").strip().split("\n"): few_shot.append(json.loads(line)) # 加载配置 config_path = version_dir / "config.yaml" config = yaml.safe_load(config_path.read_text(encoding="utf-8")) pv = PromptVersion( name=name, version=version, system=system, few_shot=few_shot, model=config.get("model", "gpt-4o-mini"), temperature=config.get("temperature", 0.7), max_tokens=config.get("max_tokens", 1024), status=config.get("status", "active"), ) self._cache[cache_key] = pv return pv def _get_latest_version(self, name: str) -> str: """获取最新 active 版本""" module_dir = self.base_dir / name versions = [] for d in module_dir.iterdir(): if d.is_dir() and d.name.startswith("v"): config = yaml.safe_load((d / "config.yaml").read_text(encoding="utf-8")) if config.get("status") == "active": versions.append(d.name[1:]) # 去掉 'v' 前缀 versions.sort(key=lambda v: [int(x) for x in v.split(".")]) return versions[-1] if versions else "1.0.0" def diff(self, name: str, v1: str, v2: str) -> dict: """对比两个版本的差异""" p1 = self.load(name, v1) p2 = self.load(name, v2) return { "system_changed": p1.system != p2.system, "few_shot_changed": p1.few_shot != p2.few_shot, "model_changed": p1.model != p2.model, "temperature_changed": p1.temperature != p2.temperature, } # 使用示例 registry = PromptRegistry("prompts") prompt = registry.load("chatbot", "latest") messages = prompt.render("你好，帮我查一下订单") 版本管理规范 版本类型 变更内容 示例 Major (x.0.0) Prompt 结构重构、角色定义变更 从单轮改为多轮对话 Minor (1.x.0) Few-shot 增删、指令逻辑调整 新增 2 个示例 Patch (1.0.x) 文案微调、错别字修正 “请” → “请帮我” A/B 测试框架 import random import hashlib from dataclasses import dataclass, field from collections import defaultdict import time @dataclass class ABTestConfig: """A/B 测试配置""" test_name: str variants: dict[str, PromptVersion] # variant_name -> Prompt traffic_split: dict[str, float] # variant_name -> 流量比例 metrics: list[str] = field(default_factory=lambda: [ "user_satisfaction", "response_length", "latency_ms", "cost" ]) min_sample_size: int = 100 def assign(self, user_id: str) -> str: """基于用户 ID 确定性分配变体（同一用户始终进入同一组）""" hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) ratio = (hash_val % 10000) / 10000.0 cumulative = 0.0 for variant, weight in self.traffic_split.items(): cumulative += weight if ratio < cumulative: return variant return list(self.traffic_split.keys())[-1] @dataclass class ExperimentResult: variant: str user_id: str metric: str value: float timestamp: float = field(default_factory=time.time) class ABTestRunner: """A/B 测试运行器""" def __init__(self): self.results: list[ExperimentResult] = [] def run( self, test_config: ABTestConfig, user_id: str, user_input: str, execute_fn, # callable: (PromptVersion, str) -> dict ) -> dict: """执行一次 A/B 测试请求""" variant_name = test_config.assign(user_id) prompt = test_config.variants[variant_name] # 执行并收集指标 result = execute_fn(prompt, user_input) # 记录指标 for metric in test_config.metrics: if metric in result: self.results.append(ExperimentResult( variant=variant_name, user_id=user_id, metric=metric, value=result[metric], )) return {"variant": variant_name, "result": result} def analyze(self, test_name: str) -> dict: """分析实验结果""" stats = defaultdict(lambda: defaultdict(list)) for r in self.results: stats[r.variant][r.metric].append(r.value) report = {} for variant, metrics in stats.items(): report[variant] = {} for metric, values in metrics.items(): vals = sorted(values) report[variant][metric] = { "count": len(vals), "mean": sum(vals) / len(vals), "median": vals[len(vals) // 2], "p95": vals[int(len(vals) * 0.95)] if len(vals) > 20 else None, } return report def is_significant(self, test_name: str, metric: str, alpha: float = 0.05) -> bool: """简单的统计显著性检验（Z 检验）""" import math variants = [r.variant for r in self.results if r.metric == metric] if len(set(variants)) < 2: return False # 按 variant 分组 groups = defaultdict(list) for r in self.results: if r.metric == metric: groups[r.variant].append(r.value) if len(groups) < 2: return False v1, v2 = list(groups.keys())[:2] s1, s2 = groups[v1], groups[v2] if len(s1) < 30 or len(s2) < 30: return False # 样本不足 m1, m2 = sum(s1) / len(s1), sum(s2) / len(s2) var1 = sum((x - m1) ** 2 for x in s1) / len(s1) var2 = sum((x - m2) ** 2 for x in s2) / len(s2) se = math.sqrt(var1 / len(s1) + var2 / len(s2)) if se == 0: return False z = abs(m1 - m2) / se return z > 1.96 # 95% 置信度 # 使用示例 runner = ABTestRunner() test_config = ABTestConfig( test_name="chatbot_tone_v2", variants={ "control": registry.load("chatbot", "1.0.0"), "treatment": registry.load("chatbot", "1.1.0"), }, traffic_split={"control": 0.5, "treatment": 0.5}, ) def execute_fn(prompt: PromptVersion, user_input: str) -> dict: # 实际调用 LLM return { "user_satisfaction": 4.5, # 用户评分 "response_length": 320, "latency_ms": 850, "cost": 0.003, } # 模拟 200 次请求 for i in range(200): runner.run(test_config, f"user-{i}", "帮我查订单", execute_fn) # 分析结果 report = runner.analyze("chatbot_tone_v2") for variant, metrics in report.items(): print(f"\n=== {variant} ===") for metric, stats in metrics.items(): print(f" {metric}: mean={stats['mean']:.2f}, p95={stats['p95']}") print(f"\n统计显著: {runner.is_significant('chatbot_tone_v2', 'user_satisfaction')}") 回归评测体系 每次 Prompt 变更前，必须通过回归测试集的验证。 ...

引言 RAG（Retrieval-Augmented Generation）是当前最流行的 LLM 应用架构之一。然而，从 Demo 到生产之间横亘着巨大的鸿沟。本文基于多个 RAG 生产项目的实战经验，总结 12 个最常见、最致命的坑，并给出经过验证的解决方案。 坑 1：文档分块策略不当 问题 天真地按固定长度分块（如每 512 字符），导致： 语义被截断（一个完整的段落从中间切断） 关键信息分散在多个块中，检索时只命中一部分 表格和列表被拆碎，失去结构信息 解决方案：语义分块 + 重叠窗口 from dataclasses import dataclass from typing import List import re @dataclass class Chunk: text: str metadata: dict token_count: int = 0 class SemanticChunker: """基于语义边界的智能分块器""" def __init__( self, target_size: int = 400, # 目标块大小（tokens） min_size: int = 100, # 最小块大小 max_size: int = 600, # 最大块大小 overlap: int = 50, # 重叠区间 ): self.target_size = target_size self.min_size = min_size self.max_size = max_size self.overlap = overlap def chunk_document(self, text: str, source: str = "") -> List[Chunk]: """分块主流程""" # Step 1: 按结构边界切分 sections = self._split_by_structure(text) # Step 2: 对每个 section 按 paragraph 切分 paragraphs = [] for section in sections: paragraphs.extend(self._split_by_paragraph(section)) # Step 3: 合并过小的段落，拆分过大的段落 chunks = self._merge_and_split(paragraphs) # Step 4: 添加重叠 chunks = self._add_overlap(chunks) # Step 5: 附加元数据 return [ Chunk( text=c, metadata={"source": source, "chunk_index": i, "total_chunks": len(chunks)}, token_count=len(c) // 2, # 粗略估算 ) for i, c in enumerate(chunks) ] def _split_by_structure(self, text: str) -> List[str]: """按标题、分隔符等结构边界切分""" # 按 Markdown 标题切分 pattern = r'(?=^#{1,6}\s)' sections = re.split(pattern, text, flags=re.MULTILINE) return [s.strip() for s in sections if s.strip()] def _split_by_paragraph(self, text: str) -> List[str]: """按段落（双换行）切分""" paras = text.split("\n\n") return [p.strip() for p in paras if p.strip()] def _merge_and_split(self, paragraphs: List[str]) -> List[str]: """合并过小段落，拆分过大段落""" chunks = [] buffer = [] buffer_size = 0 for para in paragraphs: para_size = len(para) // 2 # 粗略 token 估算 if buffer_size + para_size > self.max_size and buffer: chunks.append("\n\n".join(buffer)) buffer = [] buffer_size = 0 if para_size > self.max_size: # 单段落过长，按句子切分 if buffer: chunks.append("\n\n".join(buffer)) buffer = [] buffer_size = 0 sentences = re.split(r'(?<=[。！？.!?])\s+', para) sent_buffer = [] sent_size = 0 for sent in sentences: if sent_size + len(sent) // 2 > self.max_size and sent_buffer: chunks.append(" ".join(sent_buffer)) sent_buffer = [] sent_size = 0 sent_buffer.append(sent) sent_size += len(sent) // 2 if sent_buffer: chunks.append(" ".join(sent_buffer)) else: buffer.append(para) buffer_size += para_size if buffer_size >= self.target_size: chunks.append("\n\n".join(buffer)) buffer = [] buffer_size = 0 if buffer: chunks.append("\n\n".join(buffer)) return chunks def _add_overlap(self, chunks: List[str]) -> List[str]: """为相邻块添加重叠""" if self.overlap <= 0 or len(chunks) <= 1: return chunks result = [chunks[0]] for i in range(1, len(chunks)): prev_text = chunks[i - 1] overlap_text = prev_text[-self.overlap * 2:] # 粗略取后半段 result.append(overlap_text + " " + chunks[i]) return result # 使用示例 chunker = SemanticChunker(target_size=400, overlap=50) document = open("knowledge_base/product_manual.md").read() chunks = chunker.chunk_document(document, source="product_manual.md") print(f"分块完成: {len(chunks)} 个块") 分块策略对比 策略 优点 缺点 适用场景 固定长度 实现简单 语义截断 不推荐 按段落 保持语义 块大小不均 短文档 语义分块 语义完整 实现复杂 通用推荐 按文档结构 保持层级 需要结构化输入 Markdown/HTML 递归分块 灵活适配 可控性差 混合内容 坑 2：Embedding 模型与 LLM 不匹配 问题 用 OpenAI 的 text-embedding-3-large 做向量，但生成用的是 Claude 模型。两者对语义的理解不同，可能导致检索到的内容并非生成模型"认为"最相关的。 ...

引言 “我们应该微调还是用 RAG？"——这是 2024-2026 年 AI 工程师被问得最多的问题。两者都能让 LLM 适配特定领域，但适用场景截然不同。本文不给标准答案，而是给一个决策框架。 核心差异：先理解本质 微调（Fine-tuning） 微调是改变模型内部权重，让模型"学会"新的知识或行为模式。 原始模型 → 注入领域数据 → 新权重 → 直接推理 RAG（检索增强生成） RAG是不改变模型，在推理时"查阅"外部知识库。 用户问题 → 检索知识库 → 拼接到 Prompt → 模型生成答案 本质对比 维度 微调 RAG 知识存储位置 模型权重中 外部数据库中 知识更新方式 重新训练 更新数据库 推理时依赖 无外部依赖 需要检索系统 “记忆"方式 隐式（权重） 显式（文本） 幻觉风险 较高（知识编码不精确） 较低（有原文参考） 知识溯源 无法溯源 可追溯到来源文档 决策框架：7 个关键问题 问题 1：知识更新频率多高？ 更新频率 │ ├─ 每日/每周更新 ──────→ RAG ✅ │ （产品目录、新闻、工单） │ ├─ 每月/每季度更新 ─────→ RAG 或混合 │ （技术文档、政策） │ └─ 几乎不变 ───────────→ 微调 ✅ （行业术语、品牌语调） 原理：微调每次更新知识都需要重新训练（数小时到数天），RAG 只需更新数据库（秒级）。 ...

为什么 Agent 需要记忆 没有记忆的 Agent 就像金鱼——每次对话都从零开始。你告诉它你的名字，下一轮它就忘了。你纠正过它的错误，它下次照犯。 记忆系统让 Agent 能够： 记住用户偏好和历史交互 从过去经验中学习 保持跨会话的上下文一致性 在长任务中不丢失关键信息 记忆类型体系 人类认知科学启发 记忆类型 持续时间 容量 Agent 对应 存储方案 工作记忆 秒-分钟 4±2 项 当前对话上下文 LLM Context Window 短期记忆 分钟-小时 有限 当前会话摘要 内存/Redis 长期记忆 天-永久 近无限 用户偏好、知识 向量库 + 关系库 情景记忆 事件级 大 具体交互记录 时序向量库 架构总览 ┌──────────────────────────────────────────────┐ │ Agent 主循环 │ │ │ │ ┌─────────┐ ┌─────────┐ ┌──────────────┐ │ │ │ 工作记忆 │ │ 短期记忆 │ │ 长期记忆 │ │ │ │ Context │←→│ Session │←→│ Persistent │ │ │ │ Window │ │ Summary │ │ Store │ │ │ └─────────┘ └─────────┘ └──────────────┘ │ │ │ │ │ │ │ │ │ ┌────────┼────────┐ │ │ │ │ │ │ │ │ │ │ │ ┌──┴──┐ ┌───┴───┐ ┌──┴──┐ │ │ │ │语义 │ │情景 │ │程序 │ │ │ │ │记忆 │ │记忆 │ │记忆 │ │ │ │ │向量 │ │时序 │ │图 │ │ │ │ └─────┘ └───────┘ └─────┘ │ │ │ │ ┌────┴────────────┴────────────────────────┐│ │ │ 记忆管理器 ││ │ │ 写入 → 压缩 → 检索 → 遗忘 → 巩固 ││ │ └────────────────────────────────────────────┘│ └──────────────────────────────────────────────┘ 工作记忆：上下文窗口管理 from dataclasses import dataclass, field from typing import List, Dict @dataclass class WorkingMemory: """工作记忆: 管理 LLM 上下文窗口""" max_tokens: int = 128000 reserved_for_response: int = 4096 messages: List[dict] = field(default_factory=list) def add(self, role: str, content: str): self.messages.append({"role": role, "content": content}) self._evict_if_needed() def add_system(self, content: str): """系统消息放在最前面，不被驱逐""" self.messages.insert(0, {"role": "system", "content": content}) def _evict_if_needed(self): """当超过 token 限制时，驱逐最旧的对话""" while self._count_tokens() > self.max_tokens - self.reserved_for_response: if len(self.messages) <= 1: break # 找到第一个非 system 消息并移除 for i, msg in enumerate(self.messages): if msg["role"] != "system": evicted = self.messages.pop(i) # 触发压缩: 被驱逐的消息需要被压缩到短期记忆 self._on_evict(evicted) break def _count_tokens(self) -> int: import tiktoken enc = tiktoken.encoding_for_model("gpt-4o") return sum(len(enc.encode(m["content"])) for m in self.messages) def _on_evict(self, evicted_msg: dict): """被驱逐的消息回调，交给短期记忆处理""" # 由 MemoryManager 注册 pass def get_context(self) -> List[dict]: """获取当前上下文""" return self.messages.copy() 短期记忆：会话级摘要 class ShortTermMemory: """短期记忆: 当前会话的压缩摘要""" def __init__(self): self.summary = "" self.key_points: List[str] = [] self.token_budget = 2000 # 摘要不超 2000 token async def compress(self, messages: List[dict]) -> str: """将多轮对话压缩为摘要""" conversation = "\n".join( f"{m['role']}: {m['content'][:200]}" for m in messages ) prompt = f"""请将以下对话压缩为简洁摘要，保留: 1. 用户的核心需求 2. 已确定的关键事实 3. 未解决的问题 4. 用户的明确偏好 对话: {conversation} 摘要:""" response = await llm_call(prompt, max_tokens=self.token_budget) self.summary = response return response async def extract_key_points(self, messages: List[dict]) -> List[str]: """提取关键信息点""" prompt = f"""从以下对话中提取关键信息点，每点一行: {chr(10).join(m['content'][:200] for m in messages)} 关键信息:""" response = await llm_call(prompt, max_tokens=500) self.key_points = response.strip().split('\n') return self.key_points def to_context(self) -> str: """转换为可注入上下文的文本""" parts = [] if self.summary: parts.append(f"之前的对话摘要:\n{self.summary}") if self.key_points: parts.append(f"关键信息:\n" + "\n".join(f"- {p}" for p in self.key_points)) return "\n\n".join(parts) 长期记忆：持久化存储 语义记忆（向量库） class SemanticMemory: """语义记忆: 存储事实和知识，向量检索""" def __init__(self, vector_store, embed_model): self.store = vector_store self.embed = embed_model async def remember(self, content: str, metadata: dict = None): """记住一条信息""" embedding = await self.embed(content) await self.store.upsert({ "id": self._generate_id(content), "values": embedding, "metadata": { "content": content, "type": "semantic", "created_at": datetime.now().isoformat(), **(metadata or {}) } }) async def recall(self, query: str, top_k: int = 5) -> List[dict]: """检索相关记忆""" query_embedding = await self.embed(query) results = await self.store.query( vector=query_embedding, top_k=top_k, filter={"type": {"$eq": "semantic"}} ) return [ { "content": r.metadata["content"], "score": r.score, "metadata": r.metadata } for r in results.matches ] async def forget(self, content_id: str): """主动遗忘""" await self.store.delete(ids=[content_id]) 情景记忆（时序事件） class EpisodicMemory: """情景记忆: 记录具体交互事件，支持时间检索""" def __init__(self, db): self.db = db # PostgreSQL + pgvector async def record(self, event: dict): """记录一个交互事件""" await self.db.execute( """ INSERT INTO episodic_memory (event_id, user_id, timestamp, input, output, outcome, context_snapshot, embedding) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) """, event["id"], event["user_id"], datetime.now(), event["input"], event["output"], event["outcome"], json.dumps(event.get("context", {})), event.get("embedding") ) async def recall_by_time(self, user_id: str, start: datetime, end: datetime) -> List[dict]: """按时间范围回忆""" rows = await self.db.fetch( """SELECT * FROM episodic_memory WHERE user_id = $1 AND timestamp BETWEEN $2 AND $3 ORDER BY timestamp DESC""", user_id, start, end ) return [dict(r) for r in rows] async def recall_similar(self, user_id: str, query_embedding: list, top_k: int = 5) -> List[dict]: """检索相似的历史事件""" rows = await self.db.fetch( """SELECT *, embedding <=> $1 as distance FROM episodic_memory WHERE user_id = $2 ORDER BY embedding <=> $1 LIMIT $3""", query_embedding, user_id, top_k ) return [dict(r) for r in rows] 程序记忆（知识图谱） class ProceduralMemory: """程序记忆: 存储技能、流程、因果关系（图结构）""" def __init__(self, graph_db): self.graph = graph_db # Neo4j async def learn_procedure(self, name: str, steps: List[str], triggers: List[str]): """学习一个新流程""" await self.graph.execute( """ CREATE (p:Procedure {name: $name}) WITH p UNWIND $steps as step_text CREATE (s:Step {text: step_text}) CREATE (p)-[:HAS_STEP]->(s) WITH p UNWIND $triggers as trigger CREATE (t:Trigger {text: trigger}) CREATE (t)-[:ACTIVATES]->(p) """, name=name, steps=steps, triggers=triggers ) async def recall_procedure(self, context: str) -> List[dict]: """根据上下文检索相关流程""" result = await self.graph.execute( """ MATCH (t:Trigger)-[:ACTIVATES]->(p:Procedure)-[:HAS_STEP]->(s:Step) WHERE $context CONTAINS t.text RETURN p.name as procedure, collect(s.text) as steps """, context=context ) return [dict(r) for r in result] 记忆管理器：统一调度 class MemoryManager: """统一记忆管理器""" def __init__(self): self.working = WorkingMemory() self.short_term = ShortTermMemory() self.semantic = SemanticMemory(...) self.episodic = EpisodicMemory(...) self.procedural = ProceduralMemory(...) self.recall_count = 0 async def think(self, user_input: str, user_id: str) -> str: """带记忆的思考过程""" # 1. 从长期记忆检索相关信息 semantic_results = await self.semantic.recall(user_input, top_k=3) episodic_results = await self.episodic.recall_similar( user_id, await self.embed(user_input), top_k=3 ) procedures = await self.procedural.recall_procedure(user_input) # 2. 组装上下文 context_parts = [] if self.short_term.summary: context_parts.append(self.short_term.to_context()) if semantic_results: context_parts.append("相关知识:\n" + "\n".join(f"- {r['content']}" for r in semantic_results)) if episodic_results: context_parts.append("历史交互:\n" + "\n".join(f"- {r['input']} → {r['output']}" for r in episodic_results)) if procedures: context_parts.append("可用流程:\n" + "\n".join(f"- {p['procedure']}: {p['steps']}" for p in procedures)) context = "\n\n".join(context_parts) # 3. 注入工作记忆 if context: self.working.add("system", f"记忆上下文:\n{context[:4000]}") self.working.add("user", user_input) # 4. 生成回复 response = await llm_call(self.working.get_context()) self.working.add("assistant", response) # 5. 异步写入记忆 asyncio.create_task(self._consolidate(user_input, response, user_id)) return response async def _consolidate(self, input: str, output: str, user_id: str): """记忆巩固: 筛选重要信息写入长期记忆""" # 判断是否值得记住 importance = await self._assess_importance(input, output) if importance > 0.7: # 高重要性: 写入语义记忆 + 情景记忆 await self.semantic.remember( f"用户说了: {input}\n助手回答: {output}", metadata={"importance": importance, "user_id": user_id} ) # 总是记录情景记忆 await self.episodic.record({ "id": str(uuid4()), "user_id": user_id, "input": input, "output": output, "outcome": "success", "embedding": await self.embed(input), }) async def _assess_importance(self, input: str, output: str) -> float: """评估信息重要性 0-1""" prompt = f"""评估以下交互的信息重要性(0-1): 输入: {input[:200]} 输出: {output[:200]} 评分标准: - 0.9+: 用户明确要求记住/涉及核心偏好 - 0.7-0.8: 包含重要事实或决策 - 0.5-0.6: 有一定参考价值 - 0.3-0.4: 闲聊但含少量信息 - 0.0-0.2: 纯闲聊无信息 只输出数字:""" return float(await llm_call(prompt, max_tokens=10)) 遗忘机制 class ForgettingMechanism: """遗忘机制: 防止记忆膨胀""" async def decay(self, user_id: str): """基于时间的记忆衰减""" await self.db.execute( """ UPDATE semantic_memory SET importance = importance * 0.95 -- 每天衰减5% WHERE user_id = $1 AND last_accessed < NOW() - INTERVAL '1 day' """, user_id ) # 删除低重要性记忆 await self.db.execute( """ DELETE FROM semantic_memory WHERE user_id = $1 AND importance < 0.1 """, user_id ) async def deduplicate(self, user_id: str): """去重: 合并语义相似的记忆""" # 找到相似度 > 0.9 的记忆对 duplicates = await self.db.fetch( """ SELECT a.id as id_a, b.id as id_b, a.content as content_a, b.content as content_b FROM semantic_memory a JOIN semantic_memory b ON a.id < b.id WHERE a.user_id = $1 AND b.user_id = $1 AND a.embedding <=> b.embedding < 0.1 """, user_id ) for dup in duplicates: # 合并: 保留更详细的，删除另一个 longer = dup['content_a'] if len(dup['content_a']) > len(dup['content_b']) else dup['content_b'] shorter_id = dup['id_b'] if len(dup['content_a']) > len(dup['content_b']) else dup['id_a'] await self.db.execute("DELETE FROM semantic_memory WHERE id = $1", shorter_id) async def compress_old_memories(self, user_id: str, days: int = 30): """压缩旧记忆: 把多条相关记忆合并为一条摘要""" old_memories = await self.db.fetch( """ SELECT content FROM semantic_memory WHERE user_id = $1 AND created_at < NOW() - INTERVAL '$2 days' ORDER BY created_at """, user_id, days ) if len(old_memories) < 10: return # 用 LLM 合并 memories_text = "\n".join(m['content'] for m in old_memories) summary = await llm_call( f"将以下记忆合并为简洁摘要，保留关键信息:\n{memories_text}" ) # 删除旧记忆，写入摘要 await self.db.execute( "DELETE FROM semantic_memory WHERE user_id = $1 AND created_at < NOW() - INTERVAL '$2 days'", user_id, days ) await self.semantic.remember(summary, metadata={"compressed": True}) MemGPT / Letta 架构参考 MemGPT (现 Letta) 的核心思想是把操作系统的虚拟内存概念引入 LLM： ...

Agent 模式全景 复杂度 → 单Agent ──→ Router ──→ Fan-out/Fan-in ──→ Hierarchical ──→ Multi-Agent │ │ │ │ │ │ 简单任务 │ 按需分发 │ 并行处理 │ 管理者+执行者 │ 自主协作 │ │ │ │ │ └────────────┴──────────────┴──────────────────┴────────────────┘ 模式 适用场景 复杂度 延迟 错误恢复 单 Agent 简单问答、单一任务 低 低 简单重试 Router 多领域问答、任务分类 中 低+路由 单分支重试 Fan-out/Fan-in 并行研究、批量处理 中 高(并行) 部分失败容忍 Hierarchical 复杂项目、多步骤流程 高 高 子任务级恢复 Multi-Agent 开放式协作、辩论 极高 最高 最复杂 单 Agent 架构 最基础的形态：一个 Agent + 一组工具。 ...

成本结构分析 在深入优化之前，先搞清楚钱花在哪里。一个典型的 RAG 应用每次调用的成本构成： 成本项 占比 计算方式 可优化空间 LLM 生成 Token 50-65% output_tokens × price_per_1k 高 LLM 输入 Token 20-35% input_tokens × price_per_1k 高 Embedding 5-10% chunks × chunk_size × embed_price 中 向量检索 2-5% 按次/按量计费 低 基础设施 5-15% GPU/CPU/内存/带宽 中 关键洞察：输出 Token 通常比输入 Token 贵 3-5 倍，优化输出端收益更大。 Prompt 压缩 1. 去除冗余信息 # 冗长 Prompt（382 tokens） prompt = """ 你是一个专业的客服助手，你的主要职责是回答用户关于产品的问题。 你需要始终保持礼貌和专业，回答要简洁明了。 如果用户的问题超出了你的知识范围，请诚实告知用户你会将问题转接给人工客服。 在回答之前，请先理解用户的核心诉求，然后给出针对性的回答。 请使用以下格式回答： 1. 直接回答问题 2. 如果需要补充信息，请提供 3. 如果无法回答，请说明原因 """ # 精简 Prompt（89 tokens） prompt = "客服助手。简洁专业。不知则转人工。格式：直接回答→补充→说明原因。" 2. Context Compression 对长文档先做摘要再喂给大模型： ...

监控分层架构 AI 应用监控需要三层视角，每层关注不同问题： ┌─────────────────────────────────────────────────┐ │ Layer 3: 模型质量监控 │ │ - 回答正确率 / 幻觉率 / 毒性检测 / 偏见 │ │ - 用户满意度反馈 │ ├─────────────────────────────────────────────────┤ │ Layer 2: 应用性能监控 │ │ - 首字延迟 (TTFT) / 完整延迟 / 吞吐量 │ │ - Token 消耗 / 成本 / 缓存命中率 │ ├─────────────────────────────────────────────────┤ │ Layer 1: 基础设施监控 │ │ - GPU 利用率 / 内存 / 网络 / 磁盘 │ │ - API 错误率 / 限流次数 / 超时 │ └─────────────────────────────────────────────────┘ 指标设计 核心指标清单 层级 指标 类型 告警阈值 模型 回答质量评分 Gauge < 0.8 模型 幻觉率 Gauge > 5% 模型 用户点赞率 Gauge < 70% 应用 TTFT (首字延迟) Histogram P99 > 2s 应用 完整延迟 Histogram P99 > 15s 应用 Token 消耗 Counter 日预算 120% 应用 缓存命中率 Gauge < 15% 基础设施 GPU 利用率 Gauge > 90% 基础设施 API 错误率 Gauge > 1% 基础设施 限流次数 Counter > 100/min Prometheus 指标定义 from prometheus_client import Counter, Histogram, Gauge, Summary # === 模型质量指标 === llm_quality_score = Gauge( "llm_quality_score", "Average answer quality score (0-1)", ["model", "task_type"] ) llm_hallucination_rate = Gauge( "llm_hallucination_rate", "Hallucination rate detected", ["model"] ) llm_user_feedback = Counter( "llm_user_feedback_total", "User feedback counts", ["model", "feedback"] # feedback: positive/negative ) # === 应用性能指标 === llm_ttft = Histogram( "llm_time_to_first_token_seconds", "Time to first token in seconds", ["model"], buckets=[0.1, 0.3, 0.5, 1, 2, 5, 10] ) llm_total_latency = Histogram( "llm_total_latency_seconds", "Total request latency in seconds", ["model"], buckets=[0.5, 1, 2, 5, 10, 20, 30, 60] ) llm_tokens_total = Counter( "llm_tokens_total", "Total tokens consumed", ["model", "direction"] # direction: input/output ) llm_cost_total = Counter( "llm_cost_usd_total", "Total cost in USD", ["model"] ) llm_cache_hit_rate = Gauge( "llm_cache_hit_rate", "Cache hit rate", ["cache_type"] # cache_type: exact/semantic ) # === 基础设施指标 === gpu_utilization = Gauge( "gpu_utilization_percent", "GPU utilization percentage", ["gpu_id", "node"] ) api_error_rate = Gauge( "llm_api_error_rate", "LLM API error rate", ["provider", "error_type"] ) 指标采集 中间件式采集 import time from contextlib import asynccontextmanager class LLMInstrumentation: def __init__(self, registry): self.registry = registry @asynccontextmanager async def trace_request(self, model: str, task_type: str): request_id = generate_id() start = time.monotonic() first_token_time = None async def on_first_token(): nonlocal first_token_time first_token_time = time.monotonic() try: yield {"request_id": request_id, "on_first_token": on_first_token} # 请求成功 elapsed = time.monotonic() - start llm_total_latency.labels(model=model).observe(elapsed) if first_token_time: ttft = first_token_time - start llm_ttft.labels(model=model).observe(ttft) except Exception as e: api_error_rate.labels( provider=model, error_type=type(e).__name__ ).inc() raise def record_tokens(self, model, input_tokens, output_tokens): llm_tokens_total.labels( model=model, direction="input" ).inc(input_tokens) llm_tokens_total.labels( model=model, direction="output" ).inc(output_tokens) # 成本计算 cost = self._calc_cost(model, input_tokens, output_tokens) llm_cost_total.labels(model=model).inc(cost) def _calc_cost(self, model, in_tok, out_tok): prices = { "gpt-4o": (0.0025, 0.01), "gpt-4o-mini": (0.00015, 0.0006), "claude-sonnet": (0.003, 0.015), } in_price, out_price = prices.get(model, (0, 0)) return (in_tok * in_price + out_tok * out_price) / 1000 使用方式 instrument = LLMInstrumentation(registry) async def chat_completion(messages, model="gpt-4o"): async with instrument.trace_request(model, "chat") as ctx: response = await llm_client.chat.completions.create( model=model, messages=messages, stream=True, ) async for chunk in response: if chunk.choices[0].delta.content: if not ctx["first_token_called"]: await ctx["on_first_token"]() ctx["first_token_called"] = True yield chunk.choices[0].delta.content instrument.record_tokens(model, input_count, output_count) 质量监控：LLM-as-a-Judge class QualityMonitor: def __init__(self, judge_model="gpt-4o", sample_rate=0.1): self.judge_model = judge_model self.sample_rate = sample_rate async def maybe_evaluate(self, query, response, context=None): """按采样率随机评估回答质量""" import random if random.random() > self.sample_rate: return score, issues = await self._judge(query, response, context) llm_quality_score.labels( model=self.judge_model, task_type="auto" ).set(score) if "hallucination" in issues: llm_hallucination_rate.labels( model=self.judge_model ).inc() async def _judge(self, query, response, context): prompt = f"""评估以下回答的质量。 问题：{query} 回答：{response} 参考资料：{context or '无'} 请检查： 1. 事实准确性（是否有幻觉） 2. 回答完整性 3. 逻辑一致性 输出 JSON：{{"score": 0.0-1.0, "issues": ["..."]}}""" result = await llm_client.chat.completions.create( model=self.judge_model, messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"}, ) data = json.loads(result.choices[0].message.content) return data["score"], data.get("issues", []) Grafana 仪表板 关键 Panel 配置 Panel 1: 延迟分布（P50/P90/P99） ...

Demo 与生产的差距 Function Calling 的 demo 很简单：定义一个函数，LLM 返回参数，调用函数，返回结果。但在生产中，你需要处理： LLM 返回了不存在的函数 参数类型不对、必填字段缺失 函数调用超时 并行调用之间的依赖冲突 用户通过参数注入恶意命令 调用链过深导致上下文爆炸 Schema 设计：函数定义是契约 好的 Schema 长什么样 # ❌ 糟糕的 Schema: 模糊、无约束 BAD_SCHEMA = { "name": "search", "description": "搜索东西", "parameters": { "type": "object", "properties": { "query": {"type": "string"} } } } # ✅ 好的 Schema: 精确、有约束、有枚举 GOOD_SCHEMA = { "name": "search_product", "description": "在商品库中搜索商品。当用户询问有没有某种商品、价格、库存时使用此函数。", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词，从用户意图中提取，去除无意义词" }, "category": { "type": "string", "enum": ["electronics", "clothing", "food", "books", "other"], "description": "商品类别" }, "max_price": { "type": "number", "minimum": 0, "description": "用户能接受的最高价格(元)" }, "sort_by": { "type": "string", "enum": ["price_asc", "price_desc", "relevance", "sales"], "default": "relevance" } }, "required": ["query"], "additionalProperties": False } } Schema 设计原则 原则 说明 示例 名字即文档 函数名要自解释 search_product > search description 要写触发条件 不是描述函数做什么，而是何时用 “当用户询问商品价格时使用” 枚举优于自由文本 能枚举就不要用 string category 用 enum 设默认值 减少LLM猜测 sort_by 默认 relevance 禁止额外属性 additionalProperties: false 防止 LLM 编造参数 范围约束 用 min/max 限制数值 max_price minimum: 0 参数验证：不要信任 LLM 的输出 import json from pydantic import BaseModel, Field, ValidationError from typing import Optional, Literal # 用 Pydantic 做二次验证 class SearchProductParams(BaseModel): query: str = Field(..., min_length=1, max_length=100) category: Optional[Literal["electronics", "clothing", "food", "books", "other"]] = None max_price: Optional[float] = Field(None, ge=0, le=1000000) sort_by: Literal["price_asc", "price_desc", "relevance", "sales"] = "relevance" class FunctionCallValidator: def __init__(self): self.schemas = { "search_product": SearchProductParams, } def validate(self, function_name: str, arguments: dict) -> tuple[bool, any]: schema_cls = self.schemas.get(function_name) if not schema_cls: return False, {"error": f"Unknown function: {function_name}"} try: validated = schema_cls(**arguments) return True, validated except ValidationError as e: return False, {"error": e.errors()} 处理 LLM 返回的常见问题 class RobustFunctionExecutor: async def execute(self, llm_response: dict) -> dict: tool_calls = llm_response.get("tool_calls", []) results = [] for call in tool_calls: name = call["function"]["name"] # 1. 函数是否存在? if name not in self.registry: results.append(self._error_result(call, f"未知函数: {name}")) continue # 2. 参数解析 try: args = json.loads(call["function"]["arguments"]) except json.JSONDecodeError: # LLM 返回了非法 JSON，尝试修复 args = self._repair_json(call["function"]["arguments"]) if args is None: results.append(self._error_result(call, "参数 JSON 解析失败")) continue # 3. 参数验证 ok, validated = self.validator.validate(name, args) if not ok: results.append(self._error_result(call, str(validated))) continue # 4. 执行(带超时) try: result = await asyncio.wait_for( self.registry[name](**validated.dict()), timeout=10 ) results.append({"tool_call_id": call["id"], "result": result}) except asyncio.TimeoutError: results.append(self._error_result(call, "函数执行超时")) except Exception as e: results.append(self._error_result(call, f"执行错误: {str(e)}")) return results def _repair_json(self, broken: str) -> dict | None: """尝试修复 LLM 输出的破损 JSON""" # 去除尾部逗号 fixed = broken.rstrip().rstrip(",") # 补全括号 open_braces = fixed.count("{") - fixed.count("}") open_brackets = fixed.count("[") - fixed.count("]") fixed += "}" * open_braces + "]" * open_brackets try: return json.loads(fixed) except: return None 错误恢复：让 LLM 从错误中学习 async def function_call_loop(messages: list, max_rounds: int = 5): """多轮函数调用循环，带错误恢复""" for round_num in range(max_rounds): response = await client.chat.completions.create( model="gpt-4o-mini", messages=messages, tools=tool_definitions, tool_choice="auto" ) msg = response.choices[0].message messages.append(msg) if not msg.tool_calls: return msg.content # LLM 认为完成了 # 执行所有工具调用 for tool_call in msg.tool_calls: success, result = await executor.execute_single(tool_call) # 把结果（包括错误）反馈给 LLM messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False) }) if not success: # 在错误信息中给出修复建议 messages.append({ "role": "system", "content": f"函数 {tool_call.function.name} 调用失败: {result['error']}。请修正参数后重试，或换一种方式回答。" }) return "达到最大调用轮数，请简化请求。" 并行调用与依赖管理 import asyncio from typing import Dict, List, Set class ParallelCallOrchestrator: def __init__(self): # 声明函数间的依赖关系 self.dependencies = { "get_order_detail": [], # 无依赖 "check_inventory": [], # 无依赖 "calculate_shipping": ["check_inventory"], # 依赖库存检查 "create_order": ["get_order_detail", "check_inventory"], } async def execute_parallel(self, tool_calls: list) -> list: # 构建依赖图 independent = [] dependent = {} for call in tool_calls: name = call["function"]["name"] deps = self.dependencies.get(name, []) if not deps: independent.append(call) else: dependent[call["id"]] = {"call": call, "deps": set(deps)} # 先执行无依赖的 results = {} tasks = [self._execute_and_store(call, results) for call in independent] await asyncio.gather(*tasks, return_exceptions=True) # 再执行有依赖的（拓扑排序） while dependent: ready = [ cid for cid, info in dependent.items() if info["deps"].issubset(set(results.keys())) ] if not ready: # 有循环依赖，强制执行剩余的 ready = list(dependent.keys()) tasks = [] for cid in ready: tasks.append(self._execute_and_store(dependent[cid]["call"], results)) del dependent[cid] await asyncio.gather(*tasks, return_exceptions=True) return list(results.values()) 安全沙箱：永远不要直接执行 LLM 生成的代码 import subprocess import tempfile import os class CodeExecutionSandbox: """安全执行 LLM 生成的代码""" def __init__(self): self.allowed_modules = {"math", "statistics", "json", "re"} self.timeout = 5 self.memory_limit = "256m" async def execute_python(self, code: str) -> dict: # 1. 静态检查 violations = self._static_check(code) if violations: return {"error": "安全检查失败", "violations": violations} # 2. 在 Docker 容器中执行 with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: f.write(code) f.flush() try: result = subprocess.run( ["docker", "run", "--rm", "--memory", self.memory_limit, "--cpus", "0.5", "--network", "none", # 无网络 "--read-only", # 只读文件系统 "--tmpfs", "/tmp:size=10m", # 临时目录 "-v", f"{f.name}:/app/code.py:ro", "python:3.12-slim", "python", "/app/code.py"], capture_output=True, timeout=self.timeout, text=True ) return { "stdout": result.stdout[:5000], # 截断 "stderr": result.stderr[:2000], "exit_code": result.returncode } except subprocess.TimeoutExpired: return {"error": "执行超时"} finally: os.unlink(f.name) def _static_check(self, code: str) -> list: violations = [] # 检查 import for line in code.split("\n"): if "import" in line: module = line.split("import")[-1].strip().split(".")[0] if module not in self.allowed_modules: violations.append(f"禁止导入模块: {module}") # 检查危险函数 dangerous = ["open(", "exec(", "eval(", "os.system", "subprocess", "__import__"] for d in dangerous: if d in code: violations.append(f"禁止使用: {d}") return violations 审计日志：记录每次调用 import logging from datetime import datetime logger = logging.getLogger("function_caller") class AuditLogger: def log_call(self, function_name: str, arguments: dict, result: dict, duration_ms: float, success: bool): logger.info(json.dumps({ "timestamp": datetime.utcnow().isoformat(), "function": function_name, "arguments": self._sanitize(arguments), # 脱敏 "result_size": len(str(result)), "duration_ms": duration_ms, "success": success, "trace_id": self._get_trace_id(), }, ensure_ascii=False)) def _sanitize(self, args: dict) -> dict: """脱敏处理""" sensitive_keys = {"password", "token", "api_key", "credit_card"} sanitized = {} for k, v in args.items(): if k.lower() in sensitive_keys: sanitized[k] = "***REDACTED***" else: sanitized[k] = v return sanitized 性能优化 函数定义缓存 # 工具定义不要每次请求都重新构建 from functools import lru_cache @lru_cache(maxsize=1) def get_tool_definitions(): return [ {"type": "function", "function": schema} for schema in load_all_schemas() ] 函数调用结果缓存 import hashlib from datetime import timedelta class ResultCache: def __init__(self, redis_client): self.redis = redis_client self.ttl = timedelta(minutes=10) async def get_or_execute(self, func_name: str, args: dict, executor): # 对幂等函数做缓存 cache_key = self._make_key(func_name, args) cached = await self.redis.get(cache_key) if cached: return json.loads(cached) result = await executor(func_name, args) await self.redis.setex( cache_key, int(self.ttl.total_seconds()), json.dumps(result, ensure_ascii=False) ) return result def _make_key(self, name: str, args: dict) -> str: arg_hash = hashlib.md5(json.dumps(args, sort_keys=True).encode()).hexdigest() return f"fc:{name}:{arg_hash}" 总结 Function Calling 在生产中可靠运行的关键：严格的 Schema 设计是契约，二次验证是防线，错误恢复是韧性，并行管理是效率，安全沙箱是底线，审计日志是追溯。把每个 LLM 返回的函数调用都当作不可信输入来处理，就能避免大部分生产事故。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

为什么需要 LLM API 网关 当你的团队同时使用 OpenAI、Anthropic、本地模型等多个 LLM 提供商时，直接在业务代码中调用各家 API 会带来一系列问题：密钥散落各处、无法统一限流、缺乏调用审计、模型切换成本高。LLM API 网关就是解决这些问题的中间层。 ┌──────────────────────────────────────────────────────┐ │ 业务服务层 │ │ (Chat UI / RAG / Agent / Code Gen) │ └──────────────────┬───────────────────────────────────┘ │ 统一 API: POST /v1/chat/completions ┌──────────────────▼───────────────────────────────────┐ │ LLM API Gateway │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌────────────┐ │ │ │ 路由引擎 │ │ 限流器 │ │ 缓存层 │ │ 审计日志 │ │ │ └─────────┘ └─────────┘ └─────────┘ └────────────┘ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ 密钥管理 │ │ 重试器 │ │ 负载均衡│ │ │ └─────────┘ └─────────┘ └─────────┘ │ └──────┬──────────┬──────────┬──────────┬─────────────┘ │ │ │ │ ┌────▼────┐┌───▼────┐┌───▼─────┐┌───▼──────┐ │ OpenAI ││Claude ││ Llama ││ Qwen │ │ API ││ API ││ Local ││ API │ └─────────┘└────────┘└─────────┘└──────────┘ 多模型路由 路由引擎是网关的核心。它根据任务复杂度、成本预算、延迟要求将请求分发到最合适的模型。 ...

单点风险分析 LLM 服务链路中的每个环节都可能成为单点故障： 故障点 影响 概率 恢复时间 LLM 提供商 API 宕机 完全不可用 中 5-60min LLM 提供商限流 部分请求失败 高 1-5min 模型版本下线 特定模型不可用 低 需代码变更 网络抖动 延迟升高/超时 高 秒级 自建 GPU 节点故障 推理能力下降 中 10-30min Embedding 服务故障 RAG 检索失效 低 5-15min 核心原则：任何单一 LLM 提供商都不应成为系统的单点依赖。 多模型冗余 模型池设计 from dataclasses import dataclass, field from enum import Enum import asyncio from collections import defaultdict class ModelStatus(Enum): HEALTHY = "healthy" DEGRADED = "degraded" # 可用但质量/延迟有问题 UNHEALTHY = "unhealthy" @dataclass class ModelEndpoint: name: str provider: str # openai / anthropic / local priority: int # 1=首选, 2=备选... status: ModelStatus = ModelStatus.HEALTHY consecutive_errors: int = 0 last_error_time: float = 0 class ModelPool: def __init__(self): self.endpoints = [ ModelEndpoint("gpt-4o", "openai", priority=1), ModelEndpoint("claude-sonnet", "anthropic", priority=2), ModelEndpoint("llama-70b-local", "local", priority=3), ] self.circuit_breakers = defaultdict(CircuitBreaker) def get_available(self, exclude_degraded=False): """按优先级返回可用端点""" endpoints = sorted(self.endpoints, key=lambda e: e.priority) result = [] for ep in endpoints: if ep.status == ModelStatus.UNHEALTHY: continue if exclude_degraded and ep.status == ModelStatus.DEGRADED: continue result.append(ep) return result 故障检测与自动切换 class FailoverHandler: def __init__(self, pool: ModelPool, max_retries=3): self.pool = pool self.max_retries = max_retries async def complete_with_failover(self, messages, **kwargs): endpoints = self.pool.get_available() last_error = None for ep in endpoints: try: response = await self._call_with_timeout( ep, messages, timeout=30, **kwargs ) # 成功调用，重置错误计数 ep.consecutive_errors = 0 ep.status = ModelStatus.HEALTHY return response, ep.name except asyncio.TimeoutError: self._record_error(ep, "timeout") last_error = "timeout" except RateLimitError: self._record_error(ep, "rate_limit") last_error = "rate_limit" except Exception as e: self._record_error(ep, str(e)) last_error = str(e) # 所有端点都失败 raise AllEndpointsFailedError(f"Last error: {last_error}") def _record_error(self, endpoint: ModelEndpoint, reason: str): endpoint.consecutive_errors += 1 endpoint.last_error_time = time.time() if endpoint.consecutive_errors >= 5: endpoint.status = ModelStatus.UNHEALTHY logger.warning(f"Endpoint {endpoint.name} marked unhealthy: {reason}") elif endpoint.consecutive_errors >= 2: endpoint.status = ModelStatus.DEGRADED logger.info(f"Endpoint {endpoint.name} degraded: {reason}") async def _call_with_timeout(self, endpoint, messages, timeout, **kwargs): return await asyncio.wait_for( self._dispatch(endpoint, messages, **kwargs), timeout=timeout ) 优雅降级 当所有高端模型不可用时，系统应自动降级到更小/更便宜的模型，而非直接报错。 ...