Python 本地 AI 问答系统搭建：环境配置与 RAG 实践

第三步：验证安装是否成功

import torch
print(f"PyTorch 版本：{torch.__version__}")
print(f"CUDA 是否可用：{torch.cuda.is_available()}")
print(f"GPU 数量：{torch.cuda.device_count()}")
if torch.cuda.is_available():
    print(f"当前 GPU: {torch.cuda.get_device_name(0)}")
    print(f"显存总量：{torch.cuda.get_device_properties(0).total_memory / 1024**3:.1f} GB")

3.3 依赖管理：用 requirements.txt 锁定版本

# 生成当前环境的依赖快照
pip freeze > requirements.txt

# 在新环境中还原
pip install -r requirements.txt

推荐的 requirements.txt 示例：

torch==2.2.0
transformers==4.38.0
langchain==0.1.9
langchain-community==0.0.24
faiss-cpu==1.7.4
sentence-transformers==2.5.1
ollama==0.1.7
gradio==4.19.2

四、模型下载：别让网络毁了你的心情

4.1 使用 Ollama 管理本地模型（强烈推荐）

Ollama 是目前最省心的本地模型管理工具，一行命令搞定下载和运行：

# 安装 Ollama（macOS/Linux）
curl -fsSL https://ollama.com/install.sh | sh

# 下载并运行模型
ollama pull llama3.2        # Meta Llama 3.2 (3B)
ollama pull qwen2.5:7b      # 阿里通义千问 2.5 (7B)
ollama pull deepseek-r1:7b  # DeepSeek R1 (7B)

# 验证模型列表
ollama list

4.2 用 Python 调用 Ollama

import ollama

def ask_local_llm(question: str, model: str = "qwen2.5:7b") -> str:
    """
    调用本地 Ollama 模型进行问答
    Args:
        question: 用户问题
        model: 模型名称
    Returns:
        模型回答
    """
    response = ollama.chat(
        model=model,
        messages=[
            {"role": "system", "content": "你是一个专业的 AI 助手，请用中文简洁准确地回答问题。"},
            {"role": "user", "content": question}
        ]
    )
    return response["message"]["content"]

# 测试
if __name__ == "__main__":
    answer = ask_local_llm("Python 中的 GIL 是什么？")
    print(answer)

五、搭建 RAG 问答系统

RAG（Retrieval-Augmented Generation）是让 AI 能回答你私有文档问题的核心技术。

5.1 RAG 完整流程

离线阶段（一次性）：

1. 加载文档（PDF/TXT/MD）
2. 文本分块（Chunk Splitting）
3. 向量化（Embedding）
4. 存入向量库（FAISS/Chroma）

在线阶段（每次查询）：

1. 用户提问
2. 问题向量化
3. 相似度检索
4. 召回 Top-K 文档
5. 构建增强 Prompt
6. LLM 生成回答
7. 返回答案

5.2 完整代码实现

"""
本地 RAG 问答系统
依赖：pip install langchain langchain-community faiss-cpu sentence-transformers ollama
"""
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.document_loaders import TextLoader, DirectoryLoader
from langchain.chains import RetrievalQA
from langchain_community.llms import Ollama
import os

class LocalRAGSystem:
    """本地 RAG 问答系统"""
    def __init__(
        self,
        docs_dir: str = "./docs",
        model_name: str = "qwen2.5:7b",
        embedding_model: str = "BAAI/bge-small-zh-v1.5",
        chunk_size: int = 500,
        chunk_overlap: int = 50,
    ):
        self.docs_dir = docs_dir
        self.model_name = model_name
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        print("🔧 初始化 Embedding 模型...")
        # 使用本地 Embedding 模型，避免调用外部 API
        self.embeddings = HuggingFaceEmbeddings(
            model_name=embedding_model,
            model_kwargs={"device": "cpu"},  # 改为 "cuda" 可用 GPU 加速
            encode_kwargs={"normalize_embeddings": True},
        )
        self.vectorstore = None
        self.qa_chain = None

    def load_and_index(self):
        """加载文档并建立向量索引"""
        print(f"📂 加载文档目录：{self.docs_dir}")
        # 支持多种文档格式
        loader = DirectoryLoader(
            self.docs_dir,
            glob="**/*.txt",
            loader_cls=TextLoader,
            loader_kwargs={"encoding": "utf-8"},
        )
        documents = loader.load()
        print(f"✅ 加载了 {len(documents)} 个文档")

        # 文本分块
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap,
            separators=["\n\n", "\n", "。", "！", "？", " ", ""],
        )
        chunks = splitter.split_documents(documents)
        print(f"✅ 分割为 {len(chunks)} 个文本块")

        # 建立向量索引
        print("🔍 建立向量索引（首次较慢，请耐心等待）...")
        self.vectorstore = FAISS.from_documents(chunks, self.embeddings)
        print("✅ 向量索引建立完成")

        # 保存索引到本地（下次直接加载，无需重建）
        self.vectorstore.save_local("./faiss_index")
        print("💾 索引已保存到 ./faiss_index")

    def load_existing_index(self):
        """加载已有的向量索引"""
        if os.path.exists("./faiss_index"):
            print("📦 加载已有向量索引...")
            self.vectorstore = FAISS.load_local(
                "./faiss_index",
                self.embeddings,
                allow_dangerous_deserialization=True,
            )
            print("✅ 索引加载完成")
        else:
            print("⚠️ 未找到已有索引，请先调用 load_and_index()")

    def build_qa_chain(self):
        """构建问答链"""
        if self.vectorstore is None:
            raise ValueError("请先调用 load_and_index() 或 load_existing_index()")
        print(f"🤖 连接本地 LLM: {self.model_name}")
        llm = Ollama(model=self.model_name, temperature=0.1)  # 降低随机性，让回答更稳定
        retriever = self.vectorstore.as_retriever(
            search_type="similarity",
            search_kwargs={"k": 3},  # 召回最相关的 3 个文档块
        )
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=llm,
            chain_type="stuff",
            retriever=retriever,
            return_source_documents=True,
        )
        print("✅ 问答系统就绪！")

    def ask(self, question: str) -> dict:
        """
        提问并获取答案
        Returns:
            dict: {"answer": str, "sources": list}
        """
        if self.qa_chain is None:
            raise ValueError("请先调用 build_qa_chain()")
        result = self.qa_chain.invoke({"query": question})
        return {
            "answer": result["result"],
            "sources": [
                doc.metadata.get("source", "未知来源")
                for doc in result["source_documents"]
            ],
        }

# ============ 使用示例 ============
if __name__ == "__main__":
    # 初始化系统
    rag = LocalRAGSystem(
        docs_dir="./my_docs",
        model_name="qwen2.5:7b",
    )
    # 首次使用：加载文档并建立索引
    rag.load_and_index()
    # 后续使用：直接加载已有索引（更快）
    # rag.load_existing_index()
    # 构建问答链
    rag.build_qa_chain()
    # 开始问答
    while True:
        question = input("\n❓ 请输入问题（输入 q 退出）: ").strip()
        if question.lower() == "q":
            break
        result = rag.ask(question)
        print(f"\n💡 回答:\n{result['answer']}")
        print(f"\n📎 参考来源：{', '.join(result['sources'])}")

六、内存/显存管理：别让 OOM 毁了你

6.1 显存需求参考

不同规模模型的显存需求（FP16 精度）：

• 1B 模型 (~2GB)
• 3B 模型 (~6GB)
• 7B 模型 (~14GB)
• 13B 模型 (~26GB)
• 量化 4bit 7B (~4GB)

6.2 显存不够？用量化压缩模型

from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
import torch

# 4-bit 量化配置（显存减少约 75%）
quantization_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_compute_dtype=torch.float16,
    bnb_4bit_use_double_quant=True,
    bnb_4bit_quant_type="nf4",
)
model_id = "Qwen/Qwen2.5-7B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_id)
model = AutoModelForCausalLM.from_pretrained(
    model_id,
    quantization_config=quantization_config,
    device_map="auto",  # 自动分配到 GPU/CPU
)
print(f"模型加载完成，占用显存：{torch.cuda.memory_allocated()/1024**3:.2f} GB")

6.3 流式输出，避免等待超时

import ollama

def stream_answer(question: str, model: str = "qwen2.5:7b"):
    """流式输出，边生成边显示"""
    print("💬 ", end="", flush=True)
    for chunk in ollama.chat(
        model=model,
        messages=[{"role": "user", "content": question}],
        stream=True,  # 开启流式输出
    ):
        content = chunk["message"]["content"]
        print(content, end="", flush=True)
    print()

stream_answer("用一句话解释什么是 Transformer 架构")

七、加一个 Web 界面（可选）

用 Gradio 5 分钟搭一个好看的 Web 界面：

import gradio as gr
from local_rag import LocalRAGSystem  # 引用上面的代码

# 初始化 RAG 系统
rag = LocalRAGSystem()
rag.load_existing_index()
rag.build_qa_chain()

def chat(message: str, history: list) -> str:
    """Gradio 聊天回调函数"""
    if not message.strip():
        return "请输入问题"
    result = rag.ask(message)
    answer = result["answer"]
    sources = result["sources"]
    if sources:
        answer += f"\n\n---\n📎 **参考来源**: {', '.join(set(sources))}"
    return answer

# 创建 Gradio 界面
demo = gr.ChatInterface(
    fn=chat,
    title="🤖 本地 AI 问答系统",
    description="基于本地 LLM + RAG 的私有知识库问答",
    examples=["这个系统是如何工作的？", "请总结一下主要内容"],
    theme=gr.themes.Soft(),
)

if __name__ == "__main__":
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False,  # 改为 True 可生成公网链接
    )

运行后访问 http://localhost:7860 即可使用。

八、常见报错速查表

九、国内加速技巧

# 设置 HuggingFace 镜像（国内访问加速）
export HF_ENDPOINT=https://hf-mirror.com

# pip 使用清华镜像
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple transformers

# conda 使用清华镜像
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main

# 在代码中指定镜像
import os
os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"
from transformers import AutoTokenizer

# 之后的下载会自动走镜像
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B-Instruct")

十、总结：搭建清单

1. ✅ 创建虚拟环境（venv / conda）
2. ✅ 查 CUDA 版本（nvidia-smi）
3. ✅ 安装对应 PyTorch（官网生成命令）
4. ✅ 安装 Ollama（管理本地模型）
5. ✅ 下载模型（ollama pull）
6. ✅ 准备文档（放入 docs 目录）
7. ✅ 建立向量索引（FAISS / Chroma）
8. ✅ 构建问答链（LangChain）
9. ✅ 直接调用 LLM（ollama.chat）
10. ✅ 可选：Gradio Web UI

🎉 系统就绪！

参考资源

• Ollama 官网 — 本地模型管理
• LangChain 文档 — RAG 框架
• HuggingFace 镜像站 — 国内模型下载
• PyTorch 安装向导 — 版本对齐工具
• FAISS 文档 — 向量检索库