新一代AI工具深度实战：基于LangChain和GPT-4构建企业级RAG知识库问答系统

┌─────────────────┐
│  用户提问       │
└────────┬────────┘
         ▼
┌─────────────────────────────────────┐
│  LangChain 检索链 (RetrievalQA)     │
│  ┌────────────────────────────┐     │
│  │ 1. 文本嵌入 (Embedding)     │     │
│  │ 2. 向量相似性检索 (ANN)     │     │
│  │ 3. 上下文压缩 / 重排序      │     │
│  └────────────────────────────┘     │
└────────┬────────────────────────────┘
         ▼
┌─────────────────────────────────────┐
│  向量数据库 (Chroma / FAISS)        │
│  索引：文档片段 + 向量              │
└────────┬────────────────────────────┘
         ▼
┌─────────────────────────────────────┐
│  LLM 生成 (GPT-4)                   │
│  输入：Prompt + 检索到的文档片段    │
│  输出：答案 + 引用                  │
└─────────────────────────────────────┘

python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

pip install langchain langchain-community chromadb openai tiktoken pypdf python-docx

from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader, TextLoader

# 加载 PDF
pdf_loader = PyPDFLoader("docs/technical_manual.pdf")
pdf_docs = pdf_loader.load()

# 加载 Word
docx_loader = Docx2txtLoader("docs/design_spec.docx")
docx_docs = docx_loader.load()

# 合并
documents = pdf_docs + docx_docs

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,          # 每块最大字符数
    chunk_overlap=50,        # 重叠长度，保持语义连贯
    separators=["\n\n", "\n", "。", "！", "？", " ", ""]
)
chunks = splitter.split_documents(documents)
print(f"共生成 {len(chunks)} 个片段")

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-ada-002",
    openai_api_key="your-api-key"
)

from langchain_community.vectorstores import Chroma

vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"   # 持久化路径
)
# 持久化保存
vectorstore.persist()

vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

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

from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA

llm = ChatOpenAI(
    model="gpt-4-turbo",
    temperature=0.2,
    openai_api_key="your-api-key"
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",   # 将全部检索片段一次性放入上下文
    retriever=retriever,
    return_source_documents=True   # 返回引用
)

query = "如何配置服务的超时参数？"
result = qa_chain.invoke({"query": query})
print("答案:", result["result"])
print("来源:", result["source_documents"])

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    output_key="answer"
)

conv_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=retriever,
    memory=memory,
    return_source_documents=True
)

# 第一轮
response = conv_chain.invoke({"question": "什么是RAG？"})
print(response["answer"])

# 第二轮（带上下文）
response2 = conv_chain.invoke({"question": "它和微调比有什么优势？"})
print(response2["answer"])

from langchain.prompts import PromptTemplate

template = """
你是一个专业的技术顾问。请根据以下“上下文”片段回答用户的问题。
如果无法从上下文中找到答案，请明确说“根据现有资料无法回答”。
请在回答末尾列出所引用的文档来源（文件名和页码）。

上下文：
{context}

问题：{question}
回答：
"""

prompt = PromptTemplate(template=template, input_variables=["context", "question"])

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True,
    chain_type_kwargs={"prompt": prompt}
)

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainFilter

compressor = LLMChainFilter.from_llm(llm)   # 让LLM判断哪些片段相关
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)
# 替换原有检索器
qa_chain.retriever = compression_retriever

from langchain_community.vectorstores import FAISS

vectorstore = FAISS.from_documents(documents, embeddings)
vectorstore.save_local("faiss_index")
# 加载
vectorstore = FAISS.load_local("faiss_index", embeddings, allow_dangerous_deserialization=True)

import hashlib
import redis

cache = redis.Redis(host='localhost', port=6379, db=0)

def cached_query(question):
    key = hashlib.md5(question.encode()).hexdigest()
    cached = cache.get(key)
    if cached:
        return cached.decode()
    answer = qa_chain.invoke({"query": question})["result"]
    cache.setex(key, 3600, answer)  # 缓存1小时
    return answer

pip install ragas

from ragas.metrics import faithfulness, answer_relevancy
from ragas import evaluate

# 准备测试数据（包含 question, answer, contexts）
# 参考 ragas 官方文档

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Query(BaseModel):
    question: str
    session_id: str = None

@app.post("/ask")
async def ask(query: Query):
    # 根据 session_id 管理不同 memory（可使用字典或 Redis）
    result = qa_chain.invoke({"question": query.question})
    return {"answer": result["result"], "sources": [doc.metadata for doc in result["source_documents"]]}