元数据卡
- 前置知识:ch11-ch13 全部
- 预计时间:50 分钟
- 核心难度:进阶
- 阅读模式:高度专注
- 完成标志:能设计 Cache-Augmented 生成系统、理解多 Agent 协作模式、使用评估框架评测 LLM 应用
你的进度
你在模型工坊里走完了从经典搜索到 LLM 的完整路径。工坊的墙上挂着你一路搭建的每一套系统。
但模型工坊的老工匠看了你的作品,问了一个最务实的问题:“这个系统上线了怎么维护?用户并发上来的时候会不会崩?模型的回答质量怎么持续监控?”
最后一堂课的教具不是算法,而是——系统架构图。
你的任务
单次调用 LLM 只是起点。真实世界的 AI 系统需要缓存层提效、多 Agent 协作完成复杂任务、系统和持续的评估机制。本章给出三个经过实战验证的模式。
本章分层
- 必读:Cache-Augmented Generation、多 Agent 协作架构、评估框架
- 选读:语义缓存 vs 精确缓存的设计权衡
- 进阶:LLM 系统可观测性、A/B 评估设计
破局 · 溯源
你的 RAG 系统跑了三个月,用户说"每次问同样的问题,它都重新去向量数据库查一遍同样的东西"。这是浪费——相同的查询,返回同样的文档,产生同样的答案,但每次都要走一遍完整的检索+生成流程。
你的 Agent 系统处理复杂任务时发现:一个 LLM 做决策时频繁出错(幻觉、遗忘上下文、不能自我修正)。你需要让多个模型/代理协同工作,各司其职。
你需要一个可靠的评估体系——不只是在发布前测一次,而是在持续运营中跟踪每一个指标的变化。
这就是系统模式的用武之地。它们回答的不是"模型怎么训练",而是"系统怎么设计"。
Cache-Augmented Generation
缓存策略不是新概念——但在 LLM 系统中,缓存可以发生在多个层次。
# 语义缓存:缓存的不是精确匹配,而是语义近似的查询
class SemanticCache:
def __init__(self, embedding_model, similarity_threshold=0.95):
self.embedding_model = embedding_model
self.threshold = similarity_threshold
self.cache = {} # embedding_key -> (response, metadata)
def get(self, query):
query_emb = self.embedding_model.embed(query)
for cached_emb, (response, meta) in self.cache.items():
similarity = cosine_similarity(query_emb, cached_emb)
if similarity > self.threshold:
return response
return None
def set(self, query, response):
query_emb = self.embedding_model.embed(query)
self.cache[tuple(query_emb)] = (response, {"timestamp": time.time()})多层缓存策略:
| 层级 | 策略 | 命中场景 | 效果 |
|---|---|---|---|
| 精确缓存 | 完全相同的问题 | FAQ、重复查询 | 0ms 命中 |
| 语义缓存 | 相似的问题 | 同一句话的不同表达 | 毫秒级命中 |
| 检索缓存 | 相同的检索结果 | 多轮对话中同一知识点被反复引用 | 减少向量查询 |
| 生成缓存 | 相同上下文的相同输出 | 模板化生成、固定格式回答 | 减少 LLM 调用 |
多 Agent 协作
复杂任务需要多个具有不同专长的 Agent 协作。标准模式:
# 多 Agent 协作框架(简化版)
class Orchestrator:
def __init__(self):
self.agents = {} # name -> Agent
self.workflows = {} # task_type -> [agent_sequence]
def add_agent(self, name, agent):
self.agents[name] = agent
def register_workflow(self, task_type, agent_sequence):
"""注册一个工作流:按顺序调用 Agent"""
self.workflows[task_type] = agent_sequence
def execute(self, task):
task_type = self.classify_task(task)
workflow = self.workflows.get(task_type)
context = {"task": task, "history": []}
for agent_name in workflow:
agent = self.agents[agent_name]
result = agent.run(context)
context["history"].append({
"agent": agent_name,
"result": result
})
return context["history"][-1]["result"]常见多 Agent 协作模式:
- 主管-工作节点(Orchestrator-Worker):一个协调者分派子任务,多个工作节点各司其职
- 辩论(Debate):多个 Agent 各自提出方案,然后讨论、反驳、改进
- 反射(Reflection):一个 Agent 执行,另一个 Agent 检查输出质量
- 流水线(Pipeline):A 的输出作为 B 的输入,逐步精炼结果
# 反射模式:生成 + 质量检查
class ReflectiveAgent:
def __init__(self, generator, critic):
self.generator = generator
self.critic = critic
def generate_with_review(self, prompt, max_iterations=3):
for i in range(max_iterations):
response = self.generator(prompt)
review = self.critic.evaluate(prompt, response)
if review.passed:
return response
prompt = f"{prompt}\nPrevious attempt: {response}\nImprove: {review.feedback}"
return self.generator(prompt) # 折返回最后一次生成评估框架
"如果你不能测量它,你就不能改进它。"LLM 系统的生产级评估需要多维度、自动化、持续跟踪。
# 生产级 LLM 评估框架
class LLMEvaluationFramework:
def __init__(self):
self.metrics = {}
def register_metric(self, name, func):
"""注册一个评估维度"""
self.metrics[name] = func
def evaluate(self, dataset, model_func):
"""在数据集上运行全部注册的评估"""
results = {name: [] for name in self.metrics}
for example in dataset:
model_output = model_func(example["input"])
for name, metric_func in self.metrics.items():
score = metric_func(example["expected"], model_output)
results[name].append(score)
return {
name: {
"mean": np.mean(scores),
"std": np.std(scores),
"p5": np.percentile(scores, 5), # 低分位数(最差情况)
"p95": np.percentile(scores, 95) # 高分位数
}
for name, scores in results.items()
}
# 使用示例
eval_framework = LLMEvaluationFramework()
eval_framework.register_metric("accuracy", exact_match)
eval_framework.register_metric("faithfulness", check_faithfulness) # 是否忠实于上下文
eval_framework.register_metric("safety", check_safety)
eval_framework.register_metric("usefulness", llm_as_judge)生产环境的评估维度至少涵盖:
- 质量:准确率、忠实度、有用性
- 性能:延迟(P50/P95/P99)、吞吐量
- 成本:每 token 成本、每次调用成本
- 安全:有害内容比率、越狱成功率
- 运营:错误率、降级率、缓存命中率
将评估集成到 CI/CD 流水线中:每次模型/提示词/系统架构变更,自动跑全量评估,结果回归到仪表盘。
结尾:模型工坊的最后一课
"让程序从规则执行走向数据驱动决策。"这就是 Vol 13 的全部故事。你在模型工坊里从零开始:搜索算法让它能找路径,知识推理让它会思考,强化学习让它能试错。然后从经典 ML 的线性模型到树模型,从神经网络的层叠到 Transformer 的注意力革命,最后走到今天的 LLM 生态——对齐、检索增强、多 Agent 协作。
这一路不是技术名词的堆叠。每一层都在回答同一个问题:如何让机器从数据中自主发现规律,并用它做有效决策。 从手动写规则到预训练千亿参数模型,"智能"的来源从工程师的头脑转向了海量数据的统计结构。
模型工坊的门已经打开。你的工具不再只是 if-else 和 for 循环——现在是数据集、模型权重、向量索引、Attention 矩阵。阿花在远方写来的信,你终于可以拆开来看:是的,机器能自己学了。
常见陷阱
- Cache 的过期策略容易被忽略:过时的知识被缓存下来,用户拿到的是三个月前的回答。需要 TTL 或版本化缓存。
- 多 Agent 系统的通信成本不可忽视——每轮消息传递都在消耗 tokens。需要设计精简的上下文。
- 评估指标之间的权衡:安全性和实用性往往冲突(过度安全导致"抱歉,我无法回答这个问题"泛滥)。
- 评估集的质量决定评估的可靠性——如果测试集本身有偏见或过时,评估结果没有意义。
- Agent 系统的可观测性不足时,排查一个错误响应可能需要回溯数十步的交互历史。
通关挑战
- 热身(15 分钟):为你的 RAG 系统设计一个三层缓存策略。画出流程图,标注每层的命中率和延迟数据。
- 挑战(45 分钟):实现一个"主管-工作节点"多 Agent 系统:一个 Agent 负责计划(分解任务为子步骤),两个工作 Agent(一个负责搜索,一个负责生成),一个质检 Agent 检查最终输出。
- 最终挑战(60 分钟):搭建一个完整的 LLM 应用评估流水线。包括:测试集构建、多维度指标计算(准确性 + 忠实度 + 延迟)、性能退化检测(与基线版本对比)。以可视化的方式呈现结果。
旅人笔记
AI 系统设计不是"把模型部署到服务器上"就完了。CAG 让系统高效,多 Agent 让系统聪明,评估框架让系统可信。三个模式合在一起,你才真正拥有一个在生产环境中运行的 AI 系统。至此,模型工坊的课程全部结束——但你在这里学会的思维方式,会在更广阔的世界里持续生效。
-> 下一站
Vol 13 到此结束。AI 与机器学习的旅程从经典搜索走到 LLM 系统工程。前方路还很长——但你已经有了地图和指南针。