python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python搭建AI问答系统

使用Python搭建本地 AI问答系统

作者:我不是呆头

这篇文章主要为大家详细介绍了如何使用Python从零搭建一个本地 AI问答系统,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下

前言

想在本地跑一个 AI 问答系统?听起来很酷,但现实往往是这样的:

这些问题,90% 的新手都踩过。本文将带你从零搭建一个本地 AI 问答系统,并系统性地帮你绕开那些"经典陷阱"。

一、整体架构概览

在动手之前,先看清楚我们要搭建的是什么:

整个系统分为三层:

二、新手踩坑分布图

根据社区反馈,新手遇到的问题主要集中在以下几类:

接下来,我们按照这个优先级,逐一击破。

三、环境搭建:最容易翻车的第一步

3.1 用虚拟环境隔离,别污染全局

新手常见错误:

pip install torch transformers langchain  # 直接装到全局

正确做法:用 venvconda 隔离环境

# 方式一:使用 venv(推荐,Python 内置)
python -m venv ai-qa-env
source ai-qa-env/bin/activate        # Linux/macOS
ai-qa-env\Scripts\activate           # Windows
# 方式二:使用 conda
conda create -n ai-qa python=3.11
conda activate ai-qa

为什么要隔离? 不同项目依赖不同版本的库,全局安装会导致版本冲突,出了问题极难排查。

3.2 PyTorch 安装:版本对齐是关键

这是 最高频的踩坑点。PyTorch 的安装命令取决于你的 CUDA 版本,不能无脑 pip install torch

第一步:查看你的 CUDA 版本

nvidia-smi  # 查看 GPU 驱动支持的最高 CUDA 版本
nvcc --version  # 查看已安装的 CUDA Toolkit 版本

第二步:去官网生成对应命令

# CUDA 12.1 对应的安装命令示例
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121
# 没有 GPU,只用 CPU
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu

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

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 完整流程

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 显存需求参考

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 即可使用。

八、常见报错速查表

报错信息原因解决方案
CUDA out of memory显存不足使用量化模型或减小 batch_size
ModuleNotFoundError: No module named 'torch'虚拟环境未激活激活对应的 venv/conda 环境
RuntimeError: CUDA error: no kernel image is availablePyTorch 与 CUDA 版本不匹配重新安装对应 CUDA 版本的 PyTorch
ConnectionRefusedError: [Errno 111]Ollama 服务未启动运行 ollama serve
OSError: [Errno 28] No space left on device磁盘空间不足清理磁盘或更换存储路径
ValueError: Tokenizer class ... not foundtransformers 版本过低pip install -U transformers
huggingface_hub.utils._errors.EntryNotFoundError模型名称错误或网络问题检查模型 ID 或使用镜像源

九、国内加速技巧

# 设置 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")

十、总结:搭建清单

到此这篇关于使用Python搭建本地 AI问答系统的文章就介绍到这了,更多相关Python搭建AI问答系统内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
阅读全文