《LLM Engineers Handbook》读书笔记
2026-03-24 22:26:53 # LLM #LLM #DeepLearning #Engineering

写在前面：

本文章是关于《LLM Engineer’s Handbook》的部分学习笔记。原书以开发一个名为 LLM Twin 的、模仿某人的写作风格的端到端项目，展示如何运用 LLM 工程和 MLOps ，构建一个最小可行产品以解决特定问题。其中，作者使用较多笔墨来介绍 LLM 在工程化中的开发方法论，对于原理的探讨着墨较少。所以在以原书为主线的前提下会添加部分对于原理的介绍，作为对于方法论的补充说明。碍于本人学识有限，部分叙述难免存在纰漏，请读者注意甄别。感谢所有学者、工程师的开源贡献。

参考资料：

零、前言

机器学习运维（Machine Learning Operations, MLOps）在成功实施 LLM 于生产环境中起着至关重要的作用。MLOps 将 DevOps 的原则扩展至机器学习项目，专注于自动化和简化整个 ML 生命周期。对于 LLM 而言，由于其模型的复杂性和规模，MLOps 尤为重要。它解决了诸如管理大型数据集、处理模型版本控制、确保可重复性以及维持模型性能等挑战。通过融入 MLOps 实践，LLM 项目可以实现更高的效率、可靠性和可扩展性，最终促成更成功和有影响力的部署。

《LLM Engineer’s Handbook》是一本全面指南（以下简称“手册”），旨在将最佳实践应用于 LLM 工程领域。手册中涵盖数据工程、监督微调、模型评估、推理优化和检索增强生成（Retrieval-Augmented Generation, RAG）管道开发等主题。

为直观展示这些概念，手册中将开发一个名为 LLM Twin 的端到端项目，目标是模仿某人的写作风格和个性。该用例将展示如何运用 LLM 工程和 MLOps ，构建一个最小可行产品以解决特定问题。我们读者将了解：

  • 如何为 LLM 收集和准备数据、针对特定任务微调模型、优化推理性能以及实施RAG管道；
  • 如何评估LLM性能、使模型与人类偏好对齐；
  • 部署基于LLM的应用程序。

本笔记与手册结构保持一致，主要分为以下几个内容：

  • 第1章 理解 LLM Twin 的概念和架构：介绍了贯穿全手册的 LLM Twin 项目，作为生产级 LLM 应用的端到端示例。定义了构建可扩展机器学习系统的 FTI 架构，并将其应用于 LLM Twin 的用例；
  • 第2章 工具和安装：介绍用于构建实际 LLM 应用的 Python、MLOps 和云工具，如编排器、实验跟踪器、提示监控和 LLM 评估工具。展示了如何在本地安装和使用这些工具以进行测试和开发；
  • 第3章 数据工程：介绍一个数据收集管道的实现，该管道从 Medium、GitHub 和 Substack 等多个网站抓取数据，并将原始数据存储在数据仓库中。强调了在实际机器学习应用中，从动态来源收集原始数据的重要性，而非依赖静态数据集；
  • 第4章 RAG 特征管道：介绍了检索增强生成（Retrieval-Augmented Generation, RAG）的基本概念，如 Embeddings、基础 RAG 框架、向量数据库，以及如何优化 RAG 应用。通过使用软件最佳实践，设计并实现了 LLM Twin 的 RAG 特征管道，以应用 RAG 理论；
  • 第5章 监督微调（Supervised Fine-Tuning, SFT）：探讨了使用**指令（instruction）-回答（answer）**对来优化预训练语言模型以执行特定任务的过程。涵盖了创建高质量数据集、实施全量微调（full fine-tuning）、LoRA 和 QLoRA 等微调技术，并提供了在自定义数据集上微调 Llama 3.1 8B 模型的实践示范；
  • 第6章 带偏好对齐的微调：介绍了将语言模型与人类偏好对齐的技术，重点关注直接偏好优化（Direct Preference Optimization, DPO）。涵盖了创建自定义偏好数据集、实施 DPO，并提供了使用 Unsloth 库对 TwinLlama-3.1-8B 模型进行对齐的实践示范；
  • 第7章 评估LLM：详细描述了评估语言模型和 LLM 系统性能的各种方法。介绍通用和特定领域的评估，讨论流行的基准测试。本章包括对 TwinLlama-3.1-8B 模型的多标准实践评估；
  • 第8章 推理优化：涵盖了关键的优化策略，如推测解码、模型并行和权重量化。将讨论如何提高推理速度、降低延迟和最小化内存使用，介绍流行的推理引擎并比较其特性；
  • 第9章 RAG推理管道：通过从头开始实施自查询、重排序和过滤向量搜索等方法，探索高级 RAG 技术。涵盖了设计和实现 LLM Twin 的 RAG 推理管道，以及类似于 LangChain 等流行框架的自定义检索模块；
  • 第10章 推理管道部署：介绍了在线、异步和批量推理等机器学习部署策略，有助于将微调后的 LLM Twin 模型架构并部署到AWS SageMaker，并构建 FastAPI 微服务，将 RAG 推理管道作为 RESTful API 公开；
  • 第11章 MLOps和LLMOps：介绍了 LLMOps 的概念，从其在 DevOps 和 MLOps 中的根源开始。本章解释了如何将 LLM Twin 项目部署到云端，如将机器学习管道部署到 AWS，并展示了如何使用 Docker 容器化代码和构建 CI/CD/CT 管道。还在 LLM Twin 的推理管道上添加了提示监控层；
  • 附录 MLOps原则：涵盖了用于构建可扩展、可重复和健壮的机器学习应用的六项 MLOps 原则。

一、理解 LLM Twin 的概念和架构

手册将教我们如何构建一个 LLM Twin，即一个通过将特定个人的写作风格、语气和个性融入 LLM 的 AI 角色。通过这个示例，我们将经历完整的 ML 生命周期，从数据收集到部署和监控。在实现 LLM Twin 的过程中学到的大多数概念都可以应用于其他基于 LLM 或 ML 的应用程序。

The best way to learn about LLMs and production machine learning (ML) is to get your hands dirty and build systems.

学习LLM和生产级机器学习（ML）的最佳方式是亲自动手构建系统。

在开始实施新产品时，从工程的角度来看，我们必须在开始构建之前经过三个规划步骤：

  • 首先，了解我们试图解决的问题以及我们想要构建的内容至关重要。在我们的案例中，LLM Twin 究竟是什么？为什么要构建它？这一步是我们必须思考并专注于**“WHY”**的地方。
  • 其次，为了反映现实世界的场景，我们将设计一个具有最小功能的产品的初级版本。在这里，我们必须清楚地定义产品所具有的核心功能。这些选择是基于时间表、资源和团队知识做出的。这是我们在构思阶段和实际实施之间架起桥梁，并最终回答以下问题：“我们要构建什么（WHAT）？”。
  • 最后，我们将进行系统设计步骤，列出用于构建 LLM 系统的核心架构和设计选型。前两个步主要与产品的设计相关，而最后一个是技术性的，专注于“HOW”。

这三个步骤是在构建现实世界产品时自然发生的。虽然前两个不需要太多的 ML 知识，但是这对于产品的开发而言同样至关重要。简而言之，本章涵盖以下主题：

  • 理解 LLM Twin 概念
  • 规划 LLM Twin 产品的最小可行产品（Minimum Viable Product, MVP）
  • 使用特征/训练/推理管道构建 ML 系统
  • 设计 LLM Twin 的系统架构

什么是 LLM Twin

LLM Twin 是一个将个性化的写作风格、语气融入大型语言模型（LLM）中的 AI 角色。与在整个互联网数据上训练的通用 LLM 不同，LLM Twin 是在个人数据上进行微调的，将这些个人数据“投射（projected）”到大语言模型中。

[!NOTE]

这里有意使用了“投射（projected）”一词。正如其他投射一样，在此过程中丢失大量信息，大模型只能反映出训练数据中信息。

如果我们用鲁迅的文学作品微调 LLM，LLM 将会模仿鲁迅的写作风格，这也被称为风格迁移。我们将利用风格迁移策略，使 LLM 模仿我们个人的风格。

为了将 LLM 调整为特定的风格和语气，除了微调外，我们还将利用各种高级的检索增强生成（RAG）技术，以使用我们先前的 Embedding 来调节自回归过程。我们将在后续章节中详细探讨这些内容。

[!TIP]

什么是嵌入 Embedding ？

在 LLM 的开发领域中，向量 Embedding 在获取文本信息的本质方面起着关键作用。向量 Embedding 的核心是指在数学空间中将单词、句子甚至整个文档表示为密集的低维向量的过程。与依赖于稀疏表示（如one-hot编码）的传统方法不同，向量 Embeddings封装了单词之间的语义关系，并使算法能够理解它们的上下文含义。

通过使用词 Embeddings、句子 Embeddings 或上下文 Embedding 等技术，向量 Embeddings 提供了文本数据的紧凑而有意义的表示。例如，单词 Embeddings 将单词映射到固定长度的向量，其中具有相似含义的单词在向量空间中的位置更接近。这允许高效的语义搜索、信息检索和语言理解任务。

向量 Embedding 的重要性在于它能够将原始文本转换为算法可以理解和推理的数字表示。这种转换过程不仅促进了各种自然语言处理（NLP）任务，而且还作为大型语言模型的基本构建块。向量 Embeddings 使这些模型能够利用嵌入在文本数据中的丰富语义信息，使它们能够生成更连贯和上下文更合适的响应。

我们可以想象这样一个可以对 LLM 进行微调的场景：

  • 小红书、知乎等社交平台：使 LLM 仿照我们自己的风格来编写社交媒体内容；
  • 学术论文和文章：微调 LLM 以撰写正式和学术性的内容；
  • 代码：微调 LLM 使其以特定的代码规范来编写代码。

所有上述场景都可以归结为一个核心策略：收集个人数据集（或其中的一部分），使用不同的算法将其输入到 LLM 中。最终，LLM 将反映所收集数据的语气和风格。

为什么不用 Qwen 这些通用大模型？

Qwen 这些通用大模型非常通用、缺乏独特表达，且往往冗长。盲目使用通用大模型可能会导致以下问题：

  • 幻觉导致的错误信息：需要手动检查生成内容是否存在幻觉，或使用第三方工具进行验证，是一项繁琐且低效的任务。
  • 繁琐的手动提示工程：需要手动编写提示词并注入外部信息，这个过程既耗时又麻烦。此外，由于无法完全控制提示词和输入数据，在不同会话中生成一致的答案也十分困难。虽然可以通过 API 和 LangChain 等工具部分解决此问题，但这需要一定的编程经验。

如果想要高质量且真正有价值的内容时，我们可能会花比直接写作更多的时间去调试 AI 生成的文本。

由此可见，构建私人的大语言模型的关键点在于：

  • 我们收集哪些数据
  • 如何预处理这些数据
  • 如何将数据输入 LLM
  • 如何链接多个提示以获得理想结果
  • 如何评估生成的内容

规划产品的 MVP

既然我们已经了解了什么是 LLM Twin 以及为什么要构建它，那么我们需要明确定义产品的功能。手册中重点关注 LLM Twin 的第一版，即最小可行产品（Minimum Viable Product, MVP），以遵循大多数产品的自然发展周期。

什么是 MVP?

MVP 指的是产品的最小可行版本，即仅包含足够功能来吸引早期用户，并在开发的初始阶段验证产品概念的可行性。通常，MVP 的目标是以最小的投入从市场中收集反馈。

MVP 是一种产品策略，主要有以下优势：

  • 加速产品上市（Accelerated time-to-market）：快速推出产品，以获得早期用户并建立市场影响力。
  • 验证产品理念（Idea validation）：在全面开发产品之前，通过真实用户进行测试，以验证产品是否符合需求。
  • 市场调研（Market research）：深入了解目标用户的偏好，收集有价值的市场反馈。
  • 降低风险（Risk minimization）：减少因产品市场表现不佳而浪费的时间和资源。

在 MVP 中，必须严格遵循 “V”（Viable，可行性） 的原则，即产品必须是可行的。即使产品功能最小化，它也必须提供完整的用户体验，而不是半成品。MVP 需要是一个真正可用的产品，提供流畅的使用体验，让用户愿意持续使用，并随着产品的发展而发展。

LLM Twin 的 MVP 的核心功能

为了保持简单性，我们的 LLM Twin MVP 将具备以下核心功能：

  1. 数据收集
    • 从 小红书、知乎、微信 和 GitHub 账户收集用户的数据。
  2. LLM 训练微调
    • 使用开源 LLM，结合收集的数据进行微调（fine-tuning）。
  3. RAG（检索增强生成）
    • 将收集的数字数据存入向量数据库（vector database），以支持 RAG 机制。
  4. 社交媒体内容生成（例如小红书文章）
    • 用户输入的提示（prompts）
    • RAG 检索，复用并引用用户过往内容
    • 新内容（如文章、论文等）作为 LLM 额外的知识输入
  5. 简单的 Web 界面，提供交互能力：
    • 配置社交媒体链接，并触发数据收集流程
    • 输入提示词（prompts）或外部资源链接，让 LLM Twin 生成内容

MVP 的关键挑战

尽管上述 MVP 可能看起来功能不多，但我们必须确保系统具备以下特性：

  • 成本可控（Cost-effective）：优化计算资源，避免不必要的开销。
  • 可扩展（Scalable）：随着用户增长，系统仍能稳定运行。
  • 模块化（Modular）：方便未来扩展和优化。

至此，我们已经从用户和商业角度探讨了LLM Twin 的价值。最后一步，我们需要从工程实现的角度进行分析，并制定开发计划，明确如何在技术层面实现这个系统。

[!NOTE]

从现在开始，重点将转向 LLM Twin 的具体实现。即使我们专注于上述核心功能，我们仍会基于最新的 LLM 研究成果，并结合最佳的软件工程与 MLOps 实践，构建一个成本可控、可扩展的 LLM 应用。

构建具有特征/训练/推理流水线的 ML 系统

在深入探讨 LLM Twin 架构的具体细节之前，我们需要先理解其核心 ML 体系结构模式——特征/训练/推理（Feature/Training/Inference , FTI）架构。本节将概述 FTI 流水线 的设计，以及它如何帮助我们构建一个结构化的 ML 应用。

ML 系统开发的挑战

构建生产级 ML 系统不仅仅是训练一个模型。从工程角度来看，训练模型通常是最简单的一步。然而，决定正确的架构和超参数，才是让模型真正发挥作用的挑战——这更像是一个研究问题，而不是纯粹的工程问题。

当前，我们关注的重点是如何设计一个可用于生产的架构。即使训练出了高准确率的模型，仅仅基于静态数据集训练它，距离真正的部署仍然很遥远。 我们需要考虑以下问题：

  • 数据处理：如何**摄取（ingest）、清理（clean）和验证（validate）**新的数据？
  • 训练 vs 推理环境：训练和推理（Inference）环境是否需要分开部署？计算资源 如何分配？
  • 特征存储与计算：如何在正确的环境下计算并提供模型所需的特征？
  • 模型部署与服务：如何高效、低成本地提供推理服务？如何版本化、追踪并共享数据集和模型？
  • 监控与维护：如何监控 ML 基础设施和模型的表现？模型如何扩展并持续更新？
  • 自动化：如何自动化模型的部署和训练流程？

[!TIP]

这些问题通常由 ML 或 MLOps 工程师 负责解决，而 研究团队或数据科学团队 主要关注模型训练本身。

Google Cloud 团队提出的 成熟 ML & MLOps 系统 需要包括的组件如下：

screenshot_2025-03-10_15.43.53

  • ML 代码（核心模型开发）
  • 数据收集（Data Collection）
  • 数据验证（Data Verification）
  • 测试与调试（Testing & Debugging）
  • 资源管理（Resource Management）
  • 模型分析（Model Analysis）
  • 流程 & 元数据管理（Process & Metadata Management）
  • 服务基础设施（Serving Infrastructure）
  • 监控系统（Monitoring）

可见，生产化 ML 模型远远不只是写好训练代码这么简单，它涉及多个环节和工程实践。

如何构建一个统一的 ML 系统？

关键问题：如何将所有这些组件连接成一个统一的 ML 系统？我们需要设计一个标准化的架构，使 ML 系统的搭建更加高效、可复用和可扩展。

在传统软件工程中，很多应用可以拆分为 数据库（DB）、业务逻辑（Business Logic）和用户界面（UI） 三大部分。尽管每个部分的实现可能非常复杂，但在高层次的架构设计上，它们仍然可以归纳为这三大模块。

那么，ML 应用是否也能有类似的通用架构呢？

我们需要先回顾一些现有方案，看看它们为什么不适合构建可扩展的 ML 系统，然后再探索更优的解决方案。

以往解决方案的问题

screenshot_2025-03-10_15.47.41

在上图中，我们可以看到大多数 ML 应用程序中常见的架构。这种架构基于单体批处理（monolithic batch）模式，将特征创建（Create Features）、**模型训练（Train Model）和推理（Make Predictions）**紧密耦合在同一个组件中。

采用这种方法可以快速解决 ML 领域中的一个关键问题——训练-推理偏差（training-serving skew）。

  • 训练-推理偏差 发生在训练时和推理时使用的特征计算方式不同，导致模型在生产环境中的表现不如预期。
  • 在这种单体架构中，训练和推理阶段的特征是用相同的代码生成的，因此避免了训练-推理偏差。

单体批处理架构适用于小数据集，因为：

  • 训练、推理使用相同的特征计算代码，避免了训练-推理偏差
  • 通过**批处理（batch mode）**定期运行流水线
  • 预测结果通常被**第三方应用（如 dashboard）**消费

然而，这种架构在面对更大规模的数据时，会引发许多问题：

  • 特征无法复用（既不能在系统内部复用，也不能被其他系统使用）
  • 扩展性差，如果数据规模增加，必须重构代码以支持 PySpark 或 Ray
  • 性能优化困难，如果想用 C++、Java 或 Rust 重写推理模块，会变得极为复杂
  • 团队协作受限，由于特征计算、训练和推理紧耦合在一起，难以拆分给不同的团队
  • 不支持流式计算，如果需要实时训练，无法切换到流式架构
单体架构在实时推理系统中的问题

screenshot_2025-03-10_15.57.35

在上图中，我们可以看到类似的架构被应用于实时推理系统时会带来的额外问题。

在实时推理中，为了生成预测，我们必须通过客户端请求传输整个状态，以便计算特征并输入模型。例如，在电影推荐系统中，理想情况下，我们只需传递 userID 给模型，模型可以基于存储的用户数据计算推荐结果。但在单体架构中，我们必须传递整个用户状态，包括姓名、年龄、性别、观影历史等，使得客户端必须理解如何访问这些状态数据。

这种方法极易出错，因为：

  • 客户端和模型服务 强耦合，客户端必须知道如何查询和构造数据
  • 状态传输成本高，尤其在高并发情况下，传输大量状态信息会影响性能

另一个例子是 LLM + RAG（检索增强生成） 的实现：

  • 在 RAG 模型 中，我们希望能基于外部知识库增强 LLM 的推理能力。
  • 如果没有向量数据库（vector DB），我们必须在每次查询时手动附带所有文档，否则模型无法参考这些外部知识。
  • 这样就导致客户端需要手动查询和管理文档，这不仅不现实，而且是一种反模式（antipattern）。

客户端不应负责查询和计算特征，而应交由服务端处理。

我们将在第 8 章和第 9 章详细介绍 RAG 这一技术。

综上所述，我们的核心问题是如何在不依赖客户端传递完整特征的情况下进行预测。

在另一端的极端案例，Google Cloud提供了一种**生产就绪（production-ready）**的、自动化流水线的 ML 架构（见下图）。

screenshot_2025-03-10_16.10.48

这种架构确实能够解决生产环境中的 ML 部署问题，但它存在以下挑战：

  • 复杂度高，不够直观，非 ML 生产专家很难理解
  • 上手难度大，如果你没有丰富的 ML 生产部署经验，可能会被架构的复杂性劝退
  • 不易渐进式扩展，难以理解如何从小型系统开始并随着需求增长逐步扩展

在接下来的章节，我们将介绍 特征/训练/推理（Feature/Training/Inference , FTI）架构，它是一种直观的 ML 设计，能够有效解决前述的核心问题。

特征/训练/推理（FTI） 架构

[!TIP]

想了解更多关于 FTI 模式的信息，可以参考*“From MLOps to ML Systems with Feature/Training/Inference Pipelines”* by Jim Dowling, CEO and co-founder of Hopsworks：https://www.hopsworks.ai/post/mlops-to-ml-systems-with-fti-pipelines

特征/训练/推理（Feature/Training/Inference , FTI）架构提出了一个清晰直接的思维框架，任何团队或个人都可以遵循它，来完成特征计算、模型训练以及推理管道的部署。该模式表明，任何机器学习系统都可以归结为三个管道：

  • 计算特征*（Feature）*
  • 训练模型*（Training）*
  • 进行推理*（Inference）*

这种架构强大之处在于，我们可以清晰地定义每个管道的职责和接口。最终，系统只有三个核心模块，而不是像 Google Cloud 方案中展示的那种拥有二十个模块的复杂结构，这大大简化了操作和定义的难度。下图展示了特征、训练和推理管道架构。

screenshot_2025-03-10_16.11.00

FTI 架构的核心特点：

  • 每个管道都是独立的组件，可以在不同进程或硬件上运行。
  • 每个管道可以使用不同的技术实现，甚至可以由不同的团队开发和维护。
  • 可扩展性强，允许团队根据实际需求对不同管道独立扩展。
  • 提供清晰的思维导图，帮助团队高效组织 ML 系统架构。

特征管道（Feature Pipeline）

作用：
特征管道的主要任务是从原始数据中提取特征，并生成用于模型训练或推理的特征和标签。但这些特征不会直接传递给模型，而是**存储在特征库（Feature Store）**中。

主要职责：

  • 存储、版本管理、追踪、共享 训练和推理所需的特征。
  • 保持特征的状态，确保训练和推理阶段使用的特征一致，从而避免训练-推理偏差（Training-Serving Skew）。
  • 让训练和推理管道轻松获取数据，保证系统的稳定性和可复现性。

训练管道（Training Pipeline）

作用：
训练管道的任务是**从特征库中提取特征和标签，训练模型，并将训练好的模型存储在模型仓库（Model Registry）**中。

主要职责：

  • 训练一个或多个模型，并存储、版本管理、追踪和共享 这些模型。

  • 模型仓库（Model Registry） 的角色类似于特征库，但重点是管理模型，而不是特征。

  • 记录元数据（Metadata Store）

    ，包括：

    • 训练使用的特征、标签及其版本，确保模型的可追溯性。
    • 确保团队可以随时知道模型的训练数据，方便调试和迭代。

推理管道（Inference Pipeline）

作用：
推理管道的任务是使用特征库中的特征数据和模型仓库中的训练模型进行推理，并生成最终的预测结果。

主要职责：

  • 支持批量（Batch）或实时（Real-time）推理：
    • 批量模式：预测结果存入数据库（DB）。
    • 实时模式：预测结果直接返回给客户端。
  • 版本管理：特征、标签、模型的版本都是可追踪的，这意味着可以灵活地升级或回滚模型部署。
  • 动态调整模型与特征的连接关系：
    • 例如，$模型M_{1}$ 可能使用$特征f_1$、$特征f_2$ 和 $特征f_3$，而 $模型M_{2}$ 可能使用 $特征f_2$、$特征f_3$ 和 $特征f_4$
    • 通过版本管理，我们可以快速切换或调整特征与模型的映射关系。

设计 LLM Twin 的系统结构

需求分析

系统需要具备以下数据处理能力：

  • 数据采集：自动化并定期 从 小红书、知乎 和 GitHub（如果可行） 抓取数据。
  • 数据存储与标准化：统一格式化爬取的数据，并存入数据仓库（Data Warehouse）。
  • 数据清理：处理 噪声数据、重复数据和异常数据，确保数据质量。
  • 指令数据集（Instruction Dataset）构建：生成 用于微调 LLM 的训练数据集。
  • 数据向量化与存储：切分（Chunking）和嵌入（Embedding） 清理后的数据。存储向量化数据到向量数据库（Vector DB），以支持 RAG。

训练（Training）需求

  • 支持多种 LLM 微调：

    • 支持 不同规模的 LLM（7B、14B、30B、70B 参数）；
    • 能够基于 不同规模的指令数据集 进行微调。
    • 支持不同 LLM 模型类型（如 Mistral、Llama、GPT 之间切换）。

  • 实验管理：跟踪和比较 训练实验结果，优化模型效果。

  • 自动化训练：自动启动 训练任务，当新的 指令数据集可用 时触发训练流程。在部署前 测试潜在的 生产 LLM 候选模型，确保高质量推理能力。

推理（Inference）需求

  • REST API 接口：提供 REST API，允许客户端与 LLM Twin 交互。

  • 实时访问向量数据库（Vector DB）：支持 RAG，确保推理时可以实时检索相关知识数据。

  • 多模型推理能力：支持不同规模的 LLM 进行推理，适应不同业务场景。

  • 自动扩展（Auto-Scaling）：根据用户请求负载自动扩展推理服务，优化计算资源分配。

  • 自动化部署：通过评估机制，自动部署 通过测试的 LLM 版本，减少手动干预。

LLMOps 需求

  • 指令数据集管理：支持版本控制、数据 lineage 追踪 和 数据集复用，提高数据可管理性。

  • 模型管理：支持模型版本控制、模型 lineage 追踪 和 模型复用，便于模型管理和回溯。

  • 实验追踪：记录所有实验配置、结果和性能指标，确保可重复性和优化。

  • CI/CD + 持续训练（Continuous Training）：支持 CT/CI/CD，即持续训练（CT）、持续集成（CI）和持续部署（CD）。

  • 提示词和系统监控：监控**提示词（Prompt）**的表现，防止偏差。系统监控，确保 LLM 服务稳定运行。

如何使用 FTI 管道设计 LLM Twin 架构

我们将系统拆分为四个核心组件。除了 FTI 的三大核心管道（特征、训练、推理）外，我们还必须实现数据管道。

但在我们的场景下，我们的目标是在小团队中构建一个 MVP（最小可行产品），因此：我们必须同时实现数据收集和 FTI 管道。这种 端到端开发模式 在初创公司中非常常见，因为资源有限，无法分配独立团队。工程师需要 跨多个角色，视项目进度调整工作内容。即使未来团队扩展，理解端到端 ML 系统的架构仍然至关重要，有助于协同开发与优化。

screenshot_2025-03-11_10.13.07

数据收集管道（Data Collection Pipeline）

数据收集管道的任务是爬取你的个人数据，包括：小红书、知乎（帖子、文章），GitHub（代码）

在架构上，该管道遵循ETL（提取-加载-转换）模式，即：

  • 提取（Extract）：从社交媒体平台爬取数据；
  • 转换（Transform）：对数据进行标准化处理；
  • 加载（Load）：将数据存入数据仓库（NoSQL 数据库）。

[!NOTE]

为什么使用 NoSQL 作为数据仓库？

由于我们处理的是文本数据，它天然是非结构化的，因此 NoSQL 数据库（如 MongoDB）是最佳选择。尽管 MongoDB 不是传统的关系型数据库，但在我们的架构中，它将充当数据库的角色，因为：

  • 它存储了标准化的原始数据，这些数据由 ETL 管道收集并可以直接用于 ML 训练。
  • 它适合灵活存储和查询非结构化文本数据，便于下游管道访问和处理。

为了更好地处理数据，我们将爬取的数据分为三类：

  • 文章（Articles） → 知乎
  • 帖子（Posts） → 小红书
  • 代码（Code） → GitHub

我们希望抽象化数据来源，即：在 LLM 训练或推理时，数据的来源不重要；但为了溯源和引用，我们会将**原始 URL 作为元数据（metadata）**存储。

从 数据处理、微调训练（Fine-tuning）和 RAG（检索增强生成） 的角度来看，知道数据类别比知道来源更重要。

  • 例如，不同数据类型的切分（chunking）策略会有所不同：
    • 帖子（Post） 和 文章（Article） 的分割方式不同。
    • 代码（Code） 需要额外的解析和上下文理解。

按类别（category）而非来源（source）组织数据，能提高系统的扩展性：

  • 例如： 小红书的数据可以直接纳入Posts 类别，无须改动处理逻辑；而GitLab 的代码数据可以无缝集成到 Code 类别。

特征管道（Feature Pipeline）

特征管道的核心作用是从数据仓库获取原始数据（文章、帖子、代码），进行处理后存入特征存储（Feature Store）。FTI 设计模式的核心特点在此体现，但 LLM Twin 的特征管道有一些自定义特性：

  • 针对三种数据类型（文章、帖子、代码）分别进行不同的处理
  • 包含三大核心步骤（清洗、切分、嵌入），用于微调 LLM 和 RAG（检索增强生成）
  • 创建两种数据快照：
    • 清洗后数据（用于 LLM 微调）
    • 嵌入（Embedding）后数据（用于 RAG）：使用逻辑特征存储（Logical Feature Store），而非传统的专用特征存储。

逻辑特征存储：向量数据库（Vector DB）

在 RAG 系统中，向量数据库（Vector DB）是关键基础设施。向量数据库本质上是 NoSQL 数据库，可以按 ID 和集合名称（collection name） 访问数据点（datapoints）。我们可以查询 Vector DB 中的新数据，而不需要执行向量搜索（Vector Search）。处理后的数据会被封装为版本化、可追踪、可共享的处理后的数据（artifact）（关于 artifact 的细节将在第 2 章讨论）。

[!TIP]

What is an artifact in computer science?

To put it simply, an artifact is a by-product of software development. It’s anything that is created so a piece of software can be developed. This might include things like data models, diagrams, setup scripts — the list goes on.

（简单来说，Artifact 指的是一种软件开发的副产品。它指的是任何创建出来用以开发一套软件的一类东西，这其中也许包含了数据模型，图表，启动脚本等等。）

系统的其余部分将如何访问逻辑特征存储？**训练管道（Training Pipeline）**将指令数据集（Instruction Datasets）视为 artifact。推理管道（Inference Pipeline） 通过 向量搜索（Vector Search） 查询 Vector DB 以获取额外上下文信息。

对于我们的用例 LLM Twin，这已经足够了，因为：指令数据集（artifact）非常适合用于离线训练，而向量数据库是为在线访问而构建的，这是我们进行推理所需要的。

不过，在后续章节中，我们将解释如何**清理（cleaned）、分块（chunked）和嵌入（embedded）**这三个数据类别（文章、帖子和代码）。

训练管道（Training Pipeline）

训练管道的核心职责是 从特征存储（Feature Store）获取指令数据集（Instruct Dataset），微调 LLM，并将训练好的 LLM 权重存入模型注册表（Model Registry）。更具体地说：

  1. 触发训练：当逻辑特征存储中有新的指令数据集（artifact）可用时，我们将触发训练管道，使用工件并微调 LLM。
  2. 超参数优化：
    • 在初始阶段，数据科学团队负责这一步。他们通过自动或手动进行多次实验，以找到最适合的模型和超参数。
    • 为了比较和挑选最佳超参数集，记录所有有价值的东西，并在实验之间进行比较。
    • 最终，他们将挑选最佳超参数和微调后的 LLM，并将其作为 LLM 生产候选方案提出。然后将提议的 LLM 存储在模型注册表中。实验阶段结束后，我们存储并重复使用找到的最佳超参数。
  3. 如今，我们可以完全自动化训练过程，即持续训练（Continuous Training, CT）。我们的模块化设计使我们能够快速利用 ML 编排器来安排和触发不同的系统部分。例如，我们可以安排数据收集管道每周抓取数据。然后，当数据仓库中有新数据可用时，我们可以触发特征管道，当有新的指令数据集可用时，我们可以触发训练管道。

在将新模型推向生产之前，根据更严格的测试集对其进行评估至关重要，以确保最新候选模型比当前生产模型更好。如果此步骤通过，模型最终将被标记为已接受并部署到生产推理管道。即使在完全自动化的 ML 系统中，也建议在接受新的生产模型之前进行手动步骤。因此，在此阶段，专家会查看测试组件生成的报告。如果一切看起来都很好，它就会批准该模型，自动化可以继续。

关键技术问题

  • 如何设计一个 LLM 无关的训练管道？
  • 应该使用哪些微调技术？
  • 如何扩展微调算法，使其适用于不同规模的 LLM 和数据集？
  • 如何从多个实验中选取最优 LLM 作为生产候选？
  • 如何测试 LLM，以决定是否推送到生产环境？

在后面的章节中，我们将一一介绍解决方案。

推理管道（Inference Pipeline）

推理管道是 LLM 系统的最后一个核心组件，负责处理用户查询并返回答案。它连接 模型注册表（Model Registry） 和 逻辑特征存储（Logical Feature Store），用于加载微调后的 LLM 并执行 RAG（检索增强生成，Retrieval-Augmented Generation）。

推理流程

  1. 加载模型：从模型注册表加载已微调的 LLM；从逻辑特征存储访问向量数据库（Vector DB），用于 RAG 查询。
  2. 接收客户端请求：通过 REST API 接收用户查询；解析查询并生成 RAG 任务。
  3. 执行 RAG 以增强 LLM 生成能力：使用 向量数据库进行检索（Vector Search），找到相关外部信息；结合 LLM 进行答案生成，返回最终响应。
  4. 监控与分析：所有用户查询、RAG 处理的增强提示（Enriched Prompts）和生成结果，都会发送至提示监控系统（Prompt Monitoring System）。监控系统 分析、调试 模型输出，优化 LLM 行为。可根据 特定需求 触发警报，执行手动或自动调整。

FTI 设计与 LLM Twin 架构的最终思考

FTI（Feature-Training-Inference）模式 并不需要严格遵循，它的核心作用是帮助清晰地设计 ML（机器学习）系统。例如，我们的系统并没有使用传统的特征存储（Feature Store），而是选择了 基于向量数据库（Vector DB）和指令数据集（Artifacts） 的逻辑特征存储（Logical Feature Store），因为这样更简单且成本更低。重点是提供一个可版本化（Versioned）且可复用（Reusable） 的训练数据集，而不是形式上的标准化存储。

计算资源需求及可扩展性

  • 数据收集管道 & 特征管道：
    • 主要依赖 CPU 计算，对计算资源需求较低。
    • 基于 CPU & RAM 负载 水平扩展（Horizontal Scaling）
  • 训练管道：
    • 需要 强大的 GPU 计算能力 来加载和微调 LLM。
    • 通过 增加 GPU 资源 垂直扩展（Vertical Scaling）
  • 推理管道：
    • 计算需求介于数据管道和训练管道之间，需要 较强计算能力 以确保低延迟。
    • 基于 客户端请求数量 水平扩展（Horizontal Scaling）。

推理管道直接面向用户，因此必须严格测试，确保延迟符合预期，从而提供良好的用户体验。FTI 设计使得计算资源的分配变得灵活，我们可以为不同的组件选择最合适的计算架构。

引用

  1. Dowling, J. (2024a, July 11). From MLOps to ML Systems with Feature/Training/Inference Pipelines. Hopsworks. https://www.hopsworks.ai/post/mlops-to-ml-systems-with-fti-pipelines
  2. Dowling, J. (2024b, August 5). Modularity and Composability for AI Systems with AI Pipelines and Shared Storage. Hopsworks. https://www.hopsworks.ai/post/modularity-and-composability-for-ai-systems-with-ai-pipelines-and-shared-storage
  3. Joseph, M. (2024, August 23). The Taxonomy for Data Transformations in AI Systems. Hop- sworks. https://www.hopsworks.ai/post/a-taxonomy-for-data-transformations-in-ai-systems
  4. MLOps: Continuous delivery and automation pipelines in machine learning. (2024, August 28). Google Cloud. https://cloud.google.com/architecture/mlops-continuous-delivery-and-automation-pipelines-in-machine-learning
  5. Qwak. (2024a, June 2). CI/CD for Machine Learning in 2024: Best Practices to build, test, and Deploy | Infer. Medium. https://medium.com/infer-qwak/ci-cd-for-machine-learning-in-2024-best-practices-to-build-test-and-deploy-c4ad869824d2
  6. Qwak. (2024b, July 23). 5 Best Open Source Tools to build End-to-End MLOPs Pipeline in 2024. Medium. https://medium.com/infer-qwak/building-an-end-to-end-mlops-pipeline-with-open-source-tools-d8bacbf4184f
  7. Salama, K., Kazmierczak, J., & Schut, D. (2021). Practitioners guide to MLOPs: A framework for continuous delivery and automation of machine learning (1st ed.) [PDF]. Google Cloud. https://services.google.com/fh/files/misc/practitioners_guide_to_mlops_whitepaper.pdf

二、工具链与安装

[!CAUTION]

在手册中的对应章节介绍了使用的所有核心工具，特别是 LLM Twin 项目 的实现和部署所需的工具。因为个人偏好不同、对于工具链的选择亦有不同，所以在这里不做详细介绍，只会简要介绍手册作者推荐的工具链。

如果你熟悉这些工具，可以直接跳过本章。

在本章，我们不会深入讲解 LLM、RAG、MLOps 或 LLMOps 的概念，而是快速概览我们的技术栈和前置要求，避免后续章节重复讲解工具安装和选择原因。从 第 3 章开始，我们将正式进入 LLM Twin 的应用场景，并实现一个 ETL 数据收集流程，用于从互联网爬取数据。

本章内容概览

  • Python 生态工具：
    • 如何管理多个 Python 版本
    • 创建虚拟环境
    • 安装固定版本的依赖项
    • 如何在本地安装 LLM-Engineers-Handbook 代码库（如果你想尝试代码，地址如下）：GitHub Repo
  • MLOps & LLMOps 工具链：
    • 介绍通用 MLOps 工具（如模型注册表）
    • 深入了解LLM 相关工具（如 LLM 评估和 Prompt 监控工具）
    • 使用 ZenML 进行 ML 管道管理（ML 与 MLOps 的桥梁）
  • 数据库管理：
    • 介绍 NoSQL 和向量数据库 的使用
    • 如何使用 Docker 在本地运行所有组件
  • 云端环境准备（AWS）：
    • 创建 AWS 账户并获取访问密钥
    • 安装 & 配置 AWS CLI，以便程序化管理云资源
    • 了解 SageMaker 及其在开源 LLM 训练和部署中的作用

Python 生态工具链安装

任何 Python 项目都需要三个基本工具：python 解释器、依赖项管理和任务执行工具。Python 解释器会按预期执行您的 Python 项目。手册中的所有代码都使用 python 3.11.8 进行了测试。

1
2
python --version
# Python 3.11.8

[!TIP]

我们可以从此处下载 python 解释器：python.org

手册中推荐使用 poetry 工具来管理 python的依赖和虚拟环境，poethepoet 是 Poetry 插件，用于管理 CLI 任务，替代 Makefile、shell 脚本等；当然也可以使用其他的工具的组合替代，比如 conda(mini-conda) 和 pip。

MLOps & LLMOps 工具概览

本节简要介绍 MLOps（机器学习运维）和 LLMOps（大模型运维）工具，以及它们在 LLM Twin 项目中的作用。理论部分将在第 11 章深入讲解，而手册主要通过实战来展示这些工具的使用方式。

Hugging Face：模型注册库（Model Registry）

模型注册库 是一个 集中式存储，用于管理 ML 模型 的版本、元数据和性能指标。它在 MLOps 中起到关键作用：

  • 版本控制（Versioning）
  • 模型共享（Sharing）
  • 模型可追溯性（Traceability）
  • 集成 CI/CD 流水线（Continuous Deployment）

以下是一些常见的模型库：

  • Hugging Face Model Hub
    • 社区丰富，但中国大陆境内无法直接访问 ❌
    • 适用于：开源社区、NLP、大模型（LLMs）
    • 官网：https://huggingface.co
    • 特点：
      1. 大模型（LLM）生态支持，可存储 Transformer、Diffusion 等 AI 模型。
      2. 提供 可视化 UI，可管理模型版本、推理 demo（Spaces）。
      3. 易于与 PyTorch、TensorFlow、JAX 等框架集成。
      4. 适用于 团队协作 和 社区共享，支持 私有模型库。
  • AWS SageMaker Model Registry
    • 适用于： 企业级云端 MLOps，企业、金融、医疗等对安全性要求高的 MLOps 场景。
    • 官网： aws.amazon.com/sagemaker/
    • 特点：
      1. 与 AWS SageMaker 生态完美结合，支持 训练-注册-部署-监控 整套流程。
      2. 提供 自动化 CI/CD，模型更新后可自动部署。
      3. 高安全性，支持 IAM 权限管理，适用于企业级 ML 部署。
  • ModelScope
    • 阿里云推出的开源 AI 模型平台
    • 适用于：适用于 大模型（LLM）、计算机视觉（CV）、自然语言处理（NLP） 等 AI 任务
    • 官网：https://modelscope.cn/
    • 特点：
      1. 支持 1,000+ 预训练模型，包括大语言模型（LLM）、计算机视觉（CV）、语音处理（Speech）、多模态（Multimodal）等
      2. 一键调用 AI 工作流（Pipelines），快速搭建 AI 任务
      3. 免费在线体验，支持模型下载、本地部署、API 调用
对比项 魔搭 ModelScope Hugging Face Model Hub
适用地区 ✅ 中国大陆境内可用 🚫 需科学上网
模型数量 ⭐ 1,000+ ⭐ 10,000+
大模型支持 ✅ 通义千问、ChatGLM、Qwen ✅ LLaMA、GPT-3、Falcon
微调（Fine-tuning） ✅ LoRA, QLoRA, P-Tuning ✅ LoRA, PEFT, DPO, Unsloth
推理服务（API） ✅ 免费调用 ✅ 需付费
工作流（Pipelines） ✅ 一键执行 🚫 需手写代码，可与 LLMOps 集成
私有化部署 ✅ 企业可自建 ✅ 需自建服务器

[!NOTE]

手册中使用的 Hugging Face 模型：

ZenML：MLOps 工作流编排器

[!NOTE]

TODO

CometML：可视化实验跟踪

[!NOTE]

TODO

Opik：评估、测试和监控大型语言模型

[!NOTE]

TODO

非结构化数据库与向量数据库

MongoDB：NoSQL

[!NOTE]

TODO

Qdrant：向量数据库

[!NOTE]

TODO

云端环境准备（AWS）

[!NOTE]

TODO

三、数据工程

本章将深入探讨 LLM Twin 项目，学习如何设计和实现数据收集流水线，以获取用于 LLM 任务（如微调或推理）的原始数据。由于本书并非专门介绍数据工程，因此本章内容将保持精简，仅关注收集必要原始数据的关键部分。从第 4 章开始，我们将重点讨论 LLM 和生成式 AI，深入研究其理论和具体实现细节。

在处理项目或研究时，我们通常会使用一个静态数据集。但在 LLM Twin 项目中，我们希望模拟真实世界的场景，在其中主动收集和整理数据。因此，构建数据流水线将帮助我们了解端到端机器学习项目的工作方式。本章将讲解如何设计和实现 ETL（提取、转换、加载）流水线，从 社交平台爬取数据，并将其存储到 MongoDB 数据库。我们将介绍各种爬取方法，标准化数据，并将其加载到数据仓库中。

本章的主要内容包括：

  1. 设计数据收集流水线
    • 介绍 LLM Twin 的数据收集架构
    • 解析 ETL 流水线的设计
  2. 实现数据收集流水线
    • 使用 ZenML 作为流程编排工具
    • 构建爬虫，并实现 调度层（根据 URL 域名实例化对应爬虫类）
    • 按最佳软件开发实践，开发每个爬虫模块
  3. 数据仓库管理
    • 在 MongoDB 之上构建数据层，统一管理文档结构
    • 查询并交互数据

最后，我们将学习如何使用 ZenML 运行数据收集流水线，并查询 MongoDB 中的数据。

设计数据收集管道

在深入实施之前，我们必须了解 LLM Twin 的数据收集 ETL 架构，如下图所示。我们需要探究从哪些平台抓取数据，以及如何设计数据结构和流程。但是，第一步是了解我们的数据收集管道如何映射到 ETL 流程。

screenshot_2025-03-11_15.25.00

ETL 管道涉及三个基本步骤：

  1. 我们从各种来源提取数据。我们将从内容平台抓取数据以收集原始数据。
  2. 我们通过清理和标准化这些数据，将其转换为适合存储和分析的一致格式。
  3. 我们将转换后的数据加载到数据仓库或数据库中。

对于我们的项目，我们使用 MongoDB 作为我们的 NoSQL 数据仓库。虽然这不是标准方法，但我们很快就会解释这种选择背后的原因。
我们想要设计一个 ETL 管道，输入一个用户和一个链接列表作为输入。之后，它会单独抓取每个链接，标准化收集到的内容，并将其保存在 MongoDB 数据仓库中该特定作者下。

数据收集流水线的输入和输出

  • 输入：用户（作者）及其提供的一组链接。
  • 输出：存储在 MongoDB 数据仓库中的原始文档列表。

[!NOTE]

我们将交替使用用户和作者，因为在 ETL 管道的大多数情况下，用户是提取内容的作者。但是，在数据仓库中，我们只有一个用户集合。

screenshot_2025-03-11_15.38.12ETL 管道将检测每个链接的域名，并根据该域调用专门的爬虫。我们为三个不同的数据类别实现了四个不同的爬虫，如上图所示。首先，我们收集的所有文档都可以归结为文章、存储库（或代码)和帖子。数据来自哪里并不重要。我们主要对文档的格式感兴趣。在大多数情况下，我们必须以不同的方式处理这些数据类别。因此，我们为每个实体创建了一个不同的域实体，每个实体在 MongoDB 中都有自己的类和集合。当我们将源 URL 保存在文档的元数据中时，我们仍然会知道它的来源，并可以在 GenAI 用例中引用它。

爬虫类型 目标数据源 输出文档类型 主要步骤
Medium 爬虫 Medium 文章 文章（Article） 登录 Medium → 爬取 HTML → 解析并清理文本 → 存入数据库
通用文章爬虫 Substack / 个人博客等 文章（Article） 爬取 HTML → 解析并清理文本 → 存入数据库
GitHub 爬虫 GitHub 仓库 代码仓库（Repository） 克隆代码仓库 → 解析文件树 → 处理代码文件 → 存入数据库
LinkedIn 爬虫 LinkedIn 个人动态 帖子（Post） 登录 LinkedIn → 爬取用户动态 → 解析 HTML → 存入数据库

在下一节中，我们将详细研究每个爬虫的实现。现在，请注意，每个爬虫都以特定方式访问特定平台或站点并从中提取 HTML。之后，所有爬虫都会解析 HTML，从中提取文本，并对其进行清理和规范化，以便可以将其存储在同一个接口下的数据仓库中。

通过将所有收集的数据减少到三种数据类别，文章（Article）、代码仓库（Repository）和帖子（Post），而不是为每个新数据源创建新的数据类别，我们可以轻松地将此架构扩展到多个数据源，而无需付出太多重复适配工作。例如，如果我们想开始从 X 收集数据，我们只需要实现一个输出帖子文档的新爬虫，仅此而已。其余代码将保持不变。否则，如果我们在类和文档结构中引入源维度，我们将不得不向所有下游层添加代码以支持任何新数据源。例如，我们必须为每个新源实现一个新的文档类，并调整功能管道以支持它。

对于我们的概念验证，抓取几百个文档就足够了，但如果我们想将其扩展到实际产品，我们可能需要更多数据源来抓取。LLM 需要大量数据，通常需要数千个文档才能获得理想的结果，而不仅仅是几百个文档。但在许多项目中，实现一个不是最准确的端到端项目版本并在以后对其进行迭代是一种很好的策略。因此，通过使用这种架构，可以在未来的迭代中轻松添加更多数据源以收集更大的数据集。下一章将介绍有关 LLM 微调和数据集大小的更多信息。

ETL 过程如何连接到特征管道？特征管道从 MongoDB 数据仓库中提取原始数据，进一步清理，将其处理为特征，并将其存储在 Qdrant 向量数据库中，以使 LLM 训练和推理管道可以访问它。第 4 章提供了有关特征管道的更多信息。ETL 过程独立于特征管道。这两个管道严格通过 MongoDB 数据仓库相互通信。因此，数据收集管道可以为 MongoDB 写入数据，而功能管道可以独立地按照不同的时间表从中读取数据。

为什么我们使用 MongoDB 作为数据仓库？

  • 适用于小规模数据：本项目的文档量较小，MongoDB 能够很好地处理。
  • 适合非结构化文本：爬取的数据主要是非结构化文本，MongoDB 不强制模式（Schema），使开发更灵活。
  • 易用性：MongoDB 提供直观的 Python SDK，官方提供 Docker 镜像 和 云端免费层，适合本项目的 PoC（概念验证）。
  • 未来可扩展性：如果数据量增大（如达到百万级别），可以切换到 Snowflake 或 BigQuery 这样的专用数据仓库。

实现 LLM Twin 的数据收集流水线

LLM Twin 项目的每个流水线的入口都是一个 ZenML 流水线，且可以通过 YAML 文件在运行时进行配置，并通过 ZenML 生态系统执行。因此，我们从 ZenML 的 digital_data_etl 流水线开始，仔细分析其实现方式。你会注意到，这正是我们在第 2 章中用来演示 ZenML 的示例流水线。不过，这次我们将深入探讨其实现，并解释数据收集的具体细节。理解流水线的工作原理后，我们将探讨每个爬虫的实现，它们分别用于从不同网站收集数据并存储到 MongoDB 数据仓库中。

[!CAUTION]

这里手册的作者使用 ZenML 作为 LLMOps 的整体框架，使用 zenml.pipeline 作为流水线管道的具体实现。这里我们将重点放在流水线的工程方法，而非具体的实现中（框架无关）。如果读者需要了解 ZenML 框架内的相关实现，请在手册中第 65 页查找详情。

流水线与步骤

在流水线的实现中，它的输入是用户的全名和一组链接，这些链接将由该用户（即该链接的内容作者）进行爬取。在函数内，我们调用了两个步骤：首先，我们根据全名查找用户。接着，我们遍历所有链接并逐个爬取。

1
2
3
4
5
@pipeline
def digital_data_etl(user_full_name: str, links: list[str]) -> str:
    user = get_or_create_user(user_full_name)				# step1
    last_step = crawl_links(user=user, links=links)	# step2
    return last_step.invocation_id

接下来，我们将分别探讨 get_or_create_user 和 crawl_links 这两个步骤：

  1. 以用户的全名作为输入，并尝试从 MongoDB 数据库中查找该用户，如果用户不存在，则创建一个新的用户；

    1
    2
    3
    4
    5
    6
    7
    8
    @step
    def get_or_create_user(user_full_name: str) -> Annotated[UserDocument, "user"]:
        logger.info(f"Getting or creating user: {user_full_name}")
        first_name, last_name = utils.split_user_full_name(user_full_name)
        user = UserDocument.get_or_create(first_name=first_name, last_name=last_name)
        step_context = get_step_context()
        step_context.add_output_metadata(output_name="user", metadata=_get_metadata(user_full_name, user))	# matadata
        return user

    在这里，我们还定义了一个辅助函数 _get_metadata()，它构建了一个包含查询参数和检索到的用户信息的字典，这些信息将作为元数据添加到用户输出 artifact ：

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    def _get_metadata(user_full_name: str, user: UserDocument) -> dict:
        return {
            "query": {
                "user_full_name": user_full_name,
            },
            "retrieved": {
                "user_id": str(user.id),
                "first_name": user.first_name,
                "last_name": user.last_name,
            },
        }

  2. 接下来是 crawl_links 步骤，它用于收集提供的链接中的数据。在此函数中，我们初始化了一个爬虫分发器 CrawlerDispatcher，并配置它以处理特定的域名，如 GitHub 等：

    1
    2
    3
    4
    5
    @step
    def crawl_links(user: UserDocument, links: list[str]) -> Annotated[list[str], "crawled_links"]:
        dispatcher = CrawlerDispatcher.build().register_github()
        logger.info(f"Starting to crawl {len(links)} link(s).")

    然后，该函数初始化了存储输出元数据的变量，并计数成功的爬取次数。它遍历每个链接，尝试爬取并提取数据，并更新成功的爬取次数和链接的元数据：

    1
    2
    3
    4
    5
    6
    7
    metadata = {}
    successfull_crawls = 0
    for link in tqdm(links):
        successfull_crawl, crawled_domain = _crawl_link(dispatcher, link, user)
        successfull_crawls += successfull_crawl
        metadata = _add_to_metadata(metadata, crawled_domain, successfull_crawl)

    处理完所有链接后，函数将累积的元数据添加到输出 artifact 中：

    1
    2
    3
    4
    step_context = get_step_context()
    step_context.add_output_metadata(output_name="crawled_links", metadata=metadata)
    logger.info(f"Successfully crawled {successfull_crawls} / {len(links)} links.")
    return links

    在上述的函数中有两个辅助函数 _crawl_link 和_add_to_metadata。_crawl_link 尝试使用合适的爬虫来提取每个链接的信息，处理任何可能发生的异常，并返回一个元组，表示爬取是否成功以及链接的域名：

    1
    2
    3
    4
    5
    6
    7
    8
    9
    def _crawl_link(dispatcher: CrawlerDispatcher, link: str, user: UserDocument) -> tuple[bool, str]:
        crawler = dispatcher.get_crawler(link)
        crawler_domain = urlparse(link).netloc
        try:
            crawler.extract(link=link, user=user)
            return (True, crawler_domain)
        except Exception as e:
            logger.error(f"An error occurred while crawling: {e!s}")
            return (False, crawler_domain)

    _add_to_metadata 用于更新元数据字典，记录每个域名的爬取成功与总数：

    1
    2
    3
    4
    5
    6
    def _add_to_metadata(metadata: dict, domain: str, successfull_crawl: bool) -> dict:
        if domain not in metadata:
            metadata[domain] = {}
        metadata[domain]["successful"] = metadata.get(domain, {}).get("successful", 0) + successfull_crawl
        metadata[domain]["total"] = metadata.get(domain, {}).get("total", 0) + 1
        return metadata

爬虫分发器 CrawlerDispatcher 的实现

正如上面提到的，CrawlerDispatcher 类会根据每个链接的域名来确定该使用哪个爬虫。

screenshot_2025-03-12_11.12.34

爬虫的提取逻辑被封装在 extract() 方法中。例如，如果提供的链接属于 https://github.com，它会创建一个 GithubCrawler 实例来爬取该平台的数据。接下来，我们深入探讨 CrawlerDispatcher 的实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import BaseCrawler, GithubCrawler, CustomArticleCrawler

class CrawlerDispatcher:
  def __init__(self) -> None:
    # 存储域名（如 guthub.com）到 爬虫类（如 GithubCrawler）的映射关系。
    self._crawlers = Dict[str, Type[BaseCrawler]] = {}
    
	@classmethod
  def build(cls) -> "CrawlerDispatcher":
    dispatcher = cls()
    return dispatcher
  
  def register_github(self) -> "CrawlerDispatcher":
    """ 注册一个新的爬虫类型 """
    self.register("https://github.com", GithubCrawler)
    return self
  
  def register(self, domain:str, crawler:type[BaseCrawler]) -> None:
    parsed_domain = urlparse(domain)
    domain = parsed_domain.netloc
    self._crawlers[r"https://(www\.?{}/*)".format(re.eacape(domain))] = crawler
    
	def get_crawler(self, url:str) -> BaseCrawler:
    """ 根据 URL 解析出域名，并实例化正确的爬虫 """
    for pattern, crawler in self._crawlers.items():
      if re.match(pattern, url):
        return crawler()
      else:
        logger.warning(f"No crawler found for {url}. Defaulting to CustomArticleCrawler")
        return CustomArticleCrawler()

爬虫（Crawlers）

在深入探讨各个爬虫的具体实现之前，我们需要先介绍它们的 基类 BaseCrawler。

基类定义了一个统一的接口，使得所有爬虫都遵循相同的结构。我们之所以能够实现分发层（Dispatcher Layer），正是因为所有爬虫都遵循相同的方法签名。

基类 BaseCrawler

现在，我们来看 BaseCrawler 的实现，它定义了所有爬虫必须实现的方法：

1
2
3
4
5
6
7
8
9
10
from abc import ABC, abstractmethod

class BaseCrawler(ABC):
  """ 所有爬虫的基类，定义了通用接口 """
  model: type[NoSQLBaseDocument]
  
  @abstractmethod
  def extract(self, link: str, **kwargs):
    """ 爬取数据并存入数据库 """
    pass

解析：

  1. BaseCrawler 继承 ABC（抽象基类），确保不能直接实例化它。
  2. extract() 是一个 抽象方法，所有具体爬虫都必须实现它。

这样，我们可以在不更改 CrawlerDispatcher 代码的情况下，轻松 扩展新爬虫。

拓展爬虫：基于 Selenium 的爬虫基类

在 BaseCrawler 基础上，我们进一步扩展出了 BaseSeleniumCrawler，用于 自动化浏览器操作，以便爬取需要动态加载或需要登录的网站（如 Medium、LinkedIn）。

[!NOTE]

为什么使用 Selenium？

  1. 支持动态内容加载：许多现代网站使用 JavaScript 加载内容，普通的 HTTP 请求（如 requests）无法抓取完整数据，而 Selenium 可以模拟用户操作，触发 JavaScript 代码。
  2. 支持登录：对于 需要用户认证 的网站（如 LinkedIn），Selenium 可以模拟用户输入账号密码并自动登录。
  3. 支持交互：Selenium 可以执行点击、滚动、表单填写等操作，使其更适用于复杂网页的爬取。
  4. 使用 Selenium-based 的爬虫，必须先在本机安装 Chrome （或者其他基于 chromium 内核的浏览器）

简述

  • 代码使用 Selenium 和 ChromeDriver 初始化程序设置 Web 爬取所需的导入和配置
  • chromedriver_autoinstaller 确保安装适当版本的 ChromeDriver 并将其添加到系统路径，从而与已安装的 Google Chrome 浏览器版本（或其他基于 Chromium 的浏览器）保持兼容性。
  • Selenium 将使用 ChromeDriver 与浏览器通信并打开无头会话模式，我们可以在其中以编程方式操作浏览器以访问各种 URL、单击特定元素（例如按钮）或滚动浏览新闻源。
  • 使用 chromedriver_autoinstaller，我们确保始终安装与我们机器的 Chrome 浏览器版本匹配的正确 ChromeDriver 版本。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import time
from tempfile import mkdtemp
import chromedriver_autoinstaller
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from llm_engineering.domain.documents import NoSQLBaseDocument
# Check if the current version of chromedriver exists
# and if it doesn't exist, download it automatically,
# then add chromedriver to path
chromedriver_autoinstaller.install()

class BaseSeleniumCrawler(BaseCrawler, ABC):
  def __init__(self, scroll_limit:int=5) -> None:
    chromedriver_autoinstaller.install()		# 自动安装 ChromeDriver 并初始化
    
    options = webdriver.ChromeOptions()
		options.add_argument("--no-sandbox")
		options.add_argument("--headless=new")	# 无头模式，浏览器界面不可见
    options.add_argument("--disable-dev-shm-usage")
    options.add_argument("--log-level=3")
    options.add_argument("--disable-popup-blocking")
    options.add_argument("--disable-notifications")
    options.add_argument("--disable-extensions")
		options.add_argument("--disable-background-networking")
		options.add_argument("--ignore-certificate-errors")
		options.add_argument(f"--user-data-dir={mkdtemp()}")
		options.add_argument(f"--data-path={mkdtemp()}")
		options.add_argument(f"--disk-cache-dir={mkdtemp()}")
		options.add_argument("--remote-debugging-port=9226")
    
    self.set_extra_driver_options(options)	# 子类可以自定义额外的选项
		self.scroll_limit = scroll_limit
		self.driver = webdriver.Chrome(options=options)
    
	def set_extra_driver_options(self, options: Options) -> None:
    """ 子类可以重写此方法，添加额外的浏览器选项 """
    pass
  
  def login(self) -> None:
    """ 子类可以重写此方法，实现登录逻辑 """
    pass

  def scroll_page(self) -> None:
    """ 滚动页面，直到达到滚动限制 """
    current_scroll = 0
    last_height = self.driver.execute_script("return document.body.scrollHeight")
    while True:
      self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
      time.sleep(5)
      new_height = self.driver.execute_script("return document.body.scrollHeight")
      if new_height == last_height or (self.scroll_limit and current_scroll >= self.scroll_limit):
        break
        last_height = new_height
        current_scroll += 1

至此，我们已经定义了两个爬虫基类：BaseCrawler 和 BaseSeleniumCrawler。下一步我们将通过继承这两个基类，从而实现具体的爬虫：

  • GithubCrawler(BaseCrawler)
  • CustomArticleCrawler(BaseCrawler)
  • MediumCrawler(BaseSeleniumCrawler)

Github 爬虫类 GithubCrawler

GitHubCrawler 类继承自 BaseCrawler，旨在爬取 GitHub 仓库的内容。与其他需要 Selenium 的爬虫不同，由于 GitHub 支持 Git 的 clone 功能，爬虫无需通过浏览器模拟登录或操作，而是直接利用 Git 将仓库克隆到本地进行内容提取。下面是这个类的详细实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class GithubCrawler(BaseCrawler):
  model = RepositoryDocument	# MongoDB 中定义的代码仓库类型的内容
  def __init__(self, ignore=(".git", ".toml", ".lock", ".png")) -> None:
    '''ignore 参数定义了要忽略的标准文件和目录类型。例如 .git, .toml, .lock, .png 等。这样可以确保爬虫在处理 GitHub 仓库时不会抓取这些不需要的文件。'''
    super().__init__()
    self._ignore = ignore		# 文件过滤规则，忽略标准文件和目录
    
	@override
	def extract(self, link: str, **kwargs) -> None:
    # 检查仓库是否已经在数据库中存在，避免重复存储
    old_model = self.model.find(link=link)
    if old_model is not None:
      logger.info(f"Repository already exists in the database: {link}")
      return
    
    try:
      logger.info(f"Starting scrapping GitHub repository: {link}")
      repo_name = link.rstrip("/").split("/")[-1]
      local_temp = tempfile.mkdtemp()
      os.chdir(local_temp)
      
      # 自进程执行克隆 GitHub 仓库
      subprocess.run(["git", "clone", link])
      
      # 获取克隆后的仓库路径
      repo_path = os.path.join(local_temp, os.listdir(local_temp)[0])  # 获取仓库目录路径
      tree = {}  # 用于存储仓库文件内容的字典
      
      # 遍历仓库目录，跳过忽略的文件和目录
      for root, _, files in os.walk(repo_path):
      	dir = root.replace(repo_path, "").lstrip("/")
      	if dir.startswith(self._ignore):
      		continue
      	
      	for file in files:
      		if file.endswith(self._ignore):
      			continue
      		file_path = os.path.join(dir, file)
      		# 读取文件内容，去除空格并存储
      		with open(os.path.join(root, file), "r", errors="ignore") as f:
      			tree[file_path] = f.read().replace(" ", "")
    
    # 创建 RepositoryDocument 实例并保存到 MongoDB
    	user = kwargs["user"]
    	instance = self.model(
        content=tree,
        name=repo_name,
        link=link,
        platform="github",
        author_id=user.id,
        author_full_name=user.full_name,
      )
      instance.save()
      
		except Exception:
      raise  # 若发生错误，抛出异常

		finally:
      # 确保删除临时目录以释放资源
      shutil.rmtree(local_temp)
      logger.info(f"Finished scraping GitHub repository: {link}")

CustomArticleCrawler 类

CustomArticleCrawler 类采用不同的方法来从互联网上收集数据。它利用 AsyncHtmlLoader 类来加载链接的完整 HTML 内容，再通过 Html2TextTransformer 类提取该 HTML 中的文本内容。这两个类由 langchain_communityPython 包提供，以下是相关模块的导入：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
from urllib.parse import urlparse
from langchain_community.document_loaders import AsyncHtmlLoader
from langchain_community.document_transformers.html2text import Html2TextTransformer
from loguru import logger
from domain.documents import ArticleDocument

import BaseCrawler
class CustomArticleCrawler(BaseCrawler):
  model = ArticleDocument	# MongoDB 中定义的文章类型的内容
  
  def extract(self, link: str, **kwargs) -> None:
    # 检查文章是否已经存在于数据库中，避免重复爬取
    old_model = self.model.find(link=link)
    if old_model is not None:
      logger.info(f"Article already exists in the database: {link}")
      return
    
    # 如果文章不存在，进行爬取
    logger.info(f"Starting scraping article: {link}")
    loader = AsyncHtmlLoader([link])  # 使用 AsyncHtmlLoader 加载 HTML 内容
    docs = loader.load()  # 加载 HTML
    html2text = Html2TextTransformer()  # 使用 Html2TextTransformer 提取文本
    docs_transformed = html2text.transform_documents(docs)  # 转换为文本
    doc_transformed = docs_transformed[0]  # 获取第一个转换后的文档
    
    # 提取文章内容和相关元数据
    content = {
      "Title": doc_transformed.metadata.get("title"),
      "Subtitle": doc_transformed.metadata.get("description"),
      "Content": doc_transformed.page_content,
      "language": doc_transformed.metadata.get("language"),
    }
    
    # 解析 URL 来确定平台（或域名）
    parsed_url = urlparse(link)
    platform = parsed_url.netloc

    # 创建文章模型实例，并保存到 MongoDB 数据库
    user = kwargs["user"]
    instance = self.model(
      content=content,
      link=link,
			platform=platform,
			author_id=user.id,
      author_full_name=user.full_name,
    )
    instance.save()
    logger.info(f"Finished scraping custom article: {link}")

解析

  1. 初始化方法:
    • CustomArticleCrawler 类继承自 BaseCrawler，并使用 ArticleDocument 作为数据模型来存储爬取的文章信息。
  2. extract() 方法:
    • 检查重复：首先检查文章是否已经存在于数据库中，如果存在，则不再重复爬取该文章。
    • 加载 HTML：如果文章不存在，使用 AsyncHtmlLoader 类加载提供的链接的 HTML 内容。
    • 提取文本：使用 Html2TextTransformer 类将 HTML 转换为纯文本，并返回一个包含文档内容的列表。我们只关心列表中的第一个文档。
    • 提取内容：从转换后的文档中提取标题、子标题、正文和语言等元数据。
    • 解析平台：通过解析 URL 来确定文章所属的平台或域名。
    • 保存数据：使用 ArticleDocument 类创建一个文章实例，并将提取的内容保存到 MongoDB 数据库中。
  3. LangChain 的应用：
    • AsyncHtmlLoader 和 Html2TextTransformer 类遵循 LangChain 的范式，这使得我们能够快速实现爬取和转换功能。虽然它们非常适用于大多数场景，但由于灵活性较差，难以进行深度定制，因此在生产环境中可能不适合所有情况，无法对某些特定网站或内容结构进行精细化控制。

NoSQL 数据仓库文档

我们之前设计了三个文档类来结构化我们的数据类别。这些类定义了文档所需的具体属性，例如内容、作者和来源链接。最佳实践是将数据结构化为类，而不是字典，因为我们对每个项目期望的属性更为详尽，从而减少运行时错误。例如，当从 Python 字典中访问值时，我们无法确保它存在或其类型是否正确。通过将数据项封装在类中，我们可以确保每个属性都符合预期。

通过利用像 Pydantic 这样的 Python 包，我们可以获得开箱即用的类型验证，确保数据集的一致性。因此，我们将数据类别建模为以下文档类，这些类已经在代码中使用过：

  • ArticleDocument 类
  • PostDocument 类
  • RepositoryDocument 类

这些不仅仅是简单的 Python 数据类或 Pydantic 模型。它们支持在 MongoDB 数据仓库上进行读写操作。为了将读写功能注入到所有文档类中，并避免重复代码，我们使用了对象-文档映射（ODM）软件模式，它基于对象关系映射（ORM）模式。因此，接下来我们将首先探讨 ORM，然后转向 ODM，最后深入研究我们的自定义 ODM 实现和文档类。

ODM 中间件

在讨论软件模式之前，让我们了解一下 ORM。ORM 是一种技术，允许您使用面向对象的范式查询和操作数据库中的数据。无需编写 SQL 或特定 API 的查询，您可以将所有复杂性封装到一个 ORM 类中，该类知道如何处理所有数据库操作，通常是 CRUD 操作。因此，使用 ORM 可以避免手动处理数据库操作，并减少手动编写样板代码的需要。ORM 与 SQL 数据库（如 PostgreSQL 或 MySQL）交互。

大多数现代 Python 应用程序在与数据库交互时都会使用 ORM。尽管 SQL 在数据领域仍然是一种流行的选择，但在 Python 后端组件中，您很少看到原始的 SQL 查询。最流行的 Python ORM 是 SQLAlchemy（https://www.sqlalchemy.org/）。此外，随着 FastAPI 的兴起，SQLModel（https://github.com/fastapi/sqlmodel）成为了一个常见选择，它是 SQLAlchemy 的一个封装，使其与 FastAPI 的集成更加简便。

ODM 模式与 ORM 非常相似，但它是针对 NoSQL 数据库（如 MongoDB）和无结构的集合操作，而不是 SQL 数据库和表。当我们使用 NoSQL 数据库时，数据结构主要集中在集合中，存储的是类似 JSON 的文档，而不是表中的行。

总之，ODM 简化了与基于文档的 NoSQL 数据库的交互，并将面向对象的代码映射到类似 JSON 的文档。接下来，我们将实现一个轻量级的 ODM 模块，基于 MongoDB，帮助我们深入理解 ODM 的工作原理。

接下来，我们定义了一个类型变量 T，它绑定到 NoSQLBaseDocument 类。该变量利用 Python 的泛型模块，允许我们将类的类型进行泛化。例如，在实现 ArticleDocument 类时，T 所有使用的地方都会被替换为 ArticleDocument 类型。

NoSQLBaseDocument 类继承了 Pydantic 的 BaseModel、Python 的 Generic 和 ABC 类（使其成为抽象基类），它是我们的基础 ODM 类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
import uuid
from abc import ABC
from typing import Generic, Type, TypeVar
from loguru import logger
from pydantic import UUID4, BaseModel, Field
from pymongo import errors
from llm_engineering.domain.exceptions import ImproperlyConfigured
from llm_engineering.infrastructure.db.mongo import connection
from llm_engineering.settings import settings
_database = connection.get_database(settings.DATABASE_NAME)

T = TypeVar("T", bound="NoSQLBaseDocument")

class NoSQLBaseDocument(BaseModel, Generic[T], ABC):
  '''实现了 __eq__ 和 __hash__ 方法，允许比较实例并将其用于散列集合（如集合）或基于其唯一 id 属性作为字典键'''
  id: UUID4 = Field(default_factory=uuid.uuid4)
  
  def __eq__(self, value: object) -> bool:
    if not isinstance(value, self.__class__):
      return False
    return self.id == value.id
  
  def __hash__(self) -> int:
    return hash(self.id)

  @classmethod
  '''将从 MongoDB 获取的字典转换为类实例'''
	def from_mongo(cls: Type[T], data: dict) -> T:
    if not data:
      raise ValueError("Data is empty.")
		id = data.pop("_id")
    return cls(**dict(data, id=id))
    
  def to_mongo(self: T, **kwargs) -> dict:
    '''将模型实例转换为适合 MongoDB 插入的字典'''
    exclude_unset = kwargs.pop("exclude_unset", False)
    by_alias = kwargs.pop("by_alias", True)
    parsed = self.model_dump(exclude_unset=exclude_unset, by_alias=by_alias, **kwargs)
    if "_id" not in parsed and "id" in parsed:
      parsed["_id"] = str(parsed.pop("id"))
    for key, value in parsed.items():
      if isinstance(value, uuid.UUID):
        parsed[key] = str(value)
    return parsed
  
  def save(self: T, **kwargs) -> T | None:
    '''利用上面描述的 to_mongo() 方法将实例转换为与 MongoDB 兼容的文档，并尝试将其插入数据库，处理可能发生的任何写入错误'''
    collection = _database[self.get_collection_name()]
    try:
      collection.insert_one(self.to_mongo(**kwargs))
      return self
    except errors.WriteError:
      logger.exception("Failed to insert document.")
      return None
    
	@classmethod
  def get_or_create(cls: Type[T], **filter_options) -> T:
    '''如果找到匹配的文档，则将其转换为类的实例。如果没有找到，则创建一个新实例，并使用过滤选项作为其初始数据并保存到数据库'''
    collection = _database[cls.get_collection_name()]
    try:
      instance = collection.find_one(filter_options)
      if instance:
        return cls.from_mongo(instance)
      new_instance = cls(**filter_options)
      new_instance = new_instance.save()
      return new_instance
    except errors.OperationFailure:
      logger.exception(f"Failed to retrieve document with filter options: {filter_options}")
      raise
      
	@classmethod
  def bulk_insert(cls: Type[T], documents: list[T], **kwargs) -> bool:
    '''批量写入文档以提高效率'''
    collection = _database[cls.get_collection_name()]
    try:
      collection.insert_many([doc.to_mongo(**kwargs) for doc in documents])
        return True
    except (errors.WriteError, errors.BulkWriteError):
      logger.error(f"Failed to insert documents of type {cls.__name__}")
      return False
    
	@classmethod
  def find(cls: Type[T], **filter_options) -> T | None:
    '''根据filter_options来查找数据库中的文档'''
    collection = _database[cls.get_collection_name()]
    try:
      instance = collection.find_one(filter_options)
      if instance:
        return cls.from_mongo(instance)
      return None
    except errors.OperationFailure:
      logger.error("Failed to retrieve document.")
      return None
    
	@classmethod
  def bulk_find(cls: Type[T], **filter_options) -> list[T]:
    '''检索与过滤选项匹配的多个文档。将每个检索到的 MongoDB 文档转换为模型实例，并将它们收集到列表中'''
    collection = _database[cls.get_collection_name()]
    try:
      instances = collection.find(filter_options)
      return [document for instance in instances if (document := cls.from_mongo(instance)) is not None]
    except errors.OperationFailure:
      logger.error("Failed to retrieve document.")
      return []
    
	@classmethod
  def get_collection_name(cls: Type[T]) -> str:
    '''确定与该类关联的 MongoDB 集合的名称。它期望该类具有嵌套的 Settings 类，该类具有指定集合名称的 name 属性'''
    if not hasattr(cls, "Settings") or not hasattr(cls.Settings, "name"):
      raise ImproperlyConfigured(
        "Document should define an Settings configuration class with the name of the collection.")
    return cls.Settings.name

数据类别和用户文档类

最后，我们看一下继承自 NoSQLBaseDocument 基类的子类的实现。这些是定义我们数据类别的具体类。你已经在本章中看到这些类，它们用于爬虫类中的文章、仓库和帖子。

首先定义了一个 enum 类，将所有数据类别类型集中管理。这些变量将作为常量，用于配置本书中的所有 ODM 类。

1
2
3
4
5
6
7
8
9
10
11
12
13
from enum import StrEnum

class DataCategory(StrEnum):
  PROMPT = "prompt"
  QUERIES = "queries"
  INSTRUCT_DATASET_SAMPLES = "instruct_dataset_samples"
  INSTRUCT_DATASET = "instruct_dataset"
	PREFERENCE_DATASET_SAMPLES = "preference_dataset_samples"
	PREFERENCE_DATASET = "preference_dataset"
	POSTS = "posts"
	ARTICLES = "articles"
	REPOSITORIES = "repositories"
  USERS = "users"

Document 类被引入作为其他文档类的抽象基类，基于 NoSQLBaseDocument ODM 类。它包括通用的属性，如内容、平台和作者信息，为所有继承自它的文档提供了标准化结构：

1
2
3
4
5
class Document(NoSQLBaseDocument, ABC):
	content: dict
	platform: str
	author_id: UUID4 = Field(alias="author_id")
	author_full_name: str = Field(alias="author_full_name")

具体的文档类型通过继承 Document 类来定义。RepositoryDocument、PostDocument 和 ArticleDocument 类分别代表不同的数据类别，每个类别都有独特的字段和设置，指定它们在数据库中的集合名称。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class RepositoryDocument(Document):
  name: str
  link: str
  class Settings:
    name = DataCategory.REPOSITORIES

class PostDocument(Document):
  image: Optional[str] = None
  link: str | None = None
  class Settings:
    name = DataCategory.POSTS

class ArticleDocument(Document):
  link: str
  class Settings:
    name = DataCategory.ARTICLES

最后，我们定义了 UserDocument 类，用于存储和查询所有来自 LLM Twin 项目的用户：

1
2
3
4
5
6
7
8
9
class UserDocument(NoSQLBaseDocument):
  first_name: str
  last_name: str
  class Settings:
    name = DataCategory.USERS
    
	@property
  def full_name(self):
    return f"{self.first_name} {self.last_name}"

通过实现 NoSQLBaseDocument ODM 类，我们将重点放在了每个文档或领域实体的字段和特定功能上。所有 CRUD 功能都委托给了父类。通过利用 Pydantic 定义字段，我们还获得了开箱即用的类型验证。例如，当创建 ArticleDocument 类的实例时，如果提供的链接为 None 或不是字符串，系统将抛出一个错误，提示数据无效。

到此为止，我们已经完成了数据采集管道的实现，首先是 ZenML 组件，然后是爬虫的实现，最后封装成 ODM 类和数据类别文档。接下来的步骤是运行数据采集管道并将原始数据导入 MongoDB 数据仓库。

总结

最后，我们用一个 UML 来直观的总结这一章节中关于数据收集的代码设计结构：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
classDiagram
    class NoSQLBaseDocument {
    	+UUID4 id
    	+__eq__(value: object) bool
    	+__hash__() int
    	+from_mongo(data: dict) NoSQLBaseDocument
    	+to_mongo(**kwargs) dict
    	+save(**kwargs) NoSQLBaseDocument
    	+get_or_create(**filter_options) NoSQLBaseDocument
    	+bulk_insert(documents: list) bool
    	+find(**filter_options) NoSQLBaseDocument
    	+bulk_find(**filter_options) list
    	+get_collection_name() str
    }

    class Document {
    	+dict content
    	+str platform
    	+UUID4 author_id
    	+str author_full_name
    }

    class RepositoryDocument {
    	+str name
    	+str link
    	+Settings
    }

    class PostDocument {
    	+Optional[str] image
    	+Optional[str] link
    	+Settings
    }

    class ArticleDocument {
    	+str link
    	+Settings
    }

    class UserDocument {
    	+str first_name
    	+str last_name
    	+Settings
    	+full_name()
    }

    class DataCategory {
    	<<enumeration>>
    	+PROMPT
    	+QUERIES
    	+INSTRUCT_DATASET_SAMPLES
    	+INSTRUCT_DATASET
    	+PREFERENCE_DATASET_SAMPLES
    	+PREFERENCE_DATASET
    	+POSTS
    	+ARTICLES
    	+REPOSITORIES
    }
    
    class CrawlerDispatcher {
    	-Dict[str, Type[BaseCrawler]] _crawlers
    	+build() CrawlerDispatcher
    	+register_xxx() CrawlerDispatcher
    	-register(domin:str, crawler:Type[BaseCrawler])
    	+get_crawler(url:str) BaseCrawler
    }
    
    class BaseCrawler {
    	<<abstract>>
    	+NoSQLBaseDocument model
    	+extract(link:str, **kwargs) Any
    }
    
    class BaseSeleniumCrawler {
    	<<abstract>>
    	+int scroll_limit
    	+driver
    	+set_extra_driver_options(op:Options)
    	+login()
    	+scroll_page()
    }
    
    class GithubCrawler
    class CustomArticleCrawler
    class MediumCrawler
    
    class MongoDB

    NoSQLBaseDocument <|-- Document : 继承自
    Document <|-- RepositoryDocument : 继承自
    Document <|-- PostDocument : 继承自
    Document <|-- ArticleDocument : 继承自
    NoSQLBaseDocument <|-- UserDocument : 继承自
    RepositoryDocument --> DataCategory : 引用
    PostDocument --> DataCategory : 引用
    ArticleDocument --> DataCategory : 引用
    UserDocument --> DataCategory : 引用
    
    CrawlerDispatcher --> BaseCrawler : 引用
    BaseCrawler --> NoSQLBaseDocument : 引用
    BaseCrawler <|-- BaseSeleniumCrawler : 继承自
    GithubCrawler --|> BaseCrawler : 继承自
    CustomArticleCrawler --|> BaseCrawler : 继承自
    MediumCrawler --|> BaseSeleniumCrawler : 继承自
    
    NoSQLBaseDocument ..> MongoDB : 操作数据库

四、检索增强生成（RAG）管道

检索增强生成（RAG）在大多数生成式 AI 应用中是非常重要的。RAG 的核心职责是将自定义数据注入到大型语言模型（LLM）中，以执行给定的操作（例如总结、重述和提取注入的数据）。通常希望在 LLM 中使用它没有训练过的数据（例如私有数据或新数据），由于微调 LLM 是一项非常昂贵的操作，因此 RAG 成为了一种非常有吸引力的策略，可以绕过不断微调的需求来访问这些新数据。

我们将从理论部分开始，重点介绍 RAG 的基础和它如何工作。接着，我们将逐步引导你了解一个简单的 RAG 系统的所有组件：分块、嵌入和向量数据库。最终，我们将介绍用于高级 RAG 系统的各种优化。然后，我们将继续探索 LLM Twin 的 RAG 特性管道架构。在这一步，我们将应用章节开头讨论的所有理论内容。最后，我们将通过实现 LLM Twin 的 RAG 特性管道来演示实践中的应用。

本章的主要内容如下：

  • 理解 RAG
  • 高级 RAG 概述
  • 探索 LLM Twin 的 RAG 特性管道架构
  • 实现 LLM Twin 的 RAG 特性管道

理解 RAG

RAG 通过从外部数据源检索信息来增强生成式 AI 模型的准确性和可靠性。它是一种与 LLM 内部知识互补的技术。在深入细节之前，我们先来理解 RAG 的含义：

  • 检索（Retrieval）：搜索相关数据
  • 增强（Augmented）：将数据作为上下文添加到提示中
  • 生成（Generation）：使用增强后的提示与 LLM 进行生成

任何 LLM 都只能理解它所训练过的数据，这通常被称为参数化知识。因此，即使 LLM 能够完美地回答过去发生的事情，但对于最新的数据或它未曾训练过的任何外部资源，它也无法得知。另一种情况是，它可能会自信地“幻想”并提供错误的答案。

要理解 RAG，首先你需要知道，在使用 RAG 时，我们将必要的信息注入到提示中，将增强后的提示传递给 LLM，进行最终回答。此时，LLM 会利用额外的上下文来回答用户问题。

RAG 解决了两个根本问题：

  • 幻觉（Hallucinations）
  • 过时或私有信息（Old or private information）

幻觉（Hallucinations）

如果一个没有使用 RAG 的聊天机器人被问到它没有训练过的问题，它很可能会”自信“地给出一个错误答案，让人难以分辨真假。即使 LLM 并不总是产生幻觉，这种情况仍然会引发对其答案可信度的担忧。因此，就带来了两个问题：

  • 什么时候可以信任 LLM 的回答？
  • 如何评估答案是否正确？

通过引入 RAG，我们强制 LLM 仅根据提供的上下文进行回答。LLM 作为推理引擎，而 RAG 提供的额外信息则充当生成答案的唯一真实来源。这样，我们可以快速评估 LLM 的答案是否基于外部数据。

过时信息（Old Information）

任何 LLM 都只能在某个特定时间点上的全世界知识的子集上进行训练或微调，主要有以下三个原因：

  • 私有数据（Private data）：你无法在没有所有权或使用权的数据上训练模型。
  • 新数据（New data）：新数据每时每刻都在产生，因此必须不断训练 LLM 以跟上更新。
  • 成本（Costs）：训练或微调 LLM 是一项极其昂贵的操作，因此无法频繁进行。

RAG 解决了这些问题，因为你不再需要不断微调 LLM 以适应新数据（甚至是私有数据）。只需将必要的数据直接注入到 LLM 处理的提示（prompt）中，就能生成正确且有价值的答案。

简单 RAG 框架

RAG 系统的基本结构都非常相似。我们将首先集中了解最简单形式的 RAG。请注意，我们将简单的 RAG（Vanilla RAG） 和原始 RAG（Naive RAG） 交替使用，以避免重复。

RAG 系统由三个主要模块组成，彼此独立：

  • 数据输入管道（Ingestion pipeline）：一个用于填充向量数据库的批处理或流式管道。
  • 检索管道（Retrieval pipeline）：一个查询向量数据库并检索与用户输入相关的条目的模块。
  • 生成管道（Generation pipeline）：使用检索到的数据来增强提示并与 LLM 一起生成答案。

由于这三个组件是独立的类或服务，我们将分别深入了解它们。但目前，让我们尝试回答“这三个模块如何连接？”的问题。以下是一个非常简单的概述：

  1. 在后台，数据输入管道根据计划或持续运行，将外部数据填充到向量数据库中；
  2. 在客户端，用户提出一个问题；
  3. 问题传递给检索模块，检索模块对用户输入进行预处理并查询向量数据库；
  4. 生成管道使用提示模板、用户输入和检索到的上下文来创建提示；
  5. 提示传递给 LLM 以生成答案；
  6. 答案显示给用户。

screenshot_2025-03-12_16.04.00

当你需要访问任何类型的外部信息时，就必须在你的生成式 AI 应用中实现 RAG。例如，在实现一个财务助手时，你很可能需要访问最新的新闻、报告和价格，才能提供有价值的答案。或者，如果你构建一个旅行推荐系统，你必须检索并解析一份潜在景点、餐馆和活动的列表。在训练时，LLM 并没有访问你的特定数据，因此你将经常需要在生成式 AI 项目中实现 RAG 策略。

现在，让我们深入探讨数据输入、检索和生成管道。

数据输入管道（Ingestion pipeline）

RAG 摄取管道从各种数据源（例如数据仓库、数据湖、网页等）提取原始文档。然后，它会清理、分块并嵌入文档。最后，它会将嵌入的块加载到向量数据库（或其他类似的向量存储）中。

因此，RAG 数据输入管道进一步分为以下几个模块：

  1. 数据提取模块（data extraction）
    该模块负责从各种数据源（如数据库、API 或网页）中收集所需的数据。这个模块高度依赖于你的数据。它可以简单地通过查询数据仓库来完成，也可以是更复杂的操作，例如爬取维基百科等网站。
  2. 清洗层（cleaning）
    清洗层对提取的数据进行标准化处理，并移除不需要的字符。这样可以确保数据质量，提高后续处理的准确性。
  3. 切分模块（chunking）
    该模块将清洗后的文档拆分成较小的部分。由于我们希望将文档的内容传递给嵌入模型，因此必须确保内容不超过模型的最大输入大小。切分还需要确保将语义上相关的区域分开。例如，在切分一本书的章节时，最优的方式是将相似的段落分到同一个切片中。这样做可以确保在检索时，只将必要的数据添加到提示中。
  4. 嵌入组件（embedding）
    嵌入组件使用嵌入模型将切片内容（如文本、图片、音频等）映射到一个密集的向量中，这个向量承载了语义信息。在本章的后续部分，我们将深入讨论嵌入模型。
  5. 加载模块（loading）
    加载模块负责将嵌入后的切片及其元数据文档存储到数据库中。元数据将包含一些关键信息，例如嵌入的内容、该切片的源 URL、该内容在网页上发布的时间等。嵌入向量作为索引，用于查询相似的切片，而元数据则用于访问用来增强提示的信息。

检索管道（Retrieval Pipeline）

检索模块接收用户输入（文本、图像、音频等），将其嵌入，并查询向量数据库（DB）以寻找与用户输入相似的向量。

检索步骤的主要功能是将用户输入投影到与嵌入数据库中作为索引的嵌入相同的向量空间。这使得我们能够通过比较向量存储中的嵌入与用户输入的向量来找到最相似的前 K 个条目。这些条目随后作为增强提示的一部分，传递给 LLM 用于生成答案。

为了比较两个向量，必须使用一种距离度量方法，例如欧几里得距离或曼哈顿距离。但最常用的距离度量是余弦距离，计算公式为：
$$
CosineDistance = 1 - \cos(\theta) = 1 - \frac{A \cdot B}{\left | A \right | \cdot \left | B \right |}
$$
其中，$\theta$ 是两个向量之间的夹角。余弦距离的值范围从 -1 到 1：

  • 当两个向量完全相反时，余弦距离为 -1；
  • 当两个向量正交时，余弦距离为 0；
  • 当两个向量指向相同的方向时，余弦距离为 1。

通常，余弦距离在非线性复杂的向量空间中表现良好。然而，需要注意的是，选择合适的向量之间的距离度量方法取决于数据和所使用的嵌入模型。

关键因素需要特别强调的是，用户的输入和嵌入必须位于相同的向量空间中。否则，你就无法计算它们之间的距离。为此，必须以与 RAG 数据输入管道中处理原始文档相同的方式对用户输入进行预处理。这意味着必须清理用户输入，必要时进行切分，并使用相同的函数、模型和超参数来嵌入用户输入。这与训练和推理时数据特征的预处理方法相似。如果处理不一致，推理结果可能会不准确，这种现象也被称为“训练-服务偏差”（training-serving skew）。

生成管道（Generation Pipeline）

RAG 系统的最后一步是获取用户的输入，检索相关数据，将其传递给大语言模型（LLM），并生成有价值的答案。

在这一步中，最终的提示（prompt）是由系统模板和用户查询以及检索到的上下文填充而成的。根据应用的不同，您可能会使用一个单独的提示模板或多个提示模板。通常，所有的提示工程（prompt engineering）工作都是在提示模板的层面进行的。

以下是一个虚拟示例，展示了一个通用的系统和提示模板，并说明它们如何与检索逻辑和 LLM 一起使用，生成最终的答案：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
system_template = """
You are a helpful assistant who answers all the user's questions politely.
"""

prompt_template = """
Answer the user's question using only the provided context. If you cannot
answer using the context, respond with "I don't know."

Context: {context}
User question: {user_question}
"""

user_question = "<your_question>"
retrieved_context = retrieve(user_question)

prompt = f"{system_template}\n"
prompt += prompt_template.format(context=retrieved_context, user_question=user_question)
answer  = llm(prompt)

随着提示模板的不断演化，每次修改都应使用机器学习操作（MLOps）的最佳实践进行跟踪和版本控制。这样，在训练或推理时，您始终可以知道给定的答案是由特定版本的 LLM 和提示模板生成的。您可以通过 Git 跟踪版本，或者将提示模板存储在数据库中，或者使用像 LangFuse 这样的专门提示管理工具。

正如我们在检索管道中看到的，直接影响 RAG 系统准确性的一些关键方面是外部数据的嵌入，通常存储在向量数据库中，用户查询的嵌入，以及我们如何使用余弦距离等函数来衡量这两者之间的相似性。为了更好地理解 RAG 算法的这一部分，我们将深入探讨嵌入是什么以及它们如何计算。

嵌入（Embeddings）

什么是嵌入（Embeddings）？

想象一下，我们正在教计算机理解世界。嵌入就像是一个特殊的翻译器，将这些事物转化为数值代码。不过，这些代码并不是随机的，因为相似的单词或物品会被赋予彼此接近的代码。它就像一张地图，其中具有相似含义的单词被聚集在一起。

从更理论的角度来看，嵌入是物体的稠密数值表示，这些物体以向量的形式编码在一个连续的向量空间中，可能是单词、图像，或者推荐系统中的物品。这个转化帮助捕捉物体之间的语义意义和关系。例如，在自然语言处理（NLP）中，嵌入将单词转换为向量，使得语义相似的单词在向量空间中彼此接近。

嵌入的可视化

一种常见的方法是可视化嵌入，以便理解和评估它们之间的几何关系。由于嵌入通常有超过 2 或 3 维，通常在 64 到 2048 之间，因此需要将其重新投影到 2D 或 3D 空间中。

例如，你可以使用 UMAP（UMAP文档），这是一种降维方法，因其能够在将嵌入投影到 2D 或 3D 时保持点之间的几何特性而受到广泛欢迎。另一个常用的降维算法是 t-SNE（t-SNE文档）。然而，相比 UMAP，t-SNE 更具随机性，并且不总是保持点之间的拓扑关系。

总结来说，嵌入是一个将对象转换为能够捕捉其语义关系的数值向量的技术，广泛应用于自然语言处理、图像处理和推荐系统等领域。

为什么嵌入（Embeddings）如此强大

首先，机器学习模型只能处理数值型数据。当处理表格数据时，这通常不是问题，因为数据通常是数值型的，或者可以很容易地转换为数字。然而，当我们希望将文字、图像或音频数据输入模型时，嵌入就显得特别有用。

例如，在处理 Transformer 模型时，您需要对所有文本输入进行分词，每个分词都会关联一个嵌入（embedding），神经网络的密集层可以轻松地处理这些嵌入（embedding）。

基于这个例子，您可以使用嵌入来编码任何类别变量，并将其传递给机器学习模型。那么，为什么不使用其他简单的方法，如独热编码（One-hot Encoding）呢？当处理具有高基数的类别变量时，如语言词汇表，使用其他经典方法会遭遇“维度灾难”（curse of dimensionality）。例如，如果您的词汇表有 10,000 个标记，那么应用独热编码后，每个标记的长度就是 10,000。如果输入序列有 N 个标记，那么输入参数将变成 $N \times 10,000$。若 $N >= 100$，则文本输入通常会过于庞大，无法使用。另一个经典方法，如哈希编码（Feature Hashing），虽然不受维度灾难的困扰，但它会丢失向量之间的语义关系。

[!TIP]

独热编码：

  • 是一种将类别变量转换为二进制矩阵表示的技术。每个类别都会被表示为一个唯一的二进制向量。对于每个类别变量，都会创建一个二进制向量，其长度等于唯一类别的数量，除对应类别的索引外，所有值都为零。
  • 优点：保留了类别的所有信息，简单且可解释。
  • 缺点：当类别变量有许多唯一值时，特征空间会变得非常高维，导致该方法在实际应用中不可行。

特征哈希：

  • 也称为哈希编码或“哈希技巧”，是一种通过对类别值应用哈希函数将类别变量转换为数值特征的方法。与独热编码相比，这种方法不受唯一类别数的限制，而是通过将类别映射到固定数量的箱或桶中来减少特征空间的维度。
  • 优点：减少了特征空间的维度，这在处理高基数类别变量时特别有用。在内存使用和计算时间方面具有较高的效率。
  • 缺点：哈希可能会发生冲突，即不同的类别可能会映射到同一个桶中，从而导致信息丢失。此外，这种映射使得方法不可解释，并且很难理解原始类别和哈希特征之间的关系。

嵌入的优势

嵌入帮助我们编码类别变量，并且可以控制输出向量的维度。与朴素的哈希技巧不同，嵌入使用巧妙的方法将信息压缩到比哈希技巧更低维的空间中。

其次，通过对输入进行嵌入，我们可以降低它的维度，并将所有语义信息压缩成一个密集的向量。这在处理图像时是一种非常流行的技术，其中卷积神经网络（CNN）编码模块将高维的图像信息映射为嵌入，然后通过CNN解码器对该嵌入进行处理，进行分类或回归操作。

Typical_cnn

上图是一个典型的CNN结构的示意图。想象一下每个层中的小方块，它们是“感受区块”。每个小方块将信息传递给前一层的单个神经元。在网络的不同层级中，主要发生了两个关键操作：

  • 缩小图像尺寸：特殊的“采样 subsampling”操作将图像尺寸减小，集中关注图像的关键信息。
  • 学习特征：另一方面，“卷积 convolution”操作会增加图像特征层的尺寸，以便网络能够从图像中学习更复杂的特征。

最终，网络中的完全连接层会将所有这些处理过的信息转化为最终的向量嵌入，即图像的数值表示。

如何创建嵌入（Embeddings）

嵌入是通过深度学习模型创建的，这些模型能够理解输入的上下文和语义，并将其投影到一个连续的向量空间中。根据数据输入的类型，嵌入的创建方法有所不同。因此，在选择嵌入模型之前，理解你的数据和需求至关重要。

文本数据中的嵌入

例如，当处理文本数据时，Word2Vec 和 GloVe 是早期用于创建词汇嵌入的常见方法，它们在一些简单的应用中仍然广泛使用。另一种流行的方法是使用 encoder-only 的 transformers，例如 BERT，及其系列中的其他方法，例如 RoBERTa。这些模型利用转换器架构的编码器将您的输入智能地投影到密集向量空间中，稍后可将其用作嵌入。

[!TIP]

BERT

基于 Transformer 的双向编码器表示技术（英语：Bidirectional Encoder Representations from Transformers，BERT）是用于自然语言处理（NLP）的预训练技术，由 Google 提出。通过在所有层中共同调整左右情境，利用无标记文本预先训练深度双向表示（深度双向、无监督式）。同时考虑句子中单词的左右上下文，BERT 使用双向方法，而不是按顺序分析文本，BERT 同时查看句子中的所有单词。

例如：“The bank is situated on the ___ of the river.”

  • 在单向模型中，对空白的理解将严重依赖于前面的单词，并且模型可能难以辨别*“bank”*是指银行还是河的一侧。

  • BERT 是双向的，它同时考虑左侧（“The bank is situated on the”）和右侧上下文（“of the river”），从而实现更细致的理解。它理解缺失的单词可能与银行的地理位置有关，展示了双向方法带来的语境丰富性。

BERT 的创新之处在于借助 Transformer 学习双向表示，Transformer 是一种深度学习组件，不同于递归神经网络 (RNN) 对顺序的依赖性，它能够并行处理整个序列。因此可以分析规模更大的数据集，并加快模型训练速度。Transformer 能够使用注意力机制收集词语相关情境的信息，并以表示该情境的丰富向量进行编码，从而同时处理（而非单独处理）与句中所有其他词语相关的词语。该模型能够学习如何从句段中的每个其他词语衍生出给定词语的含义。

BERT GPT
架构 BERT 专为双向表征学习而设计。它使用掩码语言模型目标，根据左右上下文预测句子中缺失的单词。 另一方面，GPT 是为生成式语言建模而设计的。它利用单向自回归方法，根据前面的上下文预测句子中的下一个单词。
预训练目标 BERT 使用掩码语言模型目标和下一句预测进行预训练。它专注于捕捉双向上下文并理解句子中单词之间的关系。 GPT 经过预先训练，可以预测句子中的下一个单词，这有助于模型学习语言的连贯表示并生成上下文相关的序列。
上下文理解 BERT 对于需要深入理解句子内的上下文和关系的任务非常有效，例如文本分类、命名实体识别和问答。 GPT 擅长生成连贯且上下文相关的文本。它常用于创意任务、对话系统和需要生成自然语言序列的任务。
任务类型和用例 常用于文本分类、命名实体识别、情感分析和问答等任务。 应用于文本生成、对话系统、总结和创意写作等任务。
微调与小样本学习 BERT 通常使用标记数据针对特定的下游任务进行微调，以使其预训练表示适应当前的任务。 GPT 旨在执行小样本学习，它可以用最少的特定任务训练数据推广到新任务。

下面是一个简单的 Python 代码示例，演示如何使用 SentenceTransformer 来计算句子的嵌入，并计算它们之间的余弦相似度：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")
sentences = [
    "The dog sits outside waiting for a treat.",
    "I am going swimming.",
    "The dog is swimming."
]

# 计算嵌入
embeddings = model.encode(sentences)
print(embeddings.shape)  # 输出: [3, 384]

# 计算余弦相似度
similarities = model.similarity(embeddings, embeddings)
print(similarities)
# 输出:
# tensor([[ 1.0000, -0.0389, 0.2692],
#         [-0.0389, 1.0000, 0.3837],
#         [ 0.2692, 0.3837, 1.0000]])

在这个例子中，我们开源看到这三个句子的嵌入向量与它们之间的余弦相似度：

  • 第一个句子与其自身的相似度为 1。
  • 第一个和第二个句子的相似度接近 0，说明它们没有什么相似性。
  • 第一个和第三个句子的相似度较高，表明它们有一些共同的上下文。

嵌入模型因为使用场景的不同而变化。我们可以在 Hugging Face 上的 Massive Text Embedding Benchmark (MTEB) 上找到特定的模型。根据需求，可以考虑性能最佳的模型、准确率最高的模型或内存占用最小的模型。同时，使用 Hugging Face 和 SentenceTransformer 使不同模型之间的切换变得简单。因此，您始终可以尝试各种选项。

处理图像时，您可以使用卷积神经网络 (CNN) 嵌入它们。流行的 CNN 网络基于 ResNet 架构。但是，我们不能直接将图像嵌入技术用于录音。相反，我们可以创建音频的视觉表示，例如频谱图，然后将图像嵌入模型应用于这些视觉效果。这使我们能够以计算机可以理解的方式捕捉图像和声音的本质。

通过利用 CLIP 之类的模型，您实际上可以将一段文本和一张图片嵌入到同一个向量空间中。这样您就可以使用句子作为输入来查找相似的图像，反之亦然，这证明了 CLIP 的实用性。

crazy_cat
  • “A crazy cat smiling.”
  • “A white and brown cat with a yellow bandana.”
  • “A man eating in the garden.”

在下面的代码片段中，我们使用 CLIP 对一张小猫图像（如上图）和三句话进行编码。最后，我们使用余弦相似度来计算图片和句子之间的相似度：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from io import BytesIO
import requests
from PIL import Image
from sentence_transformers import SentenceTransformer

image = Image.open(image_dir)
model = SentenceTransformer("clip-ViT-B-32")
img_emb = model.encode(image)
text_emb = model.encode(
[
  "A crazy cat smiling.",
  "A white and brown cat with a yellow bandana.",
  "A man eating in the garden."
])
print(text_emb.shape) # noqa # Output: (3, 512)
similarity_scores = model.similarity(img_emb, text_emb) print(similarity_scores) # noqa
# Output: tensor([[0.3068, 0.3300, 0.1719]])

至此，我们简要介绍了如何计算嵌入。具体实现的范围很广，大多数数据类别（如单词、句子、文档、图像、视频和图形）都可以计算嵌入。当我们需要计算两个不同数据类别之间的距离（例如句子向量和图像向量之间的距离）时，必须使用专门的模型，这一点至关重要。这些模型旨在将两种数据类型投影到同一个向量空间（如 CLIP），以确保准确的距离计算。

向量数据库

向量数据库是专门设计用于高效存储、索引和检索向量嵌入的数据库。传统的基于标量的数据库难以应对向量数据的复杂性，因此向量数据库对于实时语义搜索等任务至关重要。

虽然像 FAISS 这样的独立向量索引对于相似性搜索很有效，但它们缺乏向量数据库的全面数据管理功能。向量数据库支持 CRUD 操作、元数据过滤、可扩展性、实时更新、备份、生态系统集成和强大的数据安全性，因此它们比独立索引更适合生产环境。

向量数据库如何工作？

想想您通常如何搜索数据库。您输入一些特定内容，系统会输出精确匹配。这就是传统数据库的工作方式。向量数据库则不同。我们不是寻找完美匹配，而是寻找查询向量的最近邻居。在底层，向量数据库使用近似最近邻 （Approximate Nearest Neighbor, ANN） 算法来查找这些近邻。

虽然 ANN 算法不会返回给定搜索的最佳匹配，但标准最近邻算法在实践中太慢了。此外，经验表明，仅使用给定输入查询的最佳匹配的近似值就可以很好地工作。因此，准确性和延迟之间的权衡之下最终选择了 ANN 算法。

这是向量数据库的典型工作流程：

  1. 索引向量：使用针对高维数据优化的数据结构对向量进行索引。常见的索引技术包括分层可导航小世界 (HNSW)、随机投影、乘积量化 (PQ) 和局部敏感哈希 (LSH)。

  2. 查询相似性：在搜索过程中，数据库查询索引向量以找到与输入向量最相似的向量。此过程涉及基于相似性度量（例如余弦相似性、欧几里得距离或点积）比较向量。每种方法都有独特的优势，适用于不同的用例。

  3. 结果的后处理：在识别潜在匹配后，对结果进行后处理以提高准确性。此步骤可确保将最相关的向量返回给用户。

向量数据库可以在向量搜索之前或之后根据元数据过滤结果。这两种方法在性能和准确性方面都有权衡。查询还依赖于元数据（以及向量索引），因此它包含用于过滤操作的元数据索引用户。

向量索引的算法

向量数据库使用各种算法来创建向量索引并高效地管理搜索数据：

  • 随机投影：随机投影通过使用随机矩阵将向量投影到低维空间来降低向量的维数。该技术保留了向量之间的相对距离，从而有助于更快地进行搜索。
  • 乘积量化（PQ）：PQ 通过将向量分成更小的子向量，然后将这些子向量量化为代表性代码来压缩向量。这减少了内存使用量并加快了相似性搜索。
  • 局部敏感哈希 （LSH）：LSH 将相似的向量映射到存储桶中。此方法通过关注数据子集来实现快速近似最近邻搜索，从而降低计算复杂度。在查询时，只查找同一个桶内的近邻，从而提高查询效率。
  • 分层可导航小世界 （HNSW）：HNSW 构建一个多层图，其中每个节点代表一组向量。相似的节点相互连接，允许算法导航图并有效地找到最近的邻居。

这些算法使向量数据库能够有效地处理复杂和大规模数据，使其非常适合各种 AI 和 ML 应用程序。

数据库操作

向量数据库还与普通数据库拥有共同的一些特点，以确保高性能、容错性和在生产环境中的易管理性。关键操作包括：

  • 分片和复制：数据会被划分（分片）到多个节点上，以确保可扩展性和高可用性。数据在节点之间的复制有助于维护数据完整性，并在节点失败时确保数据的可用性。
  • 监控：持续监控数据库性能，包括查询延迟和资源使用情况（RAM、CPU、磁盘），有助于维持最佳操作，并在潜在问题影响系统之前发现它们。
  • 访问控制：实施强大的访问控制机制，确保只有授权用户可以访问和修改数据。这包括基于角色的访问控制和其他安全协议来保护敏感信息。
  • 备份：定期的数据库备份对灾难恢复至关重要。自动备份过程确保在数据损坏或丢失的情况下，能够恢复到之前的状态。

高级 RAG 框架

我们刚刚介绍的基础RAG框架并没有解决许多影响检索和答案生成质量的基本方面，比如：

  • 检索到的文档是否与用户的问题相关？
  • 检索到的上下文是否足够回答用户的问题？
  • 是否有冗余信息，只是增加了增强提示的噪音？
  • 检索步骤的延迟是否符合我们的要求？
  • 如果我们无法使用检索到的信息生成有效答案，应该怎么办？

从上述问题中，我们可以得出两个结论。第一个结论是，我们需要一个强大的评估模块来衡量和量化检索数据的质量，并根据用户的问题生成答案。我们将在第九章详细讨论这个话题。第二个结论是，我们必须改进RAG框架，直接在算法中解决检索的限制。这些改进被称为高级RAG。

基础RAG设计可以在三个不同的阶段进行优化：

  • 检索前：这个阶段关注如何构建和预处理数据，以优化数据索引和查询。
  • 检索：这个阶段主要围绕改进嵌入模型和元数据过滤，以提高向量检索步骤的效果。
  • 检索后：这个阶段主要聚焦于通过不同方式过滤检索到的文档中的噪音，并在将其输入LLM进行答案生成之前对提示进行压缩。

检索前优化

检索前的优化步骤有两种不同的方式：

  • 数据索引：这是RAG摄取管道的一部分，主要是在清理或分块模块中实现，目的是对数据进行预处理，以便更好地进行索引。
  • 查询优化：该算法直接作用于用户的查询，在嵌入查询并从向量数据库中检索数据之前。

数据索引

在我们使用嵌入来索引数据时，嵌入语义上代表了分块文档的内容，因此，大部分数据索引技术都专注于通过更好的预处理和数据结构化来提高检索效率，例如：

  • 滑动窗口（Sliding Window）：滑动窗口技术在文本块之间引入重叠，确保在块边界附近的重要上下文得到保留，从而提高检索的准确性。这在一些领域中尤为有用，如法律文件、科研论文、客户支持日志和医疗记录等，因为关键信息往往跨越多个部分。嵌入是在文本块及其重叠部分上计算的，因此滑动窗口技术通过保持跨越边界的上下文来增强系统检索相关和连贯信息的能力。

  • 增强数据粒度：包括数据清理技术，如删除无关细节、验证事实准确性以及更新过时信息。一个干净且准确的数据集有助于更精确的检索。

  • 元数据（Metadata）：添加如日期、URL、外部ID或章节标记等元数据标签，帮助在检索时有效地过滤结果。

  • 优化索引结构：基于不同的数据索引方法，如不同的块大小和多重索引策略。

  • 小到大（Small-to-big）：该算法将用于检索的文本块与最终生成答案时使用的上下文解耦。该算法使用较小的文本序列来计算嵌入，同时保留该序列本身以及周围较大的窗口作为元数据。因此，使用较小的块可以提高检索的准确性，而更大的上下文则为LLM提供更多的上下文信息。但是，如果我们使用整篇文本来计算嵌入，可能会引入太多噪音，或者文本包含多个主题，这会导致嵌入的整体语义表示较差。

查询优化 (Query Optimization)

在查询优化方面，我们可以利用查询路由、查询重写和查询扩展等技术，进一步优化为LLM检索到的信息：

  • 查询路由 (Query Routing)：根据用户的输入，可能需要与不同类别的数据交互，并分别查询每个类别。查询路由用于根据用户的输入决定采取何种操作，类似于if/else语句，但决策完全基于自然语言，而不是逻辑语句。

    screenshot_2025-03-13_10.28.21

    如图所示，假设基于用户的输入，为了执行 RAG，我们可以使用向量搜索查询从向量数据库中检索附加的上下文，或者通过将用户查询转换为 SQL 命令来从标准 SQL 数据库中检索，或者通过利用 REST API 调用从互联网中获取。查询路由器还可以检测是否需要上下文，帮助我们避免对外部数据存储进行冗余调用。此外，查询路由器还可以用来选择最佳的提示模板。例如，在 LLM Twin 用例中，根据用户是想要一段文章、一个帖子还是一段代码片段，需要不同的提示模板来优化创作过程。查询路由通常使用 LLM 来决定采取哪条路径，或者通过选择具有最相似向量的路径来进行嵌入选择。总的来说，查询路由类似于 if/else 语句，但它更具灵活性，因为它直接与自然语言交互。

  • 查询重写 (Query Rewriting)：有时，用户的初始查询可能与数据的结构不完全对齐。查询重写通过重新构造问题，使其更好地匹配已索引的信息来解决这个问题。常见的方法包括：

    • 释义 (Paraphrasing)：在保持原意的基础上重述用户的查询（例如，“气候变化的原因是什么？”可以重写为“导致全球变暖的因素”）。
    • 同义词替换 (Synonym Substitution)：用同义词替换不常见的词，以扩大搜索范围（例如，“joyful”可以重写为“happy”）。
    • 子查询 (Sub-queries)：对于较长的查询，可以将其拆分为多个更短且更集中的子查询。这有助于检索阶段更精确地识别相关文档。

  • 假设文档嵌入 (HyDE)：该技术通过让 LLM 生成对查询的假设响应。然后，将原始查询和LLM的响应一起送入检索阶段。

  • 查询扩展 (Query Expansion)：该方法通过添加额外的术语或概念来丰富用户的提问，从而提供同一初始问题的不同视角。例如，在搜索“疾病”时，可以利用同义词和与原始查询词相关的术语，甚至包括“疾病”或“病症”。

  • 自查询 (Self-query)：核心思想是将非结构化查询映射到结构化查询。LLM 识别输入文本中的关键实体、事件和关系，并将这些身份作为过滤参数，用于减少向量搜索的空间（例如，识别查询中的城市名，如“巴黎”，并将其添加到过滤器中，以减少向量搜索空间）。

检索优化

检索步骤可以通过两种基本方式进行优化：

  • 改进嵌入模型：在 RAG 摄取管道中使用的嵌入模型，用于编码分块文档，并在推理时转换用户的输入。
  • 利用数据库的过滤和搜索功能：此步骤仅在推理时使用，当需要基于用户输入检索最相似的文本块时使用。

这两种策略的最终目标一致：通过利用查询和已索引数据之间的语义相似性，增强向量检索步骤。

改进嵌入模型

在改进嵌入模型时，通常需要对预训练的嵌入模型进行微调，以使其适应特定领域的术语和细微差别，尤其是在术语不断变化或包含稀有词汇的领域。

如果不想对嵌入模型进行微调，可以利用Instructor模型（Instructor模型示例）通过简单的指令，能够生成适用于任何任务（如分类、检索、聚类、文本评估等）和领域（如科学、金融等）的文本嵌入，而无需进行任何微调，因为微调模型会消耗更多的计算和人工资源。

假设你需要为特定领域的文本计算嵌入，可以按照以下步骤操作：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from InstructorEmbedding import INSTRUCTOR

# 加载模型
model = INSTRUCTOR('hkunlp/instructor-large')

# 准备带有指令的文本
text_instruction_pairs = [
    {"instruction": "Represent the Science title:", "text": "3D ActionSLAM: wearable person tracking in multi-floor environments"},
    {"instruction": "Represent the Medicine sentence for retrieving a duplicate sentence:", "text": "Recent studies have suggested that statins, an established drug group in the prevention of cardiovascular mortality, could delay or prevent breast cancer recurrence but the effect on disease-specific mortality remains unclear."}
]

# 计算嵌入
customized_embeddings = model.encode([[pair["instruction"], pair["text"]] for pair in text_instruction_pairs])

# 输出嵌入结果
for pair, embedding in zip(text_instruction_pairs, customized_embeddings):
    print("Instruction: ", pair["instruction"])
    print("Text: ", pair["text"])
    print("Embedding: ", embedding)
    print("")

使用 instructor 模型计算两组文本间的相似度：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from sklearn.metrics.pairwise import cosine_similarity

# 定义两组文本
sentences_a = [['Represent the Science sentence: ','Parton energy loss in QCD matter'], 
               ['Represent the Financial statement: ','The Federal Reserve on Wednesday raised its benchmark interest rate.']]
sentences_b = [['Represent the Science sentence: ','The Chiral Phase Transition in Dissipative Dynamics'],
               ['Represent the Financial statement: ','The funds rose less than 0.5 per cent on Friday']]

# 计算嵌入
embeddings_a = model.encode(sentences_a)
embeddings_b = model.encode(sentences_b)

# 计算相似度
similarities = cosine_similarity(embeddings_a, embeddings_b)

# 输出相似度结果
print(similarities)

使用 instructor 模型进行信息检索的一个例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

# 定义查询和文档
query = [['Represent the Wikipedia question for retrieving supporting documents: ','where is the food stored in a yam plant']]
corpus = [['Represent the Wikipedia document for retrieval: ','Capitalism has been dominant in the Western world since the end of feudalism...'],
          ['Represent the Wikipedia document for retrieval: ',"The disparate impact theory is especially controversial..."],
          ['Represent the Wikipedia document for retrieval: ','Disparate impact in United States labor law refers to practices...']]

# 计算嵌入
query_embeddings = model.encode(query)
corpus_embeddings = model.encode(corpus)

# 计算相似度并找到最相关文档
similarities = cosine_similarity(query_embeddings, corpus_embeddings)
retrieved_doc_id = np.argmax(similarities)

# 输出最相关文档的ID
print(retrieved_doc_id)

利用经典数据库过滤和搜索功能提高检索

在检索优化的另一个方面，你可以通过利用经典的过滤和搜索数据库功能来提高检索效率：

  • 混合搜索 （Hybrid Search）：这是一种结合向量搜索和关键词搜索的方法。关键词搜索擅长识别包含特定关键词的文档。当任务要求极高的精度，且检索的信息必须包括精确的关键词匹配时，混合搜索表现优秀。虽然向量搜索功能强大，但有时在寻找精确匹配时会遇到困难，但在寻找更广泛的语义相似性时却表现优异。通过将这两种方法结合，你可以利用关键词匹配和语义相似性。通常有一个叫做 alpha 的参数来控制两者之间的权重。算法会进行两次独立搜索，之后对结果进行标准化和统一。

  • 过滤向量搜索 （Filtered Vector Search）：这种搜索方式利用元数据索引来过滤特定关键词的元数据。与混合搜索不同，你首先仅通过向量索引检索数据，然后在检索前或检索后进行过滤，以减少搜索空间。

  • 在实际应用中，通常从过滤向量搜索或混合搜索开始，因为它们相对容易实现。这种方法让你可以根据性能灵活调整策略。如果结果不如预期，你始终可以微调嵌入模型以提高检索效果。

后检索优化 (Post-retrieval)

后检索优化主要针对已检索的数据进行处理，确保大型语言模型（LLM）的性能不受如有限上下文窗口或噪声数据等问题的影响。这是因为检索到的上下文有时可能太大，或者包含无关信息，都会干扰LLM的正常工作。

后检索阶段常用的两种优化方法是：

  • 提示压缩 (Prompt Compression)：去除不必要的细节，同时保留数据的核心要素。
  • 重排序 (Re-ranking)：使用一个跨编码器（cross-encoder）机器学习模型，计算用户输入和每个检索到的片段之间的匹配分数。根据这个分数，对检索到的项目进行排序，仅保留最相关的前 $N$ 个结果。**在初始检索阶段，系统根据某种标准（如相似度）返回一组文档。然而，由于初始排序可能并不总是能够准确反映文档与查询的真实相关性，因此需要进行重排序来提升检索结果的质量。**然而，我们不能在初始检索步骤中应用这种模型，因为它的计算成本较高。因此，一种常见的策略是通过嵌入之间的相似度距离进行数据检索，然后使用重排序模型对检索到的信息进行优化，如下图所示。

screenshot_2025-03-13_13.48.43

上述技术并不是所有可能解决方案的详尽列表。我们使用它们作为示例，帮助你理解在RAG工作流程中的每个步骤中可以（也应该）优化的内容。事实上，这些技术在不同类型的数据上可能有很大的差异。

例如，如果你处理的是多模态数据（如文本和图像），前述的许多技术可能不适用，因为它们是为文本数据设计的。

总结

这些优化的主要目标是提升RAG算法在三个关键阶段的效果：前检索、检索和后检索。这包括为更好的向量索引预处理数据、调整用户查询以获得更准确的搜索结果、增强嵌入模型、利用经典的数据库过滤操作以及去除噪声数据。牢记这些目标，你可以有效地优化你的RAG工作流程，以实现更好的数据处理和检索性能。

设计 LLM Twin 的 RAG 管道架构

现在我们已经对 RAG 及其工作原理有了基本，我们将继续探索我们的 LLM Twin 用例。目标是提供一个完整的实践示例，以巩固本章中介绍的理论。

任何RAG系统都可以分为两个独立的组件：

  • 数据摄取管道（Ingestion pipeline）：处理原始数据，清洗、分块、嵌入并将其加载到向量数据库（DB）中。
  • 推理管道（Inference pipeline）：查询向量数据库以获取相关的上下文，并最终通过大型语言模型（LLM）生成答案。

本章我们将专注于实现 RAG 数据摄取管道，而在第9章，我们将继续开发推理管道。

在此基础上，让我们快速回顾一下我们试图解决的问题以及原始数据的来源。在我们的端到端的机器学习（ML）系统中，所有组件都通过接口相互通信，每个管道都有单一职责。在我们的案例中，我们摄取原始文档，进行预处理，然后将它们加载到向量数据库中。

在本章中，我们特别希望设计一个 RAG 特性管道，该管道从我们的 MongoDB 数据仓库中获取原始社交媒体数据（例如文章、代码库和帖子）。原始文档的文本将被清理、分块、嵌入，并最终加载到向量数据库（特性存储）中。在推理阶段，用于生成答案的上下文将从向量数据库中检索。因此，数据仓库和特性存储之间的同步速度将直接影响我们 RAG 算法的准确性。

另一个关键考虑因素是如何自动化特性管道并将其与我们其余的ML系统集成。我们的目标是最小化两个数据存储之间的数据同步差异，因为这可能会危及系统的完整性。

最后，我们必须设计一个特性管道，该管道持续同步数据仓库和逻辑特性存储，同时根据需要处理数据。将数据存储在特性存储中对于生产就绪的 ML 系统至关重要。LLM Twin 推理管道将在 RAG 中查询它，而训练管道将从中消费已追踪和版本化的微调数据集。

特性存储

特性存储这是一个集中存储所有特性（用于训练和推理管道中的数据特征）的地方。特性存储是训练和推理管道之间的桥梁。训练管道会从特性存储中获取清洗过的数据（以 artifacts 形式存储），用于微调LLM（大型语言模型）。而推理管道则会查询向量数据库，以获取用于RAG（检索增强生成）的分块文档。

  • 为什么设计特性管道而不仅仅是RAG数据摄取管道：因为特性管道不仅仅是为了处理RAG相关的数据，它还处理数据的清洗、分块、嵌入等操作。RAG逻辑只是特性管道中的一个子组件。
  • 特性管道的作用：它像一个思维导图一样，用来帮助导航机器学习系统的复杂性。特性管道的输入是原始数据，输出是经过处理的特性（以及可选标签），这些特性会被存储在特性存储中。特性管道包含一个或多个子管道，它们的作用可能是对数据进行清洗、创建指令数据集、实现数据验证等。
  • 关于特性和文本数据的观察：如果遵循标准约定，存储为字符串的文本数据不算作“特性”。特性是直接输入到模型中的数据。例如，指令数据集或分块文档必须经过“标记化”（tokenization）处理，才能成为特性，因为模型接受的是“标记”而不是字符串形式的句子。这个处理通常会在运行时进行，但这一点表明了特性、训练和推理架构（FTI）不需要过于死板，可以根据实际情况进行调整。。

[!TIP]

所有原始文档都存储在 MongoDB 数据仓库中。数据仓库由第3章中介绍的数据采集ETL管道填充。该ETL管道爬取多个平台，如 Medium 和 Substack，标准化数据并将其加载到 MongoDB 中。有关此主题的更多细节，请参见第3章。

设计RAG特性管道架构

在这一部分，我们将讨论如何设计 LLM Twin 应用的 RAG 特性管道。我们计划使用一个批处理设计，定期从 MongoDB 数据仓库获取数据，进行处理，并将其加载到 Qdrant 向量数据库中。那么，第一个问题是：

“为什么选择批处理管道？”

但在回答之前，我们首先需要理解批处理架构与流式设计的区别。

批处理管道

在数据系统中，批处理管道指的是一种数据处理方法，其中数据在预定的时间间隔和较大数据量的基础上收集、处理和存储，这些数据集合称为“批次”。

与此不同的是流式数据处理，它是实时的，即数据在到达时立即处理。批处理管道的基本步骤如下：

  1. 数据收集：从多个来源收集数据并存储，直到积累到足够的量进行处理。这些来源可以包括数据库、日志、文件等。
  2. 定期处理：数据处理按固定的时间间隔进行，比如每小时或每天。在这段时间内，收集到的数据会批量处理，涉及数据清洗、转换、聚合等操作。
  3. 数据加载：处理后的数据会加载到目标系统中，比如数据库、数据仓库、数据湖或特性存储中。处理过的数据可以用于分析、查询或进一步处理。

批处理管道非常适合处理不需要实时处理的大量数据，它具有以下优势：

  • 效率：批处理可以更高效地处理大量数据，优化资源分配并支持并行处理。
  • 复杂处理：批处理管道能够执行复杂的数据转换和聚合操作，而这些操作可能对于实时处理而言过于消耗资源。
  • 简洁性：批处理系统的架构通常比流式系统简单，实施和维护较为容易。

批处理管道与流式管道的对比

流式应用的核心元素包括一个分布式事件流平台（如 Apache Kafka 或 Redpanda）来存储来自多个客户端的事件，以及一个流式引擎（如 Apache Flink 或 Bytewax）来处理这些事件。为了简化架构，你也可以使用像 RabbitMQ 这样的队列来存储事件，直到它们被处理。

方面 批处理管道 流式管道
处理调度 在固定的时间间隔（例如每分钟、每小时、每天）处理数据 实时处理数据，尽量减少延迟
更高效地处理大量数据，优化资源分配和并行处理 处理单一数据点，提供即时的更新，快速响应变化
处理复杂性 能够执行复杂的数据转换和聚合 设计用于处理高速度的数据流并保证低延迟
应用场景 适用于不需要即时处理数据的场景，常用于数据仓库、报告、ETL过程和特征管道。 适用于需要实时分析、特性、监控和事件驱动架构的应用
系统复杂性 相对于流式管道，系统实现和维护更为简单 由于需要低延迟处理、容错性和可扩展性，流式管道更复杂，工具也更先进且复杂

例如，流式管道在社交媒体推荐系统（如TikTok）中非常强大。在这种系统中，用户行为变化频繁。例如，用户可能在某一时刻只想看小狗视频，但15分钟后会厌倦并想看一些更严肃的内容，如教育内容或新闻。这要求推荐系统必须实时捕捉用户行为的变化，以保持用户的兴趣。如果使用批处理管道（例如每30分钟或每小时运行一次），则会延迟推荐生成，因此不适合这种场景。流式管道可以实时更新特定用户的特性，从而生成新的推荐。

然而，对于一些离线推荐系统（如电商平台或流媒体平台的推荐系统），用户行为变化不大，因此使用批处理管道每晚更新推荐结果是更简便、便宜的选择。

批处理管道还广泛用于ETL设计，用于从一个数据库提取、转换和加载数据。这在数据管道中非常常见，尤其是将数据从一个数据库迁移到另一个数据库的过程中，适用于汇总数据、进行分析等。

为什么选择批处理架构而不是流式架构？

对于 LLM Twin 的特性管道设计，我们选择了批处理架构，原因如下：

  • 不需要立即处理数据：尽管同步数据仓库和特性存储对RAG系统的准确性至关重要，但几分钟的延迟是可以接受的。因此，我们可以将批处理管道定期调度，每分钟运行一次，不断同步两个数据存储。因为数据量较小，整个数据仓库只有几千条记录，而不是几百万或几亿条，因此可以快速遍历并同步这两个数据库。
  • 简洁性：如前所述，实现流式管道比批处理管道复杂。实际应用中，尽量保持系统的简洁性，可以让系统更容易理解、调试和维护。此外，简洁性通常意味着更低的基础设施和开发成本。

最后，虽然批处理架构在某些场景下可能会进行冗余预测（例如在推荐系统中），但它的实现速度更快，且成本更低。因此，在产品上线初期使用批处理架构，然后根据需要逐步过渡到流式设计是一个常见的策略。

截屏2025-03-13 14.20.29

RAG特性管道核心步骤

大多数RAG特性管道由五个核心步骤组成：

  1. 数据提取：从 MongoDB 数据仓库中提取最新的文章、代码仓库和帖子。在此提取步骤中，通常会聚合所有需要处理的数据。
  2. 清理：从数据仓库中提取的数据已经是标准化并部分清理过的，但我们需要确保文本只包含有用信息，不重复且能够被嵌入模型解读。例如，我们需要在将文本传递给嵌入模型之前清理和标准化所有非ASCII字符。此外，为了保持信息的语义密度，我们决定将所有URL替换为占位符，并去除所有表情符号。清理步骤没有科学详尽的标准和方法论，我们在测试的过程中可能需要反复迭代并改进它。
  3. 分块：根据每个数据类别和嵌入模型，您需要采用不同的分块策略。例如，当处理代码仓库时，您希望分块较大；而在处理文章时，您希望分块较小，或按段落进行分块。根据您的数据，您必须决定是否根据章节、部分、段落、句子或固定窗口大小来拆分文档。此外，您还必须确保分块大小不会超过嵌入模型的最大输入大小。这就是为什么通常会根据数据结构和模型的最大输入大小来分割文档。
  4. 嵌入：将每个分块单独传递给您选择的嵌入模型。从实现的角度来看，这一步通常是最简单的，因为像 SentenceTransformer 和 Hugging Face 这样的工具提供了大多数嵌入模型的高级接口。如之前所述，在这一阶段，最关键的决定是选择哪个模型以及是否对其进行微调。例如，我们使用了 SentenceTransformer 中的 all-mpnet-base-v2 嵌入模型，它相对较小，可以在大多数机器上运行。不过，我们可以在Hugging Face上的MTEB（https://huggingface.co/spaces/mteb/leaderboard）找到其他模型。
  5. 数据加载：最后一步是将嵌入的分块文档和其元数据（如作者、文档ID、内容、URL、平台和创建日期）结合起来。最终，我们将向量和元数据封装成与 Qdrant 兼容的结构并推送到向量数据库中。由于我们希望使用 Qdrant 作为特征向量数据的唯一来源，因此我们还将清理过的文档（在分块之前）保存到 Qdrant 中（Qdrant 支持）。

同步数据仓库和特性存储

正如本章中几次强调的那样，数据在不断变化，这可能导致数据仓库和特征存储之间的不同步。数据变更捕获（CDC）是一种策略，可以在不增加计算和 I/O 开销的情况下，最优地保持两个或多个数据存储之间的同步。它捕获对源数据库进行的任何 CRUD 操作，并将其复制到目标数据库。您可以选择在复制过程中添加预处理步骤。

在构建特性管道时，同步问题同样存在。一个关键的设计选择是如何将数据仓库与特征存储同步，以确保数据的时效性，适用于我们的特定用例。

在 LLM Twin 的用例中，我们选择了一种简单的朴素方法。我们实现了一个定期或手动触发的批处理管道。它从数据仓库读取所有原始数据，以批次方式处理，并将新的记录或更新的旧记录插入 Qdrant 向量数据库中。对于处理几千条或几万条记录的小型数据集，这种方法很好用。但我们的朴素方法也引出了以下问题：

  • 如果数据突然增长到百万条记录（或更高）怎么办？（可用性）
  • 如果某条记录从数据仓库中被删除，这如何反映到特性存储中？（一致性）
  • 如果我们只想处理数据仓库中新添加或更新的项，而不是所有项，该怎么办？（一致性）

在实现 CDC 时，您可以采取多种方法，但所有方法都使用推送或拉取策略：

  • 推送：在推送方法中，源数据库是主要驱动者。它主动识别并将数据修改传输到目标系统进行处理。这种方法确保了目标系统几乎即时更新，但如果目标系统无法访问，则可能会丢失数据。为了缓解这种情况，通常会使用消息系统作为缓冲区。
  • 拉取：拉取方法将源数据库的角色设置为更被动的角色，源数据库只记录数据变更。目标系统定期请求这些变更，并相应地处理更新。虽然这种方法减轻了源数据库的负担，但会引入数据传播的延迟。同样，消息系统在目标系统不可用期间至关重要，以防止数据丢失。

总结来说，推送方法非常适合需要立即访问数据的应用，而拉取方法更适合那些大规模数据传输且对实时更新要求不高的应用。基于此，业界有不同的方法来检测数据变化。以下是一些主要的 CDC 模式：

  • 基于时间戳：这种方法通过在数据库表中添加一个修改时间列，通常称为 LAST_MODIFIED 或 LAST_UPDATED，来跟踪数据变更。下游系统可以查询此列以识别自上次检查以来已更新的记录。尽管易于实现，但该方法只能跟踪变更，而不能跟踪删除操作，并且由于需要扫描整个表，会增加性能开销。
  • 基于触发器：这种方法利用数据库触发器，在 INSERT、UPDATE 或 DELETE 操作时自动记录数据变更，并将其存储在一个单独的表中，通常称为事件表。虽然这种方法提供了全面的变更跟踪，但由于每个事件都涉及额外的写操作，它可能会影响数据库的性能。
  • 基于日志：数据库维护事务日志以记录所有数据修改，包括时间戳。尽管这些日志主要用于恢复，但也可以用来将数据变更实时传播到目标系统。该方法对源数据库的性能影响最小。它的最大优势是避免了对源数据库的额外处理开销，捕获所有数据变化，并且无需修改数据库模式。然而，它的缺点是缺乏标准化的日志格式，导致不同的数据库之间存在差异。

基于这些 CDC 技术，我们可以在 RAG 特性管道中快速实现一个基于时间戳的拉取策略，以在数据增长时更有效地同步数据仓库和特性存储。我们的实现仍然是拉取式的，但不会检查源数据库中的最后更新时间字段；它只会从数据仓库中拉取所有数据。

然而，在业界中，最流行和最优的技术是基于日志的方式。它不会对源数据库增加任何 I/O 开销，延迟低，支持所有 CRUD 操作。最大缺点是开发复杂性高，需要一个队列来捕获所有的 CRUD 事件，并使用流式管道来处理这些事件（参考缓存中间件和数据库之间数据一致性的实现方案）。

[!CAUTION]

后省略 ZenML 框架内的实现代码

五、监督微调

监督微调（Supervised Fine-Tuning, SFT）是将大语言模型（LLM）应用于实际场景中的关键步骤。在初始的预训练阶段，LLM 学习预测序列中的下一个词符，而 SFT 则通过精心挑选的指令和相应答案对模型进行微调。这个过程有两个主要目的：

  • 一是教会模型理解并遵循特定的对话格式， 高效地将其转变为对话代理；
  • 二是让模型将广泛的知识基础调整到特定任务或专业领域，从而在特定领域中表现更好。

SFT 的重要性在于它能够弥合模型的通用语言理解与实际应用之间的差距。通过将模型暴露于期望的输入输出模式，SFT 可以使 LLM 的行为与特定目标对齐，无论是任务完成（例如摘要或翻译）还是领域专业知识（如医学或法律知识）。这种定制化的方法不仅提高了模型在目标领域的表现，还增强了其跟随指令和生成更相关、更连贯响应的能力。

本章我们将讨论以下内容：

  • 创建高质量的指令数据集
  • SFT 技术
  • 实践中的微调实现

通过本章的学习，您将能够创建自己的指令数据集，并高效地对 LLM 进行微调。

指令数据集

创建指令数据集

在大多数应用场景中，创建指令数据集是微调过程中的难点之一。这是由于多种因素。大多数应用场景可以与原始文本关联，但找到自然的指令和答案对的情况却不多。原始文本需要转化为包含指令和答案的格式（<指令, 答案>）。此外，数据的质量也至关重要。因此，很多时间都会花费在手动检查和验证单个样本上。这个仔细的审查有助于确保数据集准确且对模型训练有用。

screenshot_2025-03-13_15.09.36

上图展示了本章涵盖的后训练数据管道的概述。在本节中，我们将介绍一个通用框架来创建您自己的指令数据集，无论最终用例如何。然后，我们将利用第 3 章中抓取的数据并将其转换为指令数据集。

通用框架

指令数据集被定义为指令和答案的对。

  • instruction 是模型的输入，用作微调过程中的上下文。
  • answer(output) 是模型的期望输出。

在微调过程中，您可以选择训练模型以处理指令和答案，或者仅处理答案。指令和答案的对通常遵循特定的模板。某些指令模板（如 Alpaca）会引入额外的字段，如 inputs 和 system，这两个字段可以视为指令字段的子字段。在这种情况下，

  • inputs 包含模型完成指令所需的数据，
  • system 是一个元提示，用于引导模型的整体行为

以下是来自 SlimOrca 数据集的一个示例，其中包含 “system” 和 “instruction”：

System:
你是一个乐于助人的助手，总是提供解释。假装你是在回答一个五岁孩子的问题。

Instruction:
概念：建筑、商店、镇
写一句包含这些词的句子。

Output:
在我们的小镇上，有一家商店，里面有一个大建筑，人们去那里买他们最喜欢的玩具和糖果。

这个例子说明了如何使用 “system” 字段来定义模型的具体行为，比如“做一个乐于助人的助手，总是提供解释，并且根据五岁孩子的理解来调整回应 ”。“instruction” 字段提供了必要的数据（“概念”）和任务（“构建句子”）。而 output 字段则显示了期望的答案，虽然这不是唯一的答案，但它是一个高质量的回应。

为了构建指令数据集，我们希望收集的数据能代表模型实际应用中的需求。一旦收集了足够的样本，我们的目标是筛选出高质量的数据。在这个过程中，高质量的数据可以通过以下三个主要维度来描述：

  • 准确性：指样本的事实准确性和相关性。在指令数据集的上下文中，这意味着确保回应不仅是事实正确的，而且与相应的指令相关。高准确性对于训练能够提供可靠、可信信息的模型至关重要。
  • 多样性：高质量的数据集应该涵盖广泛的使用场景，包括部署的 LLM 可能遇到的查询和任务。这种多样性应包括不同的主题、上下文、文本长度和写作风格。通过代表性地采样数据，我们能够让模型培养出强大的指令跟随能力。
  • 复杂性：简单的或过于简化的样本对 LLM 能力提升帮助不大。相反，数据集应包括复杂的多步骤推理问题和具有挑战性的任务，以推动模型在处理复杂现实问题时的能力。这种复杂性有助于开发能够应对复杂任务的模型。

在接下来的部分，我们将探讨如何根据这些维度筛选和评估指令样本。

数据量

Hugging Face Hub 包含了大量的指令数据集（点击跳转），这些数据集既可以是通用的，也可以是为特定任务或领域设计的。在处理新用例时，寻找相关的开源数据集以用于微调是非常有益的。尤其当样本数量过少时（例如，少于 1,000 个样本），通过高质量的数据进行增量训练就变得至关重要。

确定理想样本的数量是一项困难的任务，因为数据的质量和模型的大小都会产生显著的影响。对于大型模型（例如，约 700 亿个参数），所需的样本数量可以低至 1,000 个高质量样本（参见参考文献中的 LIMA 论文）。而对于较小的模型（例如，约 70 亿个参数），它们需要更多的样本才能学习正确的对话模板。无论如何，数据的质量都是关键因素，样本数量越多，越是理想。

为了提供更多的参考数据，我们可以查看由公司和开源社区开发的微调模型。我们可以区分两种类型的微调：通用模型，旨在重现类似 GPT 模型的能力，以及任务或领域特定的模型，旨在优化其在特定应用中的表现。通用模型覆盖的主题更多，因此需要更多的样本。

任务特定与领域特定模型

任务特定模型和领域特定模型代表了微调 LLM 的两种不同方法。

  • 任务特定模型旨在在特定功能上表现出色，例如翻译、摘要或情感分析。这些模型通过集中训练在单一任务上，能够高效执行，即使在较小的模型尺寸下（通常少于 80 亿个参数）。任务特定微调所需的数据量通常较小，范围从 100 到 100,000 个样本不等。这使得任务特定的微调成为许多资源有限的应用的理想选择。
  • 领域特定模型旨在通过专门的知识调整 LLM，并使其熟悉特定领域的词汇和语言模式。这些模型在医学、法律、金融、电子商务、工程和酒店等领域尤为有价值。领域特定微调的数据需求可以根据领域的复杂性和广度大不相同。一些领域，如医学或法律，可能需要与通用微调一样多的数据，因为它们有庞大的技术语料库。而其他领域，如电子商务或酒店业，可能只需要较少的样本，更多类似于任务特定微调。

决定领域特定模型数据需求的关键因素是领域的“大小”（即其专业知识和词汇的广度）以及该领域在模型预训练数据中的表示情况。在原始训练数据中有良好表示的领域可能需要较少的微调，而那些较为专业或表示不足的领域可能需要更大规模的数据集。

数据筛选

在获取用于微调的数据时，任务特定和领域特定模型的处理方式有所不同。

  • 对于任务特定模型，数据筛选通常涉及从现有数据集中收集目标任务的示例，或者创建新的数据集。例如，可能需要收集原文和摘要文本对，用于摘要模型，或者收集不同语言的句子，用于翻译模型。
  • 领域特定数据筛选可能更具挑战性。它通常需要与领域专家合作，收集和验证相关文本、研究论文、技术文档和其他领域特定的内容。在某些情况下，可能需要与拥有大量专业信息库的组织或机构合作。这些数据的质量和相关性至关重要，因为它直接影响模型在目标领域中理解和生成内容的能力。

值得注意的是，提示词工程（Ptompting engineering）中“少量示例提示”（few-shot prompting）已成为微调的替代策略，尤其适用于任务特定的应用。这种方法通过在输入提示中提供一些目标任务的示例，充分利用大型强大的模型的能力。尽管在所有场景中它不能替代微调（例如，当您希望学习一个新领域时），但少量示例提示可以是一种有效的方式，帮助模型适应新任务，而无需进行大量的额外训练。

在实践中，任务特定和领域特定模型之间的界限有时会模糊。例如，一个针对医学诊断进行微调的模型，既可以视为任务特定（专注于诊断），也可以视为领域特定（专注于医学知识）。关键是理解微调过程的主要目标，并据此调整方法。

此时，我们应该已经收集了适合我们用例的数据集。下一步是通过基于规则的过滤、数据去重、数据去污染和数据质量评估来精炼样本的质量。

处理数据集

基于规则的过滤

基于规则的过滤是一种系统的数据质量控制方法，它依赖于明确定义的规则来评估和过滤数据样本。这些规则通常旨在解决常见的质量问题，涵盖从简单检查到更复杂的逻辑操作。基于规则的过滤的主要目标是通过去除不符合特定标准的样本，保持数据质量的高标准。

  • 长度过滤 是一种简单而有效的基于规则的过滤技术。这种方法通过设置数据集中响应的可接受长度阈值来工作。极短的回答往往缺乏足够的信息，难以具有意义，而过长的回答可能包含无关或冗余的内容。需要注意的是，适当的长度阈值会根据具体任务和领域而有所不同。例如，生成简洁摘要的数据集可能会有较低的最大阈值，而详细解释的任务可能会有更高的阈值。
  • 关键词排除 是另一种强大的基于规则的过滤技术，它侧重于样本的内容，而非结构。这种方法涉及创建一个包含低质量或不当内容相关的关键词或短语的列表，然后过滤掉包含这些术语的样本。关键词列表可以包括明显的低质量指标，如脏话或与垃圾邮件相关的术语，也可以包括特定领域的词汇，这些词可能表示无关或跑题的内容。例如，在一个为专业写作助手准备的数据集中，可能会排除包含俚语或不符合预期语气和风格的非正式表达的样本。
  • 格式检查 是推荐用于包含结构化数据或需要遵循特定格式要求的数据集的技术。这种技术确保所有样本遵循预期的格式，保持一致性，并有利于后续处理。格式检查对于包含代码示例、JSON 结构或其他格式化文本的数据集尤为重要。例如，在一个编程指导和解决方案的数据集中，可能会实施规则来验证代码示例是否语法正确，并遵循指定的样式指南。

基于规则的过滤在准备指令数据集时的优缺点：

  • 优点：
    • 速度和效率使其能够迅速应用于大量数据，具有高度的可扩展性。规则应用的一致性确保了对数据的统一处理，减少了人为错误和偏差。
    • 明确的过滤标准定义提供了透明性和可解释性，便于理解、审计和调整。自动化的基于规则的过滤减少了手动干预的需要，并能够实现持续的数据质量监控。
  • 缺点：
    • 预定义的规则可能缺乏捕捉语言和上下文复杂性的细微差别，从而可能导致有效但不常见的样本被误删。规则的二元性质（通过/失败）可能并不总是与语言和指令质量的细微差别相符。
    • 随着数据模式和质量标准的变化，规则需要定期审查和更新，以保持有效性。
    • 还有一个风险是，设计不良的规则可能无意中引入或加剧数据集中的偏差。

数据去重

数据集的多样性对于训练能够很好地泛化到新数据的模型至关重要。当数据集中包含重复或近似重复的样本时，可能会导致以下几个问题：

  • 过拟合：模型可能记住特定的例子，而不是学习一般性的模式。
  • 性能偏差：过度代表的数据点可能会使模型的性能偏向某些类型的输入。
  • 低效训练：冗余数据可能会增加训练时间，但并未提供额外的有价值信息。
  • 虚高的评估指标：测试集中的重复数据可能会导致性能估计过于乐观。

为了去重数据集，我们区分了精确去重和模糊去重：

  • 精确去重通过数据标准化、哈希生成和重复项移除的简单过程去除完全相同的样本。数据标准化对条目格式进行统一，例如将文本转换为小写。然后，使用像 MD5 或 SHA-256 等算法生成哈希，为每个条目创建唯一的哈希值。这些哈希值被比较，以找出匹配项，并移除重复项，只保留每个样本的一个实例。尽管这种方法对于完全相同的条目有效，但它并不能检测到近似重复或语义相似的内容，因此在这种情况下需要更先进的技术。
  • 最常用的模糊去重方法是 MinHash 去重。与其他模糊技术相比，MinHash 在保持高准确度的同时，大大降低了计算复杂度。MinHash 通过为每个数据项生成紧凑的表示（或签名）来工作。这些签名充当指纹，捕捉数据的本质，同时大大减少其维度。在实践中，MinHash 将数据项（如文本文档）转换为一组 shingles，应用多个哈希函数于这些集合，并选择最小的哈希值来形成签名向量。然后，可以通过类似 Jaccard 相似度的相似度度量来高效地识别近似重复项。

除了精确和模糊去重，语义相似度则采取不同的方法，重点关注文本的含义进行去重。此方法涉及使用各种自然语言处理技术将单词或整个样本转换为向量表示。像 Word2Vec、GloVe 和 FastText 等词嵌入模型将单个单词转换为稠密的向量，捕捉语义关系。

对于更具上下文感知的表示，像 BERT、Sentence Transformers 或 Cross-Encoders 等语言模型可以为整个句子或文档生成嵌入。获得这些向量表示后，可以通过比较向量之间的相似度来执行去重。常见的相似度度量包括余弦相似度或欧几里得距离。具有高相似度分数的样本（超过预定的阈值）可以被视为重复样本。对于大数据集，可以应用聚类技术对相似向量进行分组。方法如 K-means、DBSCAN 或层次聚类可以有效组织向量空间，识别表示语义相似内容的聚类。在每个聚类中，可以保留一个代表性样本，其他的则标记为重复项。

数据去污染

数据去污染是确保训练数据集中不包含与评估集或测试集相同或高度相似的样本的过程。此步骤对于确保模型评估质量和防止过拟合或记住测试数据至关重要。

数据去污染使用数据去重中的技术。精确匹配可以用来去除与评估集中的样本完全相同的训练样本。这可以通过哈希函数或直接的字符串比较来实现。接下来，我们还可以使用近似重复检测方法来识别并去除与评估样本非常相似的训练样本，即使它们不完全相同。这通常涉及像 MinHash 或基于 n-grams 或嵌入的相似度分数计算等技术。

进行数据去污染的一种简单方法是在数据去重阶段将评估集添加到指令数据集中。在这种情况下，我们希望确保只去除指令数据集中的样本，可以通过不同的方式实现（仅过滤第一个重复项、记录评估样本的索引等）。理想情况下，您可以在数据去重阶段自动添加评估集，以完全自动化此过程。如果您迭代多个版本的自定义基准，这种方法特别高效。

数据去污染的另一个方面是过滤掉可能来自与评估数据相同来源的样本。这可以通过检查重叠的短语、相似的句子结构或共同的元数据来实现。实践者还可以使用来源跟踪（追踪使用的数据来源）来识别并排除已知用于评估集的特定数据来源。

数据质量评估

数据质量评估是机器学习中的一个关键方面，尤其对于 LLMs（大语言模型）。该过程涉及评估数据集的多个特征，包括准确性、多样性和复杂性。虽然一些方面（如数学准确性）可以使用 Python 解释器等工具轻松验证，但评估主观或开放式内容仍然是一个挑战。

传统的数据质量评估方法包括人工标注，这通常提供高准确性，但需要大量资源。为了应对可扩展性问题，已经开发出使用机器学习技术来自动化评估过程的方法。这些方法包括 使用 LLMs 作为评估者、奖励模型和为质量预测训练的分类器。

LLM 作为评估者 的策略涉及提示 LLM 来评估每个样本的质量。这种方法因其灵活性和易用性而广受欢迎，尽管它也存在一些挑战。不同的 LLM 在任务上的表现水平不同，其评估通常与非专家的评估更为接近。对于领域特定的数据集，您可能希望使用领域特定的模型，而不是更强大的通用 LLM。比较评估方法（例如“答案 A 是否优于答案 B？”）通常优于绝对评分方法（例如“将答案 A 评分为 1 到 4 之间”），尽管两者都可以通过足够的提示工程在大规模上使用。

上一页
2026-03-24 22:26:53 # LLM #LLM #DeepLearning #Engineering
下一页