Hcc的Blog

嵌入式 · AI · 折腾不止

0%

动手做 RAG(2)

发表于 分类于 ，

做什么

  设计一些基础的 Pipeline，实现最基本的 Rag 功能。

开始做

文件入库的做法

  新建 Pipeline 文件夹，新建 base_pipeline.py 文件，创建 Pipeline 基类，后续所有的 Pipeline 都将继承这个类。

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

class BasePipeline(ABC):
    def __init__(self, **kwargs):
        pass

    @abstractmethod
    def run(self, **kwargs):
        """执行业务流程"""
        pass

  新建 ingestion_pipeline.py 文件，实现文件加载、解析、切片录入数据库。但是在写这个 Pipeline 之前，我们应该还实现一个 ProcessorFactor 功能，让文件自适应解析，而非我们制定 processor。在 file_process 文件夹中新建 processor_factory.py 文件，并写入如下内容：

1
2
3
4
5
6
7
8
9
10
11
12
13
from pathlib import Path
from file_process.PPTx_peocess import PPTxProcessor

class ProcessorFactory:
    @staticmethod
    def create(file_path:str):
        suffix = Path(file_path).suffix.lower()
        if suffix == ".pptx":
            return PPTxProcessor()
        else:
            raise ValueError(
                f"不支持的文件类型:{suffix}"
            )

  接下来在 pipeline 中新建 ingestion_pipeline 文件，写入如下内容：

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
from pipeline.base_pipeline import BasePipeline
from embedding .tongyi_embedding import TongyiEmbedding
from file_process.PPTx_peocess import PPTxProcessor
from vector_store.chroma_store import ChromaVectorStore
from file_process.processor_factory import ProcessorFactory

class IngestionPipeline(BasePipeline):
    def __init__(self, 
        embedding,
        vector_store,
        processor_factory=ProcessorFactory
    ):
        self.embedding = embedding
        self.vector_store = vector_store
        self.processor_factory = processor_factory


    def run(self, file_path: str):
        """执行业务逻辑"""
        processor = self.processor_factory.create(file_path=file_path)
        chunks = processor.run(file_path=file_path)
        print(f"已切分 {len(chunks)} 个切块...")
        vector_chunks = self.embedding.embed(chunks=chunks)
        print("向量化已完成...")
        self.vector_store.add(vector_chunks)
        print("文件成功入库！")
        return

  在 main.py 中调用，非常成功，文件入库。

调用 LLM 回答

  首先还是得建立 LLM 基类，后期就可以通过继承基类来实现不同厂家的调用。新建 llm 文件夹，建立 base_llm.py，输入如下内容：

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

class BaseLLM(ABC):

    @abstractmethod
    def invoke(self, prompt: str) -> str:
        """调用大模型进行回答"""
        pass

  新建 deepseek_llm.py，输入如下内容

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
from .base_llm import BaseLLM
from openai import OpenAI
from langchain_core.messages import (
    BaseMessage,
    SystemMessage,
    HumanMessage,
    AIMessage
)
from config import (
    DEEPSEEK_API_KEY,
    DEEPSEEK_BASE_URL
)

class DeepseekLLM(BaseLLM):
    def __init__(self, model):
        self.client = OpenAI(
            api_key=DEEPSEEK_API_KEY,
            base_url=DEEPSEEK_BASE_URL,
        )
        self.model = model

    def invoke(self, prompt: list[BaseMessage]) -> str:
        openai_messages=[]
        for msg in prompt:
            if isinstance(msg, HumanMessage):
                role = "user"
            elif isinstance(msg, SystemMessage):
                role = "system"
            elif isinstance(msg, AIMessage):
                role = "assistant"
            else:
                raise ValueError(
                    f"不支持的消息类型:{type(msg)}"
                )
            openai_messages.append(
                {
                    "role": role,
                    "content": msg.content
                }
            )
        response = self.client.chat.completions.create(
            model=self.model,
            messages=openai_messages
        )

        return response.choices[0].message.content

  经过这两步后，我已经明白如何实现一个 LLM Client。那么现在就让我们使用 LangChain 的官方 ChatModel，嵌入我们的系统。安装包：

1
pip install langchain-openai

  在 deepseek_llm.py 中写下如下内容，然后在主程序中调用该 Client：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    BaseMessage,
    SystemMessage,
    HumanMessage,
    AIMessage
)
from config import (
    DEEPSEEK_API_KEY,
    DEEPSEEK_BASE_URL
)

Client = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
)

  测试代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from llm.deepseek_llm import Client
from langchain_core.messages import (
    HumanMessage,
    SystemMessage
)

messages = [
    SystemMessage(
        content="你是企业知识助手"
    ),

    HumanMessage(
        content="hello，你是谁"
    )
]
response = Client.invoke(messages)
print(response.content)

  如果需要流式传输的效果，那么就需要在 ChatModel 创建时加入关键字 stream=True，然后在测试时写下如下代码，即可实现流式传输效果：

1
2
3
4
5
6
for chunk in deepseek_client.stream(messages):
    print(
        chunk.content,
        end="",
        flush=True
    )

  接下来我们应该实现文本的召回。在这一步需要先将提问的文本转换为向量，在 tongyi_embedding.py 中新建 embed_text 函数，内容如下：

1
2
3
4
5
6
7
def embed_text(self, text:str) -> list[float]:
    response = self.client.embeddings.create(
        model=self.model,
        input=text
    )

    return response.data[0].embedding

  在 chroma_store.py 中新增 search 函数，实现在数据库中的相似向量检索：

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
def search(self, query_vector: list[float], top_k:int = 5)-> list[Chunk]:
    result = self.collection.query(
        query_embeddings=[
            query_vector
        ],
        n_results=top_k
    )
    print(result)

    # 转换为 Chunk
    documents = result["documents"][0]

    metadatas = result["metadatas"][0]

    distances = result["distances"][0]
    chunks = []
    for doc, metadata, distance in zip(documents, metadatas, distances):
        chunks.append(Chunk(
                chunk_content=doc,

                metadata={
                    **metadata,
                    "distance":distance
                }
            )
        )
    return chunks

  在经过转化，返回的就是相似的 chunks 列表。那么接下来我们就应该做上下文管理了。新建 context 文件夹，新建 rag_context.py 和 context_builder.py 文件。
  在 rag_context.py 文件中，我们定义专门用于 RAG 的专属上下文类型，代码如下：

1
2
3
4
5
6
7
8
9
from dataclasses import dataclass, field

@dataclass
class RAGContext:
    query: str
    documents: list
    history: list= field(default_factory=list)
    system_prompt: str = ""
    metadata: dict = field(default_factory=dict)

  在 context_builder.py 中，我们创建 ContextBuilder 类，我们将在其中实现各种类型的上下文构建。

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
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, BaseMessage
from .rag_context import RAGContext

class ContextBuilder():
    def __init__(self):
        pass

    def rag_context_build(self, context: RAGContext) -> list[BaseMessage]:
        messages=[]

        messages.append(
            SystemMessage(
                content=context.system_prompt
            )
        )

        docs_text = "\n\n".join(
            [
                doc.chunk_content
                for doc in context.documents
            ]
        )
        
        user_content = f"""

参考资料:

{docs_text}

问题：

{context.query}

"""
        messages.append(
            HumanMessage(content = user_content)
        )
        return messages

  新建 query_pipeline.py 文件，实现 rag 自动化，如下：

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
from .base_pipeline imuilder import ContextBuilder
from context.rag_context import RAGContext

SYSTEM_PROMPT = "你是企业知识助手，请严格按照提示词中提供的资料回答问题，如果资料不全无法回答，请如实说明。"
port BasePipeline
from context.context_b
class QueryPipeline(BasePipeline):
    def __init__(self, client, embedding, store):
        self.context_builder = ContextBuilder()
        self.client = client
        self.embedding = embedding
        self.store = store

    def run(self, query: str) -> str:
        """运行 RAG 流程"""
        query_vector = self.embedding.embed_text(query)
        docs = self.store.search(query_vector)
        context= RAGContext(
            query=query,
            documents=docs,
            system_prompt=SYSTEM_PROMPT
        )
        messages = self.context_builder.rag_context_build(context=context)
        answer = self.client.invoke(messages)

        return answer

优化

  到目前为止，我们已经实现了 RAG 的最小链路，但是文件结构仍需优化。我们询问 codex 如何优化这个架构，其给出如下建议：

  1、新建独立领域层，集中定义常用的类型，解除反向依赖。
  2、重构入库链路，将 PPTX 加载、清洗、切片拆成基础设施适配器，通过处理器注册表按扩展名选择实现；未知格式返回明确的“不支持文件类型”错误。
  3、重构查询链路。拆分为RetrievalService、PromptBuilder 和 AnswerService：Retrieval 负责查询向量化、top-k 检索、阈值过滤；PromptBuilder 负责上下文格式、来源编号和 token/字符预算；AnswerService 负责组合调用并返回结构化答案。
  4、集中配置。使用不可变 Settings 管理 API、模型、Chroma 路径、collection、检索参数、切片参数和网络策略。

整体框架

  codex 修改完成后，我们再次学习代码结构设计。经过更新后的项目采用了四层架构：

1
2
3
4
5
6
7
presentation     用户交互层
      ↓
application      业务编排层
      ↓
domain           领域模型与接口层
      ↑
infrastructure   外部技术实现层

  bootstrap.py 负责把四层组件组装取来，config.py 则负责提供配置

各层职责及文件作用

Domain：领域层

  领域层定义系统中的核心数据和接口，不关心 Chroma、OpenAI SDK 或 PPTX 等具体技术，其理想依赖关系如下，应用层使用领域接口编排业务，基础设施层实现领域接口，领域层不依赖其他层：

1
Application ──→ Domain ←── Infrastructure

  domain/__init__.py，标记 domain 为 Python 包，并说明该目录存放核心领域类型及端口，没有额外业务逻辑。
  domain/models.py，定义整个 RAG 系统使用的数据模型，如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@dataclass(frozen=True)
class Document:
    page_content: str
    metadata: Metadata = field(default_factory=dict)


@dataclass(frozen=True)
class Chunk:
    chunk_id: str
    chunk_content: str
    metadata: Metadata = field(default_factory=dict)

@dataclass(frozen=True)
class ChatMessage:
    role: Literal["system", "user", "assistant"]
    content: str

  domain/prots.py，使用 Python Protocol 定义系统组件应遵循的接口。

Application：应用层

  应用层描述了系统要完成的业务，例如如何完成一次文档入库、如何完成一次检索。它不应该关心 API 请求细节和数据库内部实现。

  application/errors.py 定义了统一的应用异常，如下：

1
2
3
4
5
6
RAGError
├── ConfigurationError
├── DocumentProcessingError
├── EmbeddingError
├── StorageError
└── GenerationError

  CLI 只需要捕获 RAGError，不必分别理解 OpenAI、Chroma 和 python-pptx 的底层异常。

  application/ingestion.py，定义 IngestionService，负责完整的文档入库流程：选择处理器、处理文件、处理文档 ID、批量向量化、写入向量库。它不会直接实现 PPTX 解析或 API 请求，只负责编排这些组件。
  application/retrieval.py，定义了 RetrievalService，负责：检查查询是否为空；调用 Embedder 生成查询向量；使用 SearchOptions 查询向量库；返回 SearchResult。top_k 和最大距离阈值在构造服务时注入。
  application/prompting.py，定义系统提示词和 Prompting.py。它将检索格式化，然后构造系统提示词、用户提示词、参考资料及用户问题。
  application/answering.py，定义 AnswerService，是查询链路的总编排器。

Infrastructure：基础设施层

  基础设施层处理外部技术，如读取 PPT，调用千问 Embedding 等等。

  infrastructure/loaders/registry.py 根据文件扩展名选择处理器。
  infrastructure/loaders/pptx.py 定义 PPTXProcessor，负责完整的 PPTX 预处理。
  infrastructure/embeddings/openai_compatible.py 实现 OpenAI 兼容的 Embedding 适配器。
  infrastructure/vectorstores/__init__.py 实现 Chroma 向量数据库适配器。

Presentation：交互层

  交互层负责接收用户输入、显示结果和处理命令行参数，不包含核心 RAG 逻辑。

异常处理机制的重构

  经过 Codex 优化后，该项目采用了统一应用异常 + 基础设施异常转换 + CLI 集中处理的机制，其核心目标为不让用户直接看到 OpenAI SDK、Chroma、文件系统等底层异常，而是把它们转换成 RAG 业务能够理解的异常。其整体结构如下：

1
2
3
4
5
6
7
8
9
10
11
底层库异常
OpenAI / Chroma / python-pptx / 文件系统
                 ↓ 转换
应用级异常
EmbeddingError / StorageError / ...
                 ↓ 传播
Application Service
                 ↓ 传播
CLI 统一捕获
                 ↓
向用户打印友好错误并返回退出码 1

  其异常类型集中定义在 application/errors.py 文件中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
class RAGError(Exception):
    """Base error exposed by the application layer."""


class ConfigurationError(RAGError):
    pass


class DocumentProcessingError(RAGError):
    pass


class EmbeddingError(RAGError):
    pass


class StorageError(RAGError):
    pass


class GenerationError(RAGError):
    pass

  定义自己的异常，CLI 只需要处理 RAGError 即可，无需关注底层库，可以去耦合。在项目中，大量使用了如下写法：

1
2
except Exception as exc:
    raise EmbeddingError("文档向量化失败") from exc

  第一句：如果 try 里面发生任何继承自 Exception 的错误，把这个错误对象保存到变量 exc。第二行表示主动抛出自己定义的 EmbeddingError 异常，from exc 会保留原始错误以便排查错误原因。
  其中，exc 是原始底层异常，EmbeddingError 是新的应用异常，from exc 将两者连成一条异常链。

Py 基础知识补充

Protocol 的使用

  我们之前使用的是 ABC 基类，如下：

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

class BaseEmbedding(ABC):

    @abstractmethod
    def embed(self, chunks):
        pass

  然后需要在创建新的类时继承基类。但是如果你这样定义：

1
2
3
4
class MyEmbedding:

    def embed(self, chunks):
        print("我的Embedding")

  那么 MyEmbedding 和 BaseEmbedding 就没有继承关系。

  Protocol 的思想不同，它只关心你有没有这个能力，比如：

1
2
3
4
5
6
7
from typing import Protocol


class EmbeddingProtocol(Protocol):

    def embed(self, chunks):
        ...

  现在在定义类，如下：

1
2
3
4
5
6
7
8
9
class TongyiEmbedding:

    def embed(self, chunks):
        print("通义")

class LocalEmbedding:

    def embed(self, chunks):
        print("本地模型")

  都符合 EmbeddingProtocol，因为他们都有 embed()。接下来举一个具体的实例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class Duck:

    def speak(self):
        print("嘎嘎")


class Person:

    def speak(self):
        print("你好")

def talk(obj):

    obj.speak()

talk(Duck())
talk(Person())

  Protocol 只关心方法不关心类型。domain/prots.py 就是使用了 Protocol 定义了不同组件的必需方法。

__init__ 的使用

  假设目录结构为

1
2
3
4
5
infrastructure/
├── __init__.py
└── embeddings/
    ├── __init__.py
    └── openai_compatible.py

  当 Python 看到 __init__.py 时，会把对应目录当作一个可导入的包，当一个包第一次被导入时，Python会执行这个包中的 __init__.py。

1
2
import infrastructure
import infrastructure.embeddings

  当导入 infrastructure.embeddings 时，其大致顺序为：

1
2
3
1. 执行 infrastructure/__init__.py
2. 执行 infrastructure/embeddings/__init__.py
3. 得到 infrastructure.embeddings 模块对象

  同一个 Python 进程中，模块通常只会在第一次导入时执行，之后会从 sys.modules 缓存中复用。

Python 异常机制基础

  Python 中错的错误本质是一个异常对象。例如

1
2
3
4
5
try:
    a = 1 / 0

except Exception as e:
    print(e)

  这样运行程序，会输出 division by zero，这里的 e 就是异常对象。
  try / except 基本结构是 py 中常用的错误检测机制。例如：

1
2
3
4
5
try:
    x = 1 / 0

except ZeroDivisionError:
    print("不能除0")

  执行时先进入 try，发生异常后跳转 except，执行处理代码。