以实战论 AI：诚邀一线工程师分享模型调优、工程落地真实案例

聊 AI 的文章太多了，但大部分要么停在"概念科普"层面，要么就是"我用了某个模型效果很好"的体验式分享。真正把模型从 60 分调到 85 分的过程，从 demo 到生产踩的那些坑，反而很少有人系统性地讲。

这篇文章不讲概念，只讲实战。模型调优怎么做的、工程上怎么落地的、踩了什么坑、怎么解决的——全是真实经历。如果你也在搞大模型落地，希望这些经验能帮你少走点弯路。

一、模型调优：从 60 分到 85 分的那些事

1.1 先搞清楚：你的模型为什么只有 60 分

很多人拿到一个开源模型，跑了几条测试觉得效果不行，第一反应是"换个更大的模型"。但大模型不一定能解决你的问题——很多时候不是模型不够大，是你喂给它的东西不对。

我见过太多团队，模型选型花了两周，prompt 调优只花了十分钟。这个优先级完全反了。

先看一张调优路径图，搞清楚该在哪发力：

核心原则：先调 prompt，再调检索，最后才考虑微调。微调成本最高、收益最不确定，应该是最后的手段。

1.2 实战案例：运维知识库问答的调优过程

我们做了一套运维知识库问答系统，用 Qwen2.5-14B + RAG，第一版的效果大概是：

• 检索命中率：55%（10 个问题里有 4.5 个没检索到正确文档）
• 回答准确率：60%
• 幻觉率：18%

这个数据显然上不了线。接下来花了两周时间做调优，最终指标：

• 检索命中率：88%
• 回答准确率：85%
• 幻觉率：4%

下面是每一步具体做了什么。

1.3 第一步：优化检索（命中率 55% → 88%）

检索是 RAG 的地基，检索不对，后面模型再强也白搭。我们排查发现命中率低有三个原因：

原因一：向量模型不适合中文运维场景。之前用的是通用的 text2vec，对"Redis 脑裂""连接池打满"这种运维术语理解不好。换成了 bge-large-zh-v1.5 后，命中率直接涨了 12 个百分点。

原因二：分块策略有问题。之前按固定 500 字符切，把操作步骤切断了。比如一段"第一步检查...第二步执行..."被切成了两个块，检索只命中了第二步，模型生成时缺少第一步的上下文。

改成按 Markdown 标题 + 段落分块，保留语义完整性。同时设置了 100 字符的重叠窗口，防止边界信息丢失。

原因三：没有 Rerank。向量检索召回的 top-10 里，最相关的可能排在第 7、8 位，但最终只取 top-3 喂给模型，直接把正确答案丢了。

加了 bge-reranker 做二次排序后，最相关的文档基本都排进了 top-3。

```python #!/usr/bin/env python3 """ RAG 检索优化工具包 - 多阶段检索：向量召回 → Rerank 精排 → 上下文组装 - 分块策略：按 Markdown 结构分块 + 滑动窗口重叠 """ import re import json import numpy as np from sentence_transformers import SentenceTransformer, CrossEncoder # ============ 1. 智能分块 ============ class MarkdownChunker: """按 Markdown 标题结构分块，保留语义完整性""" def __init__(self, max_chunk_size=800, overlap=100): self.max_chunk_size = max_chunk_size self.overlap = overlap def chunk(self, text, source=""): """将文档按标题+段落分块""" # 按 Markdown 标题切分 sections = self._split_by_headers(text) chunks = [] for section in sections: title = section["title"] content = section["content"].strip() if not content: continue # 如果单个 section 太长，再按段落切 if len(content) > self.max_chunk_size: paragraphs = content.split("\n\n") current_chunk = "" for para in paragraphs: if len(current_chunk) + len(para) > self.max_chunk_size: if current_chunk: chunks.append({ "content": f"{title}\n{current_chunk}", "source": source, "section": title, }) # 重叠窗口：保留上一块末尾的内容 overlap_text = current_chunk[-self.overlap:] if len(current_chunk) > self.overlap else "" current_chunk = overlap_text + "\n" + para else: current_chunk += "\n\n" + para if current_chunk else para if current_chunk.strip(): chunks.append({ "content": f"{title}\n{current_chunk}", "source": source, "section": title, }) else: chunks.append({ "content": f"{title}\n{content}", "source": source, "section": title, }) return chunks def _split_by_headers(self, text): """按 # ## ### 标题切分""" sections = [] current_title = "" current_content = "" for line in text.split("\n"): if re.match(r'^#{1,4}\s', line): if current_content.strip(): sections.append({"title": current_title, "content": current_content}) current_title = line.strip() current_content = "" else: current_content += line + "\n" if current_content.strip(): sections.append({"title": current_title, "content": current_content}) return sections # ============ 2. 多阶段检索 ============ class HybridRetriever: """向量召回 + Rerank 精排""" def __init__(self): # 向量模型（召回阶段） self.encoder = SentenceTransformer("BAAI/bge-large-zh-v1.5") # Rerank 模型（精排阶段） self.reranker = CrossEncoder("BAAI/bge-reranker-large") self.documents = [] self.embeddings = None def index(self, chunks): """构建向量索引""" self.documents = chunks texts = [c["content"] for c in chunks] self.embeddings = self.encoder.encode(texts, normalize_embeddings=True) print(f"索引完成: {len(chunks)} 个文档块") def retrieve(self, query, top_k=3, recall_k=20): """ 两阶段检索： 1. 向量召回 top-k*5（宽召回） 2. Rerank 精排取 top-k（精筛选） """ # === 第一阶段：向量召回 === query_emb = self.encoder.encode([query], normalize_embeddings=True) scores = np.dot(self.embeddings, query_emb.T).flatten() # 宽召回：取 recall_k 个候选 candidate_indices = np.argsort(scores)[-recall_k:][::-1] candidates = [self.documents[i] for i in candidate_indices] # === 第二阶段：Rerank 精排 === pairs = [[query, c["content"]] for c in candidates] rerank_scores = self.reranker.predict(pairs) # 按 rerank 分数排序，取 top_k ranked = sorted(zip(candidates, rerank_scores), key=lambda x: x[1], reverse=True) results = [] for doc, score in ranked[:top_k]: results.append({ **doc, "rerank_score": float(score), }) return results # ============ 3. 使用示例 ============ if __name__ == "__main__": # 模拟知识库文档 docs = [ { "content": """## Redis 集群脑裂处理 当 Redis 集群出现网络分区时，可能出现脑裂问题。 处理步骤： 1. 配置 min-slaves-to-write=1，确保主节点至少有一个从节点确认写入 2. 启用 Sentinel 自动故障转移 3. 检查网络分区恢复后，确保旧主节点降级为从节点 4. 验证数据一致性""", "source": "Redis运维手册.md" }, { "content": """## MySQL 慢查询优化 慢查询排查步骤： 1. 开启 slow_query_log 2. 设置 long_query_time=1 3. 使用 EXPLAIN 分析执行计划 4. 检查索引覆盖情况 5. 优化 SQL 写法（避免 SELECT *）""", "source": "MySQL调优指南.md" }, ] # 分块 chunker = MarkdownChunker(max_chunk_size=800, overlap=100) all_chunks = [] for doc in docs: chunks = chunker.chunk(doc["content"], source=doc["source"]) all_chunks.extend(chunks) # 检索 retriever = HybridRetriever() retriever.index(all_chunks) # 测试 results = retriever.retrieve("Redis 脑裂怎么处理？", top_k=3) for r in results: print(f"来源: {r['source']} | 分数: {r['rerank_score']:.4f}") print(f"内容: {r['content'][:100]}...") print() ```

1.4 第二步：优化 Prompt（准确率 60% → 75%）

检索优化完之后，准确率从 60% 涨到了 75%——因为模型终于能看到正确的上下文了。但还有 25% 的回答不够好，这部分靠 prompt 优化来解决。

调优前后的 prompt 对比：

调优前的 prompt：

``` 请根据以下参考信息回答问题。 参考信息：{context} 问题：{question} 回答： ```

问题很明显：没有约束输出范围、没有引导推理过程、没有处理"信息不足"的情况。

调优后的 prompt：

``` 你是一个专业的运维工程师。请严格按照以下要求回答问题。 ## 参考信息 {context} ## 回答要求 1. 只能基于参考信息回答，不得编造参考信息中不存在的内容 2. 如果参考信息不足以回答问题，请明确回答"当前知识库未覆盖此问题，建议查阅以下文档：[列出可能相关的文档名]" 3. 回答时先给出结论，再给出操作步骤 4. 操作步骤必须编号，每步不超过两句话 5. 如果涉及风险操作，必须标注「⚠️ 风险提示」 ## 示例 问题：Redis 内存满了怎么办？ 回答： Redis 内存满了可以通过以下方式处理： 1. 检查内存使用情况：执行 INFO memory 查看 used_memory 和 maxmemory 2. 配置淘汰策略：设置 maxmemory-policy 为 allkeys-lru ⚠️ 风险提示：设置淘汰策略会导致部分 key 被自动删除 3. 扩容：如果数据量持续增长，考虑增加 Redis 节点 ## 当前问题 {question} ```

三个关键改进：

• 加了 Few-shot 示例：让模型知道"好回答长什么样"
• 加了兜底策略：信息不足时明确说"不知道"，而不是编一个
• 加了结构约束：先结论后步骤、步骤编号、风险标注

光这三个改进，准确率就从 75% 涨到了 82%，幻觉率从 12% 降到了 6%。

1.5 第三步：领域微调（准确率 82% → 85%）

最后一步是微调。说实话，这步的投入产出比不如前两步高——花了三天准备数据、一天微调、一天评估，准确率只涨了 3 个百分点。但这个 3% 恰好是之前怎么调 prompt 都上不去的部分：领域术语理解。

比如"灰度发布""蓝绿部署""熔断降级"这些概念，通用模型的理解是"大概知道但不够精准"。用团队自己的发布流程文档做 SFT 数据，让模型学会"我们公司说的灰度发布具体是什么流程"。

```python #!/usr/bin/env python3 """ LoRA 微调数据准备脚本 - 从历史问答数据中构造 SFT 训练集 - 格式化为 Qwen 模型要求的格式 """ import json import random from pathlib import Path # 微调数据模板 SFT_TEMPLATE = { "instruction": "你是一个专业的运维工程师，请基于团队知识库回答问题。", "input_format": "问题：{question}\n\n参考信息：{context}", "output_format": "{answer}", } def build_sft_dataset(raw_qa_pairs, output_file): """ 将原始问答对转换为 SFT 训练数据 raw_qa_pairs 格式： [ { "question": "灰度发布怎么做？", "context": "相关文档片段...", "answer": "标准回答..." }, ] """ sft_data = [] for pair in raw_qa_pairs: # 构造指令 instruction = SFT_TEMPLATE["instruction"] # 构造输入 user_input = SFT_TEMPLATE["input_format"].format( question=pair["question"], context=pair["context"], ) # 构造输出（人工标注的标准答案） output = SFT_TEMPLATE["output_format"].format( answer=pair["answer"], ) sft_data.append({ "instruction": instruction, "input": user_input, "output": output, "history": [], # 单轮对话 }) # 打乱顺序 random.shuffle(sft_data) # 划分训练集和验证集（9:1） split = int(len(sft_data) * 0.9) train_data = sft_data[:split] val_data = sft_data[split:] # 保存 with open(output_file.replace(".json", "_train.json"), "w", encoding="utf-8") as f: json.dump(train_data, f, ensure_ascii=False, indent=2) with open(output_file.replace(".json", "_val.json"), "w", encoding="utf-8") as f: json.dump(val_data, f, ensure_ascii=False, indent=2) print(f"SFT 数据集生成完成:") print(f" 训练集: {len(train_data)} 条 → {output_file.replace('.json', '_train.json')}") print(f" 验证集: {len(val_data)} 条 → {output_file.replace('.json', '_val.json')}") # 数据增强：从知识库自动生成问答对 def auto_generate_qa(knowledge_docs, llm_client): """用大模型从知识库文档自动生成问答对""" generated = [] for doc in knowledge_docs: prompt = f"""请根据以下技术文档，生成 3 个问答对。 要求： 1. 问题应该是运维工程师实际会遇到的问题 2. 答案必须完全基于文档内容，不得编造 3. 答案格式：先结论，后步骤 文档内容： {doc["content"]} 输出 JSON 数组格式： [{{"question": "...", "answer": "..."}}, ...] """ response = llm_client.chat.completions.create( model="qwen2.5-14b-instruct", messages=[{"role": "user", "content": prompt}], temperature=0.5, ) try: qa_pairs = json.loads(response.choices[0].message.content) for qa in qa_pairs: generated.append({ "question": qa["question"], "context": doc["content"][:500], # 取前500字符作为上下文 "answer": qa["answer"], "source": doc["source"], }) except Exception as e: print(f"解析失败: {e}") print(f"自动生成 {len(generated)} 个问答对") return generated if __name__ == "__main__": # 模拟人工标注的问答对 manual_qa = [ { "question": "灰度发布的标准流程是什么？", "context": "灰度发布分为四个阶段：准备阶段、灰度阶段、观察阶段、全量阶段...", "answer": "灰度发布标准流程：\n1. 准备阶段：确认灰度比例（通常 5%→20%→50%→100%）\n2. 灰度阶段：通过流量网关将指定比例流量导入新版本\n3. 观察阶段：每级灰度后观察 30 分钟，检查错误率和延迟\n4. 全量阶段：确认无异常后全量切换", "source": "发布流程规范.md", }, ] build_sft_dataset(manual_qa, "sft_dataset.json") ```

微调用 LLaMA-Factory 框架，LoRA 方式（r=8, alpha=16），只训练了 3 个 epoch，在单张 A100 上跑了一个多小时。模型文件只有 80MB，推理时挂载到基础模型上即可，不影响原始模型的能力。

二、工程落地：从"能跑"到"能扛"

2.1 Demo 和生产的差距

模型调好了，离上线还差十万八千里。Demo 环境你一个人用，QPS 不到 1；生产环境几十个人同时用，QPS 可能冲到 50+。Demo 环境挂了你重启就行；生产环境挂了有人打电话找你。

下面这张图是从 demo 到生产需要补的工程能力：

2.2 实战：生产级 AI 服务架构

下面是一个完整的 AI 服务封装，包含负载均衡、超时控制、降级策略：

```python #!/usr/bin/env python3 """ 生产级 AI 服务封装 - 多模型实例负载均衡 - 请求超时 + 重试 + 降级 - 全链路监控 - 成本控制 """ import time import json import random import threading from collections import defaultdict from dataclasses import dataclass, field from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout from openai import OpenAI # ============ 1. 模型实例池 ============ @dataclass class ModelInstance: """单个模型实例""" name: str base_url: str model: str client: OpenAI = None request_count: int = 0 error_count: int = 0 avg_latency: float = 0 healthy: bool = True def __post_init__(self): self.client = OpenAI(base_url=self.base_url, api_key="internal", timeout=30) def chat(self, messages, temperature=0.3, max_tokens=2048): start = time.time() try: response = self.client.chat.completions.create( model=self.model, messages=messages, temperature=temperature, max_tokens=max_tokens, ) latency = time.time() - start self.request_count += 1 self.avg_latency = (self.avg_latency * (self.request_count - 1) + latency) / self.request_count return response.choices[0].message.content, latency except Exception as e: self.error_count += 1 if self.error_count > 10: self.healthy = False print(f"⚠️ 实例 {self.name} 标记为不健康: {e}") raise class ModelInstancePool: """模型实例池 + 负载均衡""" def __init__(self, instances): self.instances = instances self.lock = threading.Lock() def get_instance(self, strategy="least_loaded"): """选择一个健康的实例""" healthy = [i for i in self.instances if i.healthy] if not healthy: # 全部不健康，尝试恢复 for i in self.instances: i.healthy = True i.error_count = 0 healthy = self.instances print("⚠️ 所有实例不健康，尝试全部恢复") if strategy == "round_robin": return random.choice(healthy) elif strategy == "least_loaded": return min(healthy, key=lambda x: x.request_count) else: return healthy[0] def call_with_failover(self, messages, **kwargs): """带故障转移的调用""" max_retries = 3 last_error = None for attempt in range(max_retries): instance = self.get_instance() try: result, latency = instance.chat(messages, **kwargs) return result, latency, instance.name except Exception as e: last_error = e print(f"实例 {instance.name} 调用失败 (attempt {attempt+1}): {e}") continue raise last_error # ============ 2. 超时 + 降级 ============ class AIServiceWithFallback: """带超时和降级的 AI 服务""" def __init__(self, instance_pool, timeout_seconds=15): self.pool = instance_pool self.timeout = timeout_seconds self.executor = ThreadPoolExecutor(max_workers=10) # 降级策略缓存（常见问题的标准答案） self.fallback_cache = {} # 监控指标 self.metrics = { "total_requests": 0, "success_count": 0, "timeout_count": 0, "fallback_count": 0, "latency_list": [], } def chat(self, messages, question=""): """带超时和降级的对话""" self.metrics["total_requests"] += 1 start = time.time() try: # 提交到线程池，设置超时 future = self.executor.submit( self.pool.call_with_failover, messages ) result, latency, instance_name = future.result(timeout=self.timeout) self.metrics["success_count"] += 1 self.metrics["latency_list"].append(time.time() - start) # 缓存高频问题的答案 if question and self.metrics["total_requests"] % 100 == 0: self._update_fallback_cache(question, result) return { "answer": result, "latency": latency, "instance": instance_name, "fallback": False, } except FutureTimeout: self.metrics["timeout_count"] += 1 print(f"⏱️ 请求超时 ({self.timeout}s)，尝试降级") return self._fallback(question) except Exception as e: self.metrics["timeout_count"] += 1 print(f"❌ 请求失败: {e}，尝试降级") return self._fallback(question) def _fallback(self, question): """降级策略""" self.metrics["fallback_count"] += 1 # 策略1：检查缓存 if question in self.fallback_cache: return { "answer": self.fallback_cache[question], "latency": 0, "instance": "cache", "fallback": True, } # 策略2：返回友好提示 + 建议操作 return { "answer": "AI 服务当前响应较慢，请稍后重试。如紧急问题请联系值班运维。", "latency": 0, "instance": "fallback", "fallback": True, } def _update_fallback_cache(self, question, answer): """更新降级缓存（只缓存高频问题）""" if len(answer) > 50: # 太短的不缓存 self.fallback_cache[question] = answer # 缓存大小限制 if len(self.fallback_cache) > 200: # 淘汰最旧的（简化版，实际用 LRU） oldest = next(iter(self.fallback_cache)) del self.fallback_cache[oldest] def get_metrics(self): """获取监控指标""" m = self.metrics.copy() if m["latency_list"]: m["avg_latency"] = sum(m["latency_list"]) / len(m["latency_list"]) m["p95_latency"] = sorted(m["latency_list"])[int(len(m["latency_list"]) * 0.95)] m["p99_latency"] = sorted(m["latency_list"])[int(len(m["latency_list"]) * 0.99)] m["success_rate"] = m["success_count"] / max(m["total_requests"], 1) m["fallback_rate"] = m["fallback_count"] / max(m["total_requests"], 1) return m # ============ 3. 使用示例 ============ if __name__ == "__main__": # 创建多个模型实例（模拟多副本） instances = [ ModelInstance("llm-node-1", "http://10.10.100.50:8000/v1", "qwen2.5-14b-instruct"), ModelInstance("llm-node-2", "http://10.10.100.51:8000/v1", "qwen2.5-14b-instruct"), ModelInstance("llm-node-3", "http://10.10.100.52:8000/v1", "qwen2.5-14b-instruct"), ] pool = ModelInstancePool(instances) service = AIServiceWithFallback(pool, timeout_seconds=15) # 模拟并发请求 def make_request(question): messages = [ {"role": "system", "content": "你是运维助手"}, {"role": "user", "content": question}, ] result = service.chat(messages, question=question) status = "✅" if not result["fallback"] else "🔄降级" print(f"{status} [{result['instance']}] {result['latency']:.2f}s | {question[:30]}") # 并发测试 questions = [ "Redis 连接池打满怎么处理？", "K8s Pod 频繁重启是什么原因？", "MySQL 慢查询怎么排查？", "如何配置 Nginx 负载均衡？", "Docker 容器无法启动怎么办？", ] with ThreadPoolExecutor(max_workers=5) as executor: for q in questions: executor.submit(make_request, q) # 打印监控指标 print("\n" + "=" * 60) print("服务监控指标:") metrics = service.get_metrics() for k, v in metrics.items(): if isinstance(v, float): print(f" {k}: {v:.2f}") else: print(f" {k}: {v}") ```

2.3 一次完整的请求时序

生产环境下，一次用户请求从进来到返回，经过了这些环节：

2.4 实际场景：一次线上故障复盘

上线第二周遇到了一次故障，复盘过程很有参考价值。

现象：下午 2 点开始，用户反馈 AI 回答变慢，从平均 2 秒涨到 8 秒，部分请求直接超时。

排查过程：

根因分析：vLLM 默认的 ​​max_num_seqs​​（最大并发序列数）没有限制，下午高峰期并发请求堆积，KV Cache 不断膨胀，最终把实例 3 的内存撑爆了。OOM 后请求全部涌向实例 1 和 2，导致连锁雪崩。

修复措施：

• 限制 ​​max_num_seqs=128​​，防止 KV Cache 无限膨胀
• 加了显存使用率告警，超过 85% 就通知扩容
• 降级策略生效，超时请求返回了缓存答案，没有完全白屏

这个故障教会我一件事：demo 的时候你永远碰不到并发问题，但生产环境的第一个 bug 一定是并发问题。

三、成本控制：别让 AI 变成烧钱机器

3.1 成本黑洞在哪

AI 服务的成本主要来自三块：

最大的黑洞是 GPU 推理。一张 A100 每小时成本约 15-20 元（云厂商按量计费），如果你的服务 24 小时跑着但只有白天有流量，那夜间的 GPU 就是纯浪费。

3.2 成本优化策略

3.3 模型路由：简单问题用小模型

不是所有问题都需要 14B 模型回答。"Redis 怎么重启"这种简单问题，7B 模型就能搞定，没必要走 14B 浪费算力。

```python #!/usr/bin/env python3 """ 模型路由器 - 根据问题复杂度路由到不同大小的模型 - 简单问题用 7B，复杂问题用 14B - 降低平均推理成本 """ import re from openai import OpenAI # 路由判断模型（用小模型做分类，成本低） ROUTER_CLIENT = OpenAI(base_url="http://10.10.100.50:8000/v1", api_key="internal") # 业务模型池 MODEL_POOL = { "simple": OpenAI(base_url="http://10.10.100.50:8001/v1", api_key="internal"), # 7B "complex": OpenAI(base_url="http://10.10.100.50:8000/v1", api_key="internal"), # 14B } ROUTER_PROMPT = """请判断以下问题的复杂度等级。 复杂度判断标准： - simple: 简单事实查询、单一步骤操作、常见问题（如"Redis 怎么重启"） - complex: 多步骤排查、需要推理分析、跨系统关联（如"数据库连接池打满同时 Redis 延迟升高怎么排查"） 问题：{question} 只回答 simple 或 complex，不要其他内容。 """ # 高频问题缓存（命中缓存不走模型） QUERY_CACHE = {} def route_and_answer(question, context=""): """路由 + 回答""" # 1. 检查缓存 cache_key = question.strip().lower() if cache_key in QUERY_CACHE: print(f" → 缓存命中") return QUERY_CACHE[cache_key], 0, "cache" # 2. 路由判断 route_response = ROUTER_CLIENT.chat.completions.create( model="qwen2.5-7b-instruct", messages=[{"role": "user", "content": ROUTER_PROMPT.format(question=question)}], temperature=0, max_tokens=10, ) complexity = route_response.choices[0].message.content.strip().lower() # 3. 路由到对应模型 if "simple" in complexity: model_key = "simple" model_name = "qwen2.5-7b-instruct" else: model_key = "complex" model_name = "qwen2.5-14b-instruct" print(f" → 路由到 {model_name}") # 4. 调用业务模型 messages = [ {"role": "system", "content": "你是运维工程师，请基于参考信息回答问题。"}, {"role": "user", "content": f"参考信息：{context}\n\n问题：{question}"}, ] import time start = time.time() response = MODEL_POOL[model_key].chat.completions.create( model=model_name, messages=messages, temperature=0.3, max_tokens=2048, ) latency = time.time() - start answer = response.choices[0].message.content # 5. 缓存高频问题 QUERY_CACHE[cache_key] = answer return answer, latency, model_name if __name__ == "__main__": test_questions = [ ("Redis 怎么重启？", "简单问题 → 应路由到 7B"), ("数据库连接池打满同时 Redis 延迟升高，怎么排查？", "复杂问题 → 应路由到 14B"), ("怎么查看 Nginx 日志？", "简单问题 → 应路由到 7B"), ("K8s 集群中 Pod 反复重启，同时服务间调用超时，根因可能是什么？", "复杂问题 → 应路由到 14B"), ] for question, expectation in test_questions: print(f"\n问题: {question}") print(f"预期: {expectation}") answer, latency, model = route_and_answer(question) print(f"耗时: {latency:.2f}s | 模型: {model}") print(f"回答: {answer[:80]}...") ```

实际运行下来，大约 60% 的问题被路由到了 7B 模型，平均推理成本降低了约 40%。缓存命中率在 25% 左右（高频问题重复率高），进一步降低了成本。

四、调优 vs 落地：两个阶段的工作对比

最后用一张图总结一下"模型调优"和"工程落地"这两个阶段的工作重心差异：

五、避坑指南：血泪经验

最后分享几条踩坑总结：

调优阶段的坑：

1. 别一上来就微调。先把 prompt 调到极致、把检索优化好，这两个做完通常能涨 20-30 分。微调是最后 5 分的事，但成本可能是前 25 分的总和。
2. 评估集比模型重要。没有 50 条以上的人工标注评估集，你根本不知道调优有没有效果。"感觉变好了"不算数。
3. Rerank 是性价比最高的优化。加一个 reranker 模型，命中率涨 10-15 个百分点，成本只增加几毫秒延迟。投入产出比极高。

落地阶段的坑：

1. 并发是你第一个会遇到的问题。Demo 的时候永远不会碰到，上线第一天一定碰到。提前做好请求队列、超时控制、降级策略。
2. 监控不是可选项。没有监控的 AI 服务就像没有仪表盘的汽车——你以为在 60 码跑，其实已经 120 了。至少监控：请求量、延迟分布、成功率、GPU 利用率。
3. 降级策略保命。模型挂了不能让整个系统挂。缓存兜底、友好提示、人工转接，总比白屏强。
4. 成本要天天看。AI 服务很容易出现"某个用户疯狂调用"或者"某个 bug 导致无限重试"的成本事故。设置每日成本上限和异常调用告警。

模型调优和工程落地是两件事，但很多人只做了第一件就觉得完事了。模型 85 分和系统可用之间，还隔着负载均衡、监控告警、降级策略、成本控制这一大堆工程活儿。

这不是什么高深的技术的，都是脏活累活。但正是这些脏活累活，决定了你的 AI 项目是"能跑的 demo"还是"能用的产品"。

别只盯着模型分数看，工程基建一样重要。共勉。