这篇文章上次修改于 442 天前，可能其部分内容已经发生变化，如有疑问可询问作者。

概念解析

RAG（Retrieval-Augmented Generation，检索增强生成）是一种结合信息检索（Retrieval）和文本生成（Generation）的技术，常用于大模型（如GPT、LLM）中，以提升回答的准确性和可控性。

其中信息检索涉及到知识库的构建和知识库检索

而文本生成涉及大模型的使用，通常是使用现成的大模型，例如qwen，deepseek等，主要利用了大模型的逻辑推理能力。

简单的来说，RAG应用实际并没有听着那么高级，更像是大模型的先进使用，不拘泥于只使用大模型进行问答，而是提供了参考文档，让大模型根据文档进行检索和回答。

例如构建校园问答应用，只需要构建校园信息的知识库，再在每次提问的时候，将初始提问和校园信息知识库进行结合为新提问，将新提问提供给大模型，大模型会结合新提问里面给予的参考信息进行回答。

信息检索

概念解析

Embedding

Embedding（嵌入）是一种将文本、图像或其他数据转换为高维数值向量的技术，使其能够在向量空间中表示语义相似性。

例如，语义相近的句子会被映射到相近的向量。例如，queen和king在Embedding里面的相似度高于queen和apple。

现在的Embedding大多基于句子级别，根据整个句子的语义动态生成。句意思更相近的，两个向量的余弦相似度较高。

Embedding 通常用于自然语言处理（NLP）、推荐系统和检索增强生成（RAG）等任务，常见模型包括 OpenAI Embeddings、BERT、SBERT 等。

对于以上我们可以得到一种新的方式，用于判断两个句子的相近程度，而许多厂商都有提供Embedding API，即使用同一个Embedding模型的情况下，两个句子的高维向量可以表示两个句子之间的联系。

对于langchain，有一定的预备接口以满足

os.environ["OPENAI_API_BASE"] = "<https://api.gptsapi.net/v1>" # 可以修改基础url
embedding = OpenAIEmbeddings(
    model="text-embedding-3-large",
    openai_api_key=key)

或者以下是一个构建Embeddings实例，此处使用自定义的Embeddings

import torch
from transformers import AutoTokenizer, AutoModel
from langchain_core.embeddings import Embeddings
from tqdm import tqdm

class MathBERTEmbeddings(Embeddings):
    def __init__(self, model_name="tbs17/MathBERT", device=None):
        self.model_name = model_name
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
        # 加载模型和分词器
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModel.from_pretrained(model_name).to(self.device)
        print("MathBERT模型加载完成")
    
    def embed_documents(self, texts):
        """为文档列表创建嵌入"""
        embeddings = []
        
        # 使用tqdm显示进度
        for text in tqdm(texts, desc="创建嵌入"):
            embeddings.append(self._embed_text(text))
            
        return embeddings
    
    def embed_query(self, text):
        return self._embed_text(text)
    
    def _embed_text(self, text):
        inputs = self.tokenizer(
            text, 
            return_tensors="pt", 
            max_length=512,
            padding="max_length",
            truncation=True
        ).to(self.device)
        with torch.no_grad():
            outputs = self.model(**inputs, output_hidden_states=True)
            last_hidden_state = outputs.last_hidden_state
            cls_embedding = last_hidden_state[:, 0, :].squeeze().cpu().numpy()
            
            return cls_embedding.tolist()

使用例：

vector = embedding.embed_query("hello")
print(vector[:3])
# [-0.024605071172118187, -0.0075481850653886795, 0.004001544788479805]

向量数据库

是知识库的基础，也就是用于存放我们参考信息的数据库，由于参考信息经过Embedding变成了多个高维向量集合，将其都存储进向量数据库，接下来的需求就是：

输入一个句子，根据Embedding的原理找到相近的数据作为参考依据。

为了查找和检索效率，产生了向量数据库，对相近向量的检索有特别的优化，提高了效率。

构建知识库

在检索增强生成（RAG）框架下，向量数据库主要用于存储和高效检索知识。典型流程如下：

数据准备

采集知识（如文档、论文、对话记录）
预处理文本（分段、去噪、去重）

生成向量（Embedding）

使用 Embedding API（如 OpenAI Embeddings、Hugging Face、BERT）
将文本转换为向量并存入向量数据库

存入向量数据库（Vector DB）

选择合适的数据库（如 FAISS、Milvus、Pinecone）
存储向量及其关联的原始文本

以下是一个使用例

def create_faiss_index_from_limo():
    try:
        with open(r"limo_test.json", "r", encoding="utf-8") as f:
            data = json.load(f)
            documents = [
                Document(
                    page_content=f"问题: {item['instruction']}\\n\\n解答: {item['output']}",
                ) 
                for item in tqdm(data, desc="处理数据")
            ]
    except Exception as e:
        print(f"加载数据时出错: {e}")
        return None
    
    try:
        embedding = OpenAIEmbeddings(
            model="text-embedding-3-large",
            openai_api_key=key)
    except Exception as e:
        print(f"创建Embeddings模型时出错: {e}")
        return None
    
    try:
        vectorstore = FAISS.from_documents(documents, embedding)
    except Exception as e:

        print(f"构建FAISS索引时出错: {e}")
        return None
    
    try:
        vectorstore.save_local("faiss_index")
        return vectorstore
    except Exception as e:
        print(f"保存索引时出错: {e}")
        return None

使用知识库

查询 & 召回（Retrieval）

用户输入问题（转换为向量）
在向量数据库中进行相似度搜索（如余弦相似度）
召回最相关的知识片段

最后将检索到的知识提供给大模型作为参考。

使用Langchain进行实践

Langchain集成了RAG的整个流程,

同时，为了满足langchain中模型的使用，修改了llm的类方法

class QwenMath(LLM):
    model: object = None
    tokenizer: object = None
    max_new_tokens: int = 512
    device: object = torch.device("cuda" if torch.cuda.is_available() else "cpu")

    def __init__(self, model_name: str):
        super().__init__()

        # 加载模型和tokenizer
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype="auto",
            device_map={"": 0}
        )

    def chat(self, model, tokenizer, msgs):
        text = tokenizer.apply_chat_template(
            msgs,
            tokenize=False,
            add_generation_prompt=True
        )

        model_inputs = tokenizer(text, return_tensors="pt", max_length=512, truncation=True).to(self.device)

        generated_ids = model.generate(
            **model_inputs,
            max_new_tokens=self.max_new_tokens,
        )
        generated_ids = [
            output_ids[len(input_ids):]
            for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
        ]
        response = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]

        return response

    def _call(self, prompt: str, stop=["<|user|>"], run_manager=None, **kwargs):
        msgs = []
        for line in prompt.splitlines():
            if line.startswith("System: "):
                msgs.append({'role': 'system', 'content': line[8:]})
            elif line.startswith("Human: "):
                msgs.append({'role': 'user', 'content': line[7:]})

        response = self.chat(self.model, self.tokenizer, msgs)
        return response

    @property
    def _llm_type(self) -> str:
        return "Qwen2"

以下是一个完整使用例：

from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from QwenMath import QwenMath
from langchain_community.vectorstores import FAISS

from ed import MathBERTEmbeddings
import os
os.environ["KMP_DUPLICATE_LIB_OK"] = "TRUE"

embeddings = MathBERTEmbeddings()

# ✅ 1. 创建 LLM
llm = QwenMath("Qwen2.5-Math-7B")

# ✅ 2. 创建知识库检索器
vectorstore = FAISS.load_local("faiss_mathbert_index", embeddings,
                               allow_dangerous_deserialization=True)  # 你的向量存储
retriever = vectorstore.as_retriever()

# ✅ 3. 改写查询，使其带有历史上下文
contextualize_q_system_prompt = (
    "Given a chat history and the latest user question "
    "which might reference context in the chat history, "
    "formulate a standalone question which can be understood "
    "without the chat history. Do NOT answer the question, just "
    "reformulate it if needed and otherwise return it as is."
)
contextualize_q_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", contextualize_q_system_prompt),
        MessagesPlaceholder("chat_history"),  # 这里用于存储聊天历史
        ("human", "{input}"),  # 用户的新问题
    ]
)
history_aware_retriever = create_history_aware_retriever(
    llm, retriever, contextualize_q_prompt
)

# ✅ 4. 让 LLM 结合知识库回答问题
qa_system_prompt = (
    "你是一个数学方面的助手，需要帮助用户解决数学问题，要求答案正确，并且要提供思考步骤。\\n\\n"
    "{context}"
)
qa_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", qa_system_prompt),
        MessagesPlaceholder("chat_history"),
        ("human", "{input}"),
    ]
)
question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)

# ✅ 5. 创建 RAG（检索增强生成）链
rag_chain = create_retrieval_chain(
    history_aware_retriever, question_answer_chain
)

# ✅ 6. 运行
chat_history = []  # 用于存储历史消息
query = "求解Inx＝x-e的解"

result = rag_chain.invoke({"input": query, "chat_history": chat_history})
print(result["answer"])

Mai Icy

RAG的认识与构建

概念解析

信息检索

概念解析

Embedding

向量数据库

构建知识库

使用知识库

使用Langchain进行实践

Mai Icy