作者：吴业亮
博客：wuyeliang.blog.csdn.net

系统架构概述

本方案将构建一个完全本地化、数据隐私安全的检索增强生成（RAG）系统。系统通过LangChain框架协调各个组件，利用您本地的A40 48G显卡资源，实现高效的文档处理和智能问答功能。

技术栈选型

组件	选型	说明
操作系统	Ubuntu 22.04 LTS	稳定的基础环境
环境管理	Conda	Python环境隔离与管理
大模型推理	SGLang	高性能推理运行时，优化LLM执行效率
开发框架	LangChain	LLM应用开发框架，提供RAG核心组件
向量数据库	Chroma	轻量级、嵌入式向量数据库
Embedding模型	BGE系列或Sentence-Transformers	生成文本向量表示
Web框架	Gradio/FastAPI	提供用户友好的交互界面

核心数据流

1. 文档处理流程：本地文档 → 文本加载与清洗 → 智能分块 → 向量化 → 向量存储
2. 问答流程：用户提问 → 查询向量化 → 向量相似性检索 → 提示词构建 → LLM生成 → 结果返回

环境准备与配置

1. Ubuntu 22.04基础环境配置

首先确保系统已更新并安装必要的依赖包：

sudo apt update && sudo apt upgrade -y
sudo apt install -y wget git build-essential libgl1-mesa-glx libegl1-mesa libxrandr2 libxss1 libxcursor1 libxcomposite1 libasound2 libxi6 libxtst6

2. Conda环境安装与配置

按照以下步骤安装Anaconda并配置环境：

# 下载Anaconda（选择最新Python 3.10+版本）
wget https://repo.anaconda.com/archive/Anaconda3-2024.10-1-Linux-x86_64.sh

# 安装Anaconda
bash Anaconda3-2024.10-1-Linux-x86_64.sh
# 安装过程中一路回车并确认许可协议，最后输入yes完成安装

# 激活Conda配置
source ~/.bashrc

# 验证安装
conda --version

若安装后无法识别conda命令，需手动添加环境变量到~/.bashrc文件末尾：

export PATH="~/anaconda3/bin:$PATH"
export PATH="/home/$USER/anaconda3/bin:$PATH"  # 根据实际安装路径调整
source ~/.bashrc

配置国内镜像源加速下载：

conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main

3. 创建专用Python环境

# 创建Python 3.10环境
conda create -n rag_system python=3.10 -y

# 激活环境
conda activate rag_system

核心组件安装与配置

1. 安装LangChain及相关依赖

# 核心框架
pip install langchain langchain-community langchain-core

# 文档处理库（支持PDF、Word、Markdown等格式）
pip install pypdf python-docx markdown unstructured

# 文本分割与处理
pip install tiktoken sentence-transformers

# 向量数据库
pip install chromadb

2. 安装SGLang并配置本地模型

SGLang是一个专门优化大语言模型推理的高性能运行时，特别适合复杂的提示词结构和RAG场景。

# 安装SGLang
pip install sglang

# 安装PyTorch（根据CUDA版本选择）
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121

# 安装vLLM后端（SGLang常用后端）
pip install vllm

3. 下载并配置本地大模型

利用A40 48G显卡的强大显存，您可以部署7B-13B参数规模的量化模型：

# 创建模型存储目录
mkdir -p models/local_llm

# 示例：下载一个适合A40显卡的模型（如Qwen7B-Chat或Llama2-7B-Chat）
# 模型文件应放置到models目录下

模型选型建议：

• Qwen2.5-7B-Chat：中文支持优秀，性能均衡
• Llama2-7B-Chat：英文任务表现出色，社区支持好
• DeepSeek-R1-7B：专门优化推理能力

4. 验证CUDA环境

A40显卡需要正确的CUDA驱动支持：

# 检查显卡驱动和可支持的最高CUDA版本
nvidia-smi

# 如果未安装驱动，先安装推荐驱动
sudo apt install nvidia-driver-550

# 验证PyTorch能否识别GPU
python -c "import torch; print(f'CUDA可用: {torch.cuda.is_available()}'); print(f'设备数量: {torch.cuda.device_count()}'); print(f'当前设备: {torch.cuda.current_device()}')"

知识库系统实现

1. 文档处理模块

创建document_processor.py：

import os
from langchain.document_loaders import (
    PyPDFLoader, 
    Docx2txtLoader, 
    TextLoader,
    UnstructuredMarkdownLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter

class DocumentProcessor:
    def __init__(self, chunk_size=500, chunk_overlap=50):
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", "。", "！", "？", "．", ".", " ", ""]
        )
    
    def load_documents(self, directory_path):
        """加载目录下的所有支持格式文档"""
        documents = []
        
        for filename in os.listdir(directory_path):
            file_path = os.path.join(directory_path, filename)
            ext = os.path.splitext(filename)[1].lower()
            
            try:
                if ext == ".pdf":
                    loader = PyPDFLoader(file_path)
                elif ext == ".docx":
                    loader = Docx2txtLoader(file_path)
                elif ext == ".md":
                    loader = UnstructuredMarkdownLoader(file_path)
                elif ext == ".txt":
                    loader = TextLoader(file_path, encoding='utf-8')
                else:
                    print(f"跳过不支持的文件格式: {filename}")
                    continue
                
                loaded_docs = loader.load()
                documents.extend(loaded_docs)
                print(f"成功加载: {filename}, 文档数: {len(loaded_docs)}")
                
            except Exception as e:
                print(f"加载文档 {filename} 时出错: {str(e)}")
        
        return documents
    
    def split_documents(self, documents):
        """分割文档为适当大小的块"""
        return self.text_splitter.split_documents(documents)

2. 向量存储模块

创建vector_store.py：

from langchain.vectorstores import Chroma
from langchain.embeddings import HuggingFaceEmbeddings

class VectorStoreManager:
    def __init__(self, persist_directory="./chroma_db"):
        self.persist_directory = persist_directory
        # 使用本地Embedding模型
        self.embeddings = HuggingFaceEmbeddings(
            model_name="BAAI/bge-small-zh-v1.5",  # 中文优化模型
            model_kwargs={'device': 'cuda'},
            encode_kwargs={'normalize_embeddings': True}
        )
    
    def create_vector_store(self, documents):
        """创建向量存储"""
        vector_store = Chroma.from_documents(
            documents=documents,
            embedding=self.embeddings,
            persist_directory=self.persist_directory
        )
        vector_store.persist()
        return vector_store
    
    def load_existing_store(self):
        """加载已存在的向量存储"""
        return Chroma(
            persist_directory=self.persist_directory,
            embedding_function=self.embeddings
        )

3. SGLang模型集成

创建sglang_llm.py：

import os
from langchain.llms.base import LLM
from typing import Optional, List, Any, Mapping
from sglang import RuntimeEndpoint

class SGLangLLM(LLM):
    """自定义SGLang本地模型集成"""
    
    def __init__(self, model_path: str, host: str = "http://localhost", port: int = 30000):
        self.endpoint = RuntimeEndpoint(f"{host}:{port}")
        self.model_path = model_path
        super().__init__()
    
    def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str:
        # 使用SGLang执行模型推理
        response = self.endpoint.generate(
            prompt,
            temperature=0.1,
            max_tokens=2048,
            stop=stop
        )
        return response.text.strip()
    
    @property
    def _llm_type(self) -> str:
        return "sglang-local"

4. RAG链实现

创建rag_system.py：

from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from sglang_llm import SGLangLLM
from vector_store import VectorStoreManager

class RAGSystem:
    def __init__(self, model_path, vector_store_dir="./chroma_db"):
        self.llm = SGLangLLM(model_path=model_path)
        self.vector_store_mgr = VectorStoreManager(vector_store_dir)
        self.vector_store = self.vector_store_mgr.load_existing_store()
        self.qa_chain = self._setup_rag_chain()
    
    def _setup_rag_chain(self):
        """设置RAG链"""
        
        prompt_template = """你是一个专业的本地知识库助手。请严格根据以下上下文信息回答问题。如果上下文没有足够信息，请如实告知，不要编造信息。

上下文：
{context}

问题：{question}

请根据上下文提供准确、简洁的回答："""
        
        PROMPT = PromptTemplate(
            template=prompt_template, 
            input_variables=["context", "question"]
        )
        
        return RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.vector_store.as_retriever(
                search_type="similarity",
                search_kwargs={"k": 4, "score_threshold": 0.6}
            ),
            chain_type_kwargs={"prompt": PROMPT},
            return_source_documents=True
        )
    
    def query(self, question: str) -> dict:
        """执行查询"""
        return self.qa_chain({"query": question})

Web系统开发

使用Gradio创建友好的用户界面，创建app.py：

import gradio as gr
from rag_system import RAGSystem
import os

# 初始化RAG系统
rag_system = RAGSystem(
    model_path="./models/your_local_model",  # 替换为实际模型路径
    vector_store_dir="./chroma_db"
)

def chat_interface(question, history):
    """处理用户问答"""
    if not question.strip():
        return "请输入问题"
    
    try:
        result = rag_system.query(question)
        answer = result["result"]
        
        # 添加参考来源显示
        sources = "\n\n参考来源："
        for i, doc in enumerate(result["source_documents"]):
            source_info = f"文档{i+1}: {os.path.basename(doc.metadata.get('source', '未知'))}"
            sources += f"\n- {source_info}"
        
        return answer + sources
    
    except Exception as e:
        return f"处理问题时出错：{str(e)}"

def build_web_interface():
    """构建Gradio Web界面"""
    
    with gr.Blocks(
        title="本地知识库AI助手",
        theme=gr.themes.Soft(),
        css="""
        .message { font-size: 16px; }
        .title { text-align: center; }
        """
    ) as demo:
        
        gr.Markdown(
            """
            # <div class="title">本地知识库AI助手</div>
            > 基于LangChain + SGLang构建，数据完全本地处理
            """
        )
        
        with gr.Row():
            with gr.Column(scale=4):
                chatbot = gr.Chatbot(
                    label="问答对话",
                    height=500,
                    show_copy_button=True
                )
                msg = gr.Textbox(
                    label="输入您的问题",
                    placeholder="请输入关于您知识库的问题...",
                    lines=2
                )
                
            with gr.Column(scale=1):
                gr.Markdown("### 系统信息")
                gr.Markdown(f"**模型状态**: 已加载")
                gr.Markdown(f"**知识库**: 已就绪")
                gr.Markdown(f"**硬件**: A40 48GB GPU")
                
                clear_btn = gr.Button("清空对话", variant="secondary")
        
        # 事件处理
        def respond(message, chat_history):
            response = chat_interface(message, chat_history)
            chat_history.append((message, response))
            return "", chat_history
        
        msg.submit(respond, [msg, chatbot], [msg, chatbot])
        clear_btn.click(lambda: None, None, chatbot, queue=False)
    
    return demo

if __name__ == "__main__":
    # 启动Web服务
    demo = build_web_interface()
    demo.launch(
        server_name="0.0.0.0", 
        server_port=7860,
        share=False
    )

系统部署与运行

1. 项目结构组织

local_rag_system/
├── app.py                 # 主应用入口
├── document_processor.py  # 文档处理模块
├── vector_store.py        # 向量存储管理
├── sglang_llm.py          # SGLang模型集成
├── rag_system.py          # RAG核心逻辑
├── requirements.txt       # 依赖列表
├── knowledge_base/        # 知识库文档目录
├── models/               # 本地模型存储
└── chroma_db/           # 向量数据库持久化目录

2. 安装依赖文件

创建requirements.txt：

langchain>=0.1.0
langchain-community>=0.0.1
langchain-core>=0.1.0
chromadb>=0.4.0
sentence-transformers>=2.2.0
gradio>=4.0.0
sglang>=0.1.0
vllm>=0.3.0
torch>=2.0.0
accelerate>=0.20.0
pypdf>=3.0.0
python-docx>=1.1.0
unstructured>=0.10.0

3. 启动脚本

创建启动脚本start_server.sh：

#!/bin/bash

# 激活Conda环境
conda activate rag_system

# 启动SGLang服务端（假设已下载模型）
python -m sglang.launch_server --model-path ./models/your_model --port 30000 --host 0.0.0.0 &

# 等待服务端启动
sleep 30

# 启动Gradio Web界面
python app.py

赋予执行权限并运行：

chmod +x start_server.sh
./start_server.sh

性能优化建议

1. GPU内存优化：

   • 使用模型量化（4-bit/8-bit）减少显存占用
   • 调整批处理大小平衡吞吐量和延迟
   • 启用FlashAttention优化注意力计算

2. 检索优化：

   • 尝试不同分块策略（技术文档300-500字，长报告800-1200字）
   • 使用混合检索（语义+关键词）提升召回率
   • 实现重排序（re-ranking）改善结果相关性

3. 生成优化：

   • 设计约束性提示词减少模型幻觉
   • 设置合适的temperature参数（0.1-0.3提高准确性）
   • 使用流式输出改善用户体验

故障排除

常见问题解决：

• CUDA内存不足：减小批处理大小，使用模型量化
• 检索效果差：调整分块大小，尝试不同Embedding模型
• 模型加载失败：检查模型文件完整性，确认格式兼容性
• Web界面无法访问：检查防火墙设置，确认端口7860可访问