Skip to content

元数据卡

  • 前置知识: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 系统中,缓存可以发生在多个层次。

python
# 语义缓存:缓存的不是精确匹配,而是语义近似的查询
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 协作。标准模式:

python
# 多 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 的输入,逐步精炼结果
python
# 反射模式:生成 + 质量检查
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 系统的生产级评估需要多维度、自动化、持续跟踪。

python
# 生产级 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 系统工程。前方路还很长——但你已经有了地图和指南针。

Built with VitePress | Software Systems Atlas