基于 LangChain + 通义千问 + Chroma 的本地知识库 RAG 问答系统：从零到生产的完整实战

[私有文档] → 加载 → 切分 → Embedding → Chroma向量库
                                        ↑
[用户提问] → 语义路由 → 向量检索 → 拼接Prompt → 通义千问 → 答案
                            ↓
                    查询改写(低召回时)
                            ↓
                    LLM兜底(无命中时)

# Python 3.9+，建议虚拟环境pip install langchain langchain-chroma langchain-community langchain-core
pip install pypdf docx2txt python-dotenv
pip install sentence-transformers

DASHSCOPE_API_KEY=你的通义千问APIKey

import osfrom langchain_community.document_loaders import (
    PyPDFLoader, Docx2txtLoader, TextLoader
)from langchain_text_splitters import RecursiveCharacterTextSplitterfrom langchain_core.documents import Documentdef load_documents(path: str) -> list[Document]:    """加载单个文件或整个目录"""
    docs = []    if os.path.isdir(path):        for root, _, files in os.walk(path):            for f in files:
                docs.extend(_load_single(os.path.join(root, f)))    else:
        docs.extend(_load_single(path))    return docsdef _load_single(filepath: str) -> list[Document]:
    ext = os.path.splitext(filepath)[1].lower()    if ext == ".pdf":
        loader = PyPDFLoader(filepath)    elif ext == ".docx":
        loader = Docx2txtLoader(filepath)    elif ext in (".txt", ".md"):
        loader = TextLoader(filepath, encoding="utf-8")    else:        return []    return loader.load()def split_documents(docs: list[Document]) -> list[Document]:    """递归字符切分，兼顾中英文"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,        # 每块约500字
        chunk_overlap=50,      # 重叠50字，保证语义连贯
        separators=["\n\n", "\n", "。", "！", "？", "，", ""],
        keep_separator=True,
    )
    chunks = splitter.split_documents(docs)    # 注入来源文件名，便于后续溯源
    for chunk in chunks:
        chunk.metadata["filename"] = os.path.basename(
            chunk.metadata.get("source", "未知文件")
        )    return chunks

from langchain_chroma import Chromafrom langchain_community.embeddings import HuggingFaceEmbeddings
EMBED_MODEL = "BAAI/bge-large-zh-v1.5"DB_DIR = "./chroma_db"def get_embeddings():    return HuggingFaceEmbeddings(
        model_name=EMBED_MODEL,
        model_kwargs={"device": "cuda"},  # 无GPU改"cpu"
        encode_kwargs={"normalize_embeddings": True},
    )def build_vector_store(chunks, persist_dir=DB_DIR):    """首次构建并持久化"""
    embeddings = get_embeddings()
    vectordb = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_dir,
    )
    vectordb.persist()    print(f"✅ 向量库构建完成，共 {len(chunks)} 个片段")    return vectordbdef load_vector_store(persist_dir=DB_DIR):    """后续直接加载，避免重复计算"""
    embeddings = get_embeddings()    return Chroma(persist_directory=persist_dir,
                  embedding_function=embeddings)

import osfrom langchain_core.prompts import PromptTemplatefrom langchain_core.runnables import RunnablePassthroughfrom langchain_core.output_parsers import StrOutputParserfrom langchain_community.llms import Tongyi
os.environ["DASHSCOPE_API_KEY"] = os.getenv("DASHSCOPE_API_KEY")# ---------- 1. 检索器 ----------retriever = vectordb.as_retriever(search_kwargs={"k": 5})# ---------- 2. 提示词模板 ----------template = """
你是一个专业的问答助手，请根据下面的参考资料回答问题。
如果参考资料中没有答案，请直接说"没有找到相关信息"。
参考资料：
{context}
问题：{question}
请回答，并在最后列出你参考了哪些文件。
回答格式要求：
【回答】
xxx
【参考文件】
xxx
"""prompt = PromptTemplate.from_template(template)# ---------- 3. 格式化检索结果（带文件名溯源）----------def format_docs(docs):
    formatted = []    for doc in docs:
        content = doc.page_content
        filename = doc.metadata.get("filename", "未知文件")
        formatted.append(f"【内容】：{content}\n【来源文件】：{filename}")    return "\n\n------------------------\n\n".join(formatted)# ---------- 4. 接入通义千问 ----------llm = Tongyi(model_name="qwen-turbo", temperature=0.1, max_tokens=1024)# ---------- 5. LCEL 组装 RAG 链 ----------rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)# ---------- 6. 调用 ----------if __name__ == "__main__":
    question = "咱们公司的年假政策是怎么规定的？"
    answer = rag_chain.invoke(question)    print(answer)

【回答】
根据《员工手册》规定，入职满1年的正式员工享受5天年假，
满3年享受10天，满5年享受15天。年假需提前一周向部门负责人申请。
【参考文件】
员工手册_2025版.pdf

from langchain_core.prompts import PromptTemplatefrom langchain_community.llms import Tongyi
llm = Tongyi(model_name="qwen-turbo", temperature=0.1)# ---------- 查询改写：低召回时自动改写问题 ----------rewrite_template = """
用户原始问题：{question}
检索到的资料不足以回答。请将问题改写为更适合向量检索的简洁查询句，
只输出改写后的问题，不要解释。
"""rewrite_prompt = PromptTemplate.from_template(rewrite_template)def rewrite_query(original_question: str) -> str:    """检索命中数为0时，调用LLM改写查询"""
    rewritten = (rewrite_prompt | llm | StrOutputParser()).invoke(
        {"question": original_question}
    )    print(f"🔄 查询改写：{original_question} → {rewritten}")    return rewritten# ---------- LLM 兜底：多次检索无果时直接回答 ----------def fallback_answer(question: str) -> str:    """检索无果时的通用常识兜底，避免服务中断"""
    fallback_template = "你是一个智能助手，请尽你所能回答用户问题：{question}"
    prompt = PromptTemplate.from_template(fallback_template)    return (prompt | llm | StrOutputParser()).invoke({"question": question})

from rag_chain import rag_chainfrom enhancements import rewrite_query, fallback_answerfrom vector_store import vectordbdef ask(question: str, chat_history: list = None) -> str:    # 1. 首次检索
    docs = vectordb.as_retriever(search_kwargs={"k": 5}).invoke(question)    
    # 2. 低召回判断（阈值可根据业务调整）
    if len(docs) < 2:
        rewritten = rewrite_query(question)
        docs = vectordb.as_retriever(search_kwargs={"k": 5}).invoke(rewritten)    
    # 3. 仍无命中 → 兜底
    if len(docs) == 0:        print("⚠️ 知识库未检索到相关内容，启用LLM通用回答")        return fallback_answer(question)    
    # 4. 正常RAG链路
    return rag_chain.invoke(question)# 流式输出版本def ask_stream(question: str):    for chunk in rag_chain.stream(question):        print(chunk, end="", flush=True)if __name__ == "__main__":    print("🤖 知识库问答系统已启动，输入 exit 退出")    while True:
        q = input("\n用户：").strip()        if q.lower() == "exit":            break
        print("助手：", end="")
        ask_stream(q)

rag-system/
├── main.py              # 入口
├── config.py            # 全局配置
├── loader.py            # 文档加载与切分
├── vector_store.py      # 向量库
├── rag_chain.py         # RAG核心链
├── enhancements.py      # 查询改写、兜底
├── requirements.txt
└── chroma_db/           # 持久化向量库