当前位置：首页 > news >正文

langchain 快速入门(三)：搭建RAG知识库

news 2026/7/11 17:28:57

简介

LLM大模型一般训练的数据都是滞后的，这是就需要用到RAG知识库，RAG知识库可以降低大模型在输出答案时的幻觉，也能够让大模型知识拓展。

知识库架构知识

检索流程图

用户输入 (User Query) | v +-----------------------+ | 提示词 (Prompt) | +-----------------------+ | | (1) 转化为向量 (Embedding) v +-----------------------+ | 文字向量模型 (EMB) | +-----------------------+ | | (2) 相似度检索 (Search) v +-----------------------+ +-----------------------+ | RAG 向量数据库 (DB) | <-----> | 本地知识库/文档集 | +-----------------------+ +-----------------------+ | | (3) 召回相关片段 (Context) v +-----------------------+ | 大语言模型 (LLM) | <--- (将提示词与背景片段拼接) +-----------------------+ | | (4) 最终生成 (Generation) v +-----------------------+ | 输出结果 | +-----------------------+

RAG原理解析

构建知识库的流程如下：
文档内容切片->文字向量化->向量数据库

文档内容切片

LLM 有上下文长度限制，且向量检索在短文本上更精确。
文本切片方法有：

按字数切片
按句切片
递归切片
不管是哪个切片方法，目的是保留语义的完整性，因此不是某个好或某个坏，根据实际需求进行选择。

文字向量化

将每个切片文字块转换为关系向量

关系向量是怎么来的？这些是通过对模型进行大量的上下文训练，得到的词语与词语，句子与句子的语义关系上的关联，这是LLM的基础。
不同的文字向量模型输出的向量矩阵也会不同，因此在构建和检索时需要用同一个文字向量模型进行转换。

向量数据库

跟平时用的MySql，MS sql等本质上差不多，只不过将文本换成了向量矩阵的增删改查
检索的方法（这里就列举几个）：

两个向量的余弦角大小
两个向量的欧氏举例
两个向量的分类区域

用文本也能搜索，为什么要用向量呢？文本的数据库检索一般通过正则搜索，这种无法关联语义进行检索，因为我们无法保证大模型在检索时能够按照固定的格式。举个例子：
数据库中存储：小明一个月工资3000元
搜索：小明这个月多少钱
正则搜索会直接略过，而向量搜索就能够搜到。

构建RAG知识库

本次都利用千问的语言模型，和FAISS向量数据库演示，来演示

案例

/* by 01022.hk - online tools website : 01022.hk/zh/quchong.html */ from operator import itemgetter import os from langchain_community.chat_models.tongyi import ChatTongyi from langchain_community.vectorstores import FAISS from langchain_community.embeddings import DashScopeEmbeddings from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.documents import Document from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser from langchain_core.messages import HumanMessage, AIMessage from sqlalchemy import true #apikey os.environ["DASHSCOPE_API_KEY"] = "你的apikey" FAISS_PATH = "my_faiss_index" raw_text = """ 【公司内部机密：代号“深蓝计划”】 1. 项目目标：开发一款能听懂猫语的翻译器。 2. 核心技术：基于Transformer的“喵声波”分析算法。 3. 团队暗号：如果有人问“今天天气怎么样？”，必须回答“我想吃鱼”。 4. 截止日期：2026年12月31日。 5. 经费预算：仅剩50元人民币，主要用于购买猫条。 """ #初始化文本 docs = [Document(page_content=raw_text)] text_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20) split_docs = text_splitter.split_documents(docs) #初始化模型 llm = ChatTongyi(model="qwen-plus") embeddings = DashScopeEmbeddings(model="text-embedding-v1") #创建向量数据库 if os.path.exists(FAISS_PATH): print("向量数据库已存在") ragdb = FAISS.load_local(FAISS_PATH, embeddings, allow_dangerous_deserialization=True) else: print("创建向量数据库") ragdb = FAISS.from_documents(split_docs, embeddings) ragdb.save_local(FAISS_PATH) #构建提示词chain def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs) final_prompt = ChatPromptTemplate.from_messages([ ("system", """ 你是一个专业的问答助手，你的任务是根据上下文简洁的回答用户的问题。 <context> {context} </context> """), MessagesPlaceholder(variable_name="history"), ("human", "{input}") ]) chain = ( #查询rag RunnablePassthrough.assign( context = itemgetter("input") | ragdb.as_retriever() | format_docs ) | RunnablePassthrough.assign( answer = {"input":itemgetter("input"), "context":itemgetter("context"), "history":itemgetter("history")} | final_prompt | llm | StrOutputParser() ) ) history = [] while true: input_q = input("我：") respond = chain.invoke({ "input": input_q, "history": history}) print("answer:" + respond["answer"]) print("=="*30) history.append(HumanMessage(content=input_q)) history.append(AIMessage(content=respond['answer']))

代码解释

代码的流程如下：

初始化RAG：文本切片->文本向量模型->构建向量数据库
询问ai：提示词->文本向量模型->向量数据库检索->组合prompt->喂给LLM->回答问题->记录历史对话

文本切片

/* by 01022.hk - online tools website : 01022.hk/zh/quchong.html */ docs = [Document(page_content=raw_text)] text_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20) split_docs = text_splitter.split_documents(docs)

这里利用了langchain提供的文本分词器RecursiveCharacterTextSplitter（递归分词）

构建向量数据库

llm = ChatTongyi(model="qwen-plus") embeddings = DashScopeEmbeddings(model="text-embedding-v1") #创建向量数据库 if os.path.exists(FAISS_PATH): print("向量数据库已存在") ragdb = FAISS.load_local(FAISS_PATH, embeddings, allow_dangerous_deserialization=True) else: print("创建向量数据库") ragdb = FAISS.from_documents(split_docs, embeddings) ragdb.save_local(FAISS_PATH)

这部分要注意：新版FAISS读取现有数据库要设置：allow_dangerous_deserialization=True，不然会报错

提示词模板

final_prompt = ChatPromptTemplate.from_messages([ ("system", """ 你是一个专业的问答助手，你的任务是根据上下文简洁的回答用户的问题。 <context> {context} </context> """), MessagesPlaceholder(variable_name="history"), ("human", "{input}") ])

之前没有讲到历史对话记录，这次补充下：

MessagesPlaceholder这个是langchain框架的占位符（其实是框架写好了prompt模板，告诉ai这个是历史对话），使用时将历史对话记录的数组放在这里设置的字段中，在添加历史对话时要使用相关的类进行声明对话（告诉ai这句话是ai说的还是用户说的）
history.append(HumanMessage(content=input_q)) history.append(AIMessage(content=respond['answer']))

Chain链的解释（核心逻辑）

chain = ( #查询rag RunnablePassthrough.assign( context = itemgetter("input") | ragdb.as_retriever() | format_docs ) | RunnablePassthrough.assign( answer = {"input":itemgetter("input"), "context":itemgetter("context"), "history":itemgetter("history")} | final_prompt | llm | StrOutputParser() ) )

Chain链流程：

查询RAG的chain：获取input字段->内容交给向量数据库检索->将检索的内容(数组)转换为字符串格式->保存到context字段并传递给下一个任务
询问LLM的chain：获取input,context,history字段->填充上面定义的prompt模板->喂给LLM模型->解析成文本并保存在answer字段