LangChain构建RAG问答系统

import os

# 1.Load 导入Document Loaders
from langchain_community.document_loaders import PyPDFLoader
from langchain_community.document_loaders import Docx2txtLoader
from langchain_community.document_loaders import TextLoader

# 加载Documents
base_dir = '.\OneFlower' # 文档的存放目录
documents = []
for file in os.listdir(base_dir): 
    # 构建完整的文件路径
    file_path = os.path.join(base_dir, file)
    if file.endswith('.pdf'):
        loader = PyPDFLoader(file_path)
        documents.extend(loader.load())
    elif file.endswith('.docx'):
        loader = Docx2txtLoader(file_path)
        documents.extend(loader.load())
    elif file.endswith('.txt'):
        loader = TextLoader(file_path)
        documents.extend(loader.load())

# 2.Split 将Documents切分成块以便后续进行嵌入和向量存储
from langchain.text_splitter import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,  # 单条分块字符数（≤8192即可）
    chunk_overlap=50  # 重叠字符数（确保上下文连贯）
)
chunked_documents = text_splitter.split_documents(documents)
print(f"优化后文档分块数量：{len(chunked_documents)}")  # 查看分块数量


# 3.Store 将分割嵌入并存储在矢量数据库Qdrant中
import os
import requests
from typing import List
from langchain.embeddings.base import Embeddings
from langchain_community.vectorstores import Qdrant


# 最终完整版：支持批量拆分（适配阿里25条/批的限制）
class QwenEmbeddings(Embeddings):
    def __init__(self, api_key: str, base_url: str):
        self.api_key = api_key
        self.base_url = base_url
        self.model = "text-embedding-v1"
        self.max_batch_size = 25  # 阿里接口单次最大批量（固定为25，不可超）

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """嵌入文档：过滤空文本 + 批量拆分（每批≤25条）"""
        # 1. 过滤空文本（避免无效请求）
        valid_texts = [text.strip() for text in texts if text.strip()]
        if not valid_texts:
            return []

        all_embeddings = []  # 存储所有批量的嵌入结果
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

        # 2. 拆分批量（每批不超过 max_batch_size 条）
        for i in range(0, len(valid_texts), self.max_batch_size):
            batch_texts = valid_texts[i:i + self.max_batch_size]  # 取当前批次文本
            payload = {
                "model": self.model,
                "input": batch_texts
            }

            # 3. 逐批调用接口
            try:
                response = requests.post(
                    f"{self.base_url}/embeddings",
                    headers=headers,
                    json=payload,
                    timeout=30
                )
                response.raise_for_status()
                # 提取当前批次的嵌入结果
                batch_embeddings = [item["embedding"] for item in response.json()["data"]]
                all_embeddings.extend(batch_embeddings)  # 合并到总结果
                print(f"成功处理第 {i // self.max_batch_size + 1} 批嵌入（共 {len(batch_texts)} 条）")
            except Exception as e:
                print(f"处理第 {i // self.max_batch_size + 1} 批嵌入失败：{str(e)}")
                print(f"当前批次文本（前300字符）：{str(batch_texts[:2])[:300]}...")  # 打印部分文本排查
                raise

        # 4. 确保结果长度与有效文本长度一致（避免遗漏）
        assert len(all_embeddings) == len(valid_texts), "嵌入结果数量与文本数量不匹配！"
        return all_embeddings

    def embed_query(self, text: str) -> List[List[float]]:
        """嵌入查询：单条文本，无需拆分"""
        return self.embed_documents([text])[0] if text.strip() else []

embedding = QwenEmbeddings(
    api_key=os.environ.get("DASHSCOPE_API_KEY"),  # 确保环境变量已配置
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

vectorstore = Qdrant.from_documents(
    documents=chunked_documents,
    embedding=embedding,
    location=":memory:",  # 内存存储（测试用）
    collection_name="my_documents",
)

# 4. Retrieval 准备模型和Retrieval链
import logging # 导入Logging工具
from langchain_community.chat_models import ChatOpenAI # ChatOpenAI模型
from langchain.retrievers.multi_query import MultiQueryRetriever # MultiQueryRetriever工具
from langchain.chains import RetrievalQA # RetrievalQA链

# 设置Logging
logging.basicConfig()
logging.getLogger('langchain.retrievers.multi_query').setLevel(logging.INFO)

# 实例化一个大模型工具
llm = ChatOpenAI(
    api_key=os.environ.get("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model_name="qwen-flash",
)

# 实例化一个MultiQueryRetriever
retriever_from_llm = MultiQueryRetriever.from_llm(retriever=vectorstore.as_retriever(), llm=llm)

# 实例化一个RetrievalQA链
qa_chain = RetrievalQA.from_chain_type(llm,retriever=retriever_from_llm)

# 5. Output 问答系统的UI实现
from flask import Flask, request, render_template
app = Flask(__name__) # Flask APP

@app.route('/', methods=['GET', 'POST'])
def home():
    if request.method == 'POST':

        # 接收用户输入作为问题
        question = request.form.get('question')        
        
        # RetrievalQA链 - 读入问题，生成答案
        result = qa_chain({"query": question})
        
        # 把大模型的回答结果返回网页进行渲染
        return render_template('index.html', result=result)
    
    return render_template('index.html')

if __name__ == "__main__":
    app.run(host='0.0.0.0',debug=True,port=5000)