从0到1，LangChain+RAG全链路实战AI知识库

# 创建虚拟环境conda create -n rag-demo python=3.9conda activate rag-demo# 安装核心库pip install langchain langchain-community langchain-openai
pip install chromadb faiss-cpu  # 向量数据库pip install pypdf docx2txt  # 文档解析库

import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"# 如果使用国内服务商，可配置代理或使用兼容OpenAI接口的国产模型os.environ["OPENAI_BASE_URL"] = "https://api.your-provider.com/v1"

from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader, TextLoader# 加载指定目录下的所有PDF文件pdf_loader = DirectoryLoader(
    './knowledge_base/', 
    glob="**/*.pdf",
    loader_cls=PyPDFLoader)# 同时加载多种格式的文档loaders = [
    DirectoryLoader('./docs/', glob="**/*.txt", loader_cls=TextLoader),
    DirectoryLoader('./docs/', glob="**/*.pdf", loader_cls=PyPDFLoader),]documents = []for loader in loaders:
    documents.extend(loader.load())print(f"成功加载 {len(documents)} 个文档")

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 每个块的最大字符数
    chunk_overlap=50,      # 相邻块之间的重叠字符数
    length_function=len,
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""])chunks = text_splitter.split_documents(documents)print(f"文档分割为 {len(chunks)} 个文本块")

from langchain_openai import OpenAIEmbeddings# 使用OpenAI的text-embedding-ada-002模型embeddings = OpenAIEmbeddings(
    model="text-embedding-ada-002",
    disallowed_special=())# 备选方案：如果希望本地部署，可以使用HuggingFace的嵌入模型# from langchain_community.embeddings import HuggingFaceEmbeddings# embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-base-zh-v1.5")

from langchain_community.vectorstores import Chroma# 创建向量数据库并存入文档块vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 本地持久化存储)# 后续使用时可以直接加载已有的数据库# vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)

retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4})

retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 20, "lambda_mult": 0.5})

retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.5, "k": 4})

from langchain.prompts import ChatPromptTemplate

template = """你是一个专业的知识库助手。请基于以下提供的上下文片段，准确、简洁地回答用户的问题。要求：
1. 如果上下文中包含相关信息，请基于这些信息回答
2. 如果上下文中没有足够信息，请明确告知"知识库中没有找到相关信息"
3. 引用来源时请标注对应的文档片段编号
4. 保持回答的专业性和客观性

上下文信息：
{context}

用户问题：{question}

你的回答："""

prompt = ChatPromptTemplate.from_template(template)

from langchain_core.runnables import RunnablePassthroughfrom langchain_openai import ChatOpenAI# 初始化大语言模型llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0.3,  # 较低的温度保证回答的确定性
    max_tokens=500)# 定义文档格式化函数def format_docs(docs):
    return "\n\n".join([f"[{i+1}] {doc.page_content}" for i, doc in enumerate(docs)])# 构建RAG链rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt    | llm)

from langchain.memory import ConversationBufferMemoryfrom langchain.chains import ConversationalRetrievalChain

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    output_key="answer")qa_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=retriever,
    memory=memory,
    verbose=True)# 连续对话测试response1 = qa_chain.invoke({"question": "什么是RAG技术？"})response2 = qa_chain.invoke({"question": "它和传统搜索有什么区别？"})

# 准备测试问题及期望检索到的文档块test_queries = [
    {"query": "LangChain支持哪些向量数据库？", "relevant_docs": [3, 5, 7]},
    {"query": "如何设置文本分割的chunk_size？", "relevant_docs": [12, 13]},]# 计算检索指标def evaluate_retrieval(retriever, test_queries, k=5):
    total_precision = 0
    total_recall = 0
    
    for item in test_queries:
        retrieved_docs = retriever.get_relevant_documents(item["query"])
        retrieved_ids = [doc.metadata.get("chunk_id") for doc in retrieved_docs[:k]]
        
        relevant_ids = item["relevant_docs"]
        relevant_retrieved = set(retrieved_ids) & set(relevant_ids)
        
        precision = len(relevant_retrieved) / k
        recall = len(relevant_retrieved) / len(relevant_ids)
        
        total_precision += precision
        total_recall += recall    
    return {
        "avg_precision": total_precision / len(test_queries),
        "avg_recall": total_recall / len(test_queries)
    }

# 1. 使用异步检索提升并发能力async def async_retrieve(queries):
    tasks = [retriever.aget_relevant_documents(q) for q in queries]
    return await asyncio.gather(*tasks)# 2. 引入缓存机制减少重复计算from langchain.cache import InMemoryCacheimport langchain
langchain.llm_cache = InMemoryCache()# 3. 考虑使用更高效的向量索引# 切换到FAISS并设置IVF索引from langchain_community.vectorstores import FAISS
vectorstore = FAISS.from_documents(chunks, embeddings)vectorstore.save_local("faiss_index")# 检索时使用nprobe参数平衡速度和精度retriever = vectorstore.as_retriever(
    search_kwargs={"k": 4, "nprobe": 10})