本地部署RAG智能体系统：基于Qwen与ChromaDB的实践-AI智能范式网

本地部署RAG智能体系统：基于Qwen与ChromaDB的实践

RAGQwenChromaDB

|难度：普通|阅读约 22 分钟

1. 项目概述

这个项目展示了一个完全本地部署的RAG（检索增强生成）智能体系统，基于Qwen系列模型和ChromaDB向量数据库构建。与常见的LangChain方案不同，这个实现从底层开始构建，不依赖任何高级框架，让开发者能够完全掌控系统的每个组件。

1.1 核心功能解析

系统提供了几项关键功能设计：

深度思考模式：可以显示模型的完整推理过程，这在调试复杂问题时特别有用
知识库自由开关：根据问题复杂度决定是否检索知识库，节省简单问题的响应时间
两阶段检索策略：结合Embedding的快速召回和Reranker的精确排序，平衡速度与精度
参数精细调控：支持Temperature、Top-p等关键生成参数的实时调整

提示：这种设计特别适合需要高度定制化的场景，比如企业内部知识管理系统或特定领域的专业问答系统。

需要模型API调用？免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。

2. 系统架构设计

2.1 整体架构

系统采用模块化设计，主要包含以下组件：

Qwen3-4B：作为核心语言模型处理生成任务
Qwen3-Embedding：负责文本向量化
Qwen3-Reranker：提供检索结果重排序
ChromaDB：轻量级向量数据库存储知识库
自实现文本分割器：替代LangChain的文本处理功能

2.2 数据流设计

文档处理流程：

code复制原始文档 → 文本分割 → Embedding编码 → 向量存储

查询处理流程：

code复制用户问题 → Embedding编码 → 向量检索 → Reranker重排序 → LLM生成

这种两阶段设计是系统的核心创新点，它解决了单纯向量检索精度不足的问题，同时避免了全程使用Reranker带来的性能开销。

3. 关键技术实现

3.1 Embedding模型封装

Qwen3-Embedding的实现有几个关键点需要注意：

python复制class EmbeddingModel:
    def __init__(self, model_name="Qwen/Qwen3-Embedding-0.6B"):
        # 初始化代码...
    
    def encode_query(self, query: str):
        # 查询需要特殊的前缀指令
        text = f"Instruct: Given a web search query...\nQuery:{query}"
        return self._encode_with_pooling(text)
    
    def encode_documents(self, documents: List[str]):
        # 文档编码不需要指令前缀
        return self._encode_with_pooling(documents)
    
    def _encode_with_pooling(self, texts):
        # 使用last token pooling而非常见的mean pooling
        outputs = self.model(**encoded)
        embeddings = self._last_token_pool(outputs.last_hidden_state, attention_mask)
        return F.normalize(embeddings, p=2, dim=1)

关键细节：

查询和文档使用不同的编码格式
采用last token pooling而非mean pooling（Qwen官方推荐）
输出向量会进行L2归一化，这对余弦相似度计算很重要

3.2 Reranker模型实现

Reranker的实现方式与常规分类模型不同：

python复制class RerankerModel:
    def rerank(self, query: str, documents: List[str]):
        # 构造特定的prompt格式
        pairs = [
            f"<Instruct>: Given a web search query...\n<Query>: {query}\n<Document>: {doc}"
            for doc in documents
        ]
        
        # 添加系统提示
        prefix = '<|im_start|>system\nJudge whether...'
        suffix = "<|im_end|>\n<|im_start|>assistant\n<think>\n\n</think>\n\n"
        
        # 通过yes/no token的logits计算相关性分数
        logits = self.model(**inputs).logits[:,-1,:]
        true_logit = logits[:,self.token_true_id]
        false_logit = logits[:,self.token_false_id]
        scores = torch.softmax(torch.stack([false_logit, true_logit],dim=1),dim=1)[:,1]

这种设计利用了Qwen3-Reranker的特殊能力，通过语言模型本身对query-document对的相关性进行评分。

4. 知识库核心实现

4.1 知识库类设计

知识库类整合了各个组件：

python复制class KnowledgeBase:
    def __init__(self, db_path="./knowledge_db"):
        self.embedding_model = EmbeddingModel()
        self.reranker_model = RerankerModel()
        self.text_splitter = SimpleTextSplitter()
        
        # 初始化ChromaDB
        self.client = chromadb.PersistentClient(path=db_path)
        self.collection = self.client.get_or_create_collection(
            name="knowledge_base",
            metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
        )
    
    def add_document(self, text: str):
        chunks = self.text_splitter.split_text(text)
        embeddings = self.embedding_model.encode_documents(chunks)
        
        # 存储到ChromaDB
        self.collection.add(
            ids=[f"doc_{i}" for i in range(len(chunks))],
            embeddings=embeddings.numpy().tolist(),
            documents=chunks
        )
    
    def search(self, query: str, top_k=5, use_reranker=True):
        query_embedding = self.embedding_model.encode_query(query)
        
        # 第一阶段：向量检索
        results = self.collection.query(
            query_embeddings=[query_embedding.numpy().tolist()],
            n_results=top_k*3 if use_reranker else top_k
        )
        
        # 第二阶段：Reranker重排序
        if use_reranker:
            return self.reranker_model.rerank(query, results["documents"][0], top_k)
        return results[:top_k]

4.2 文本分割器实现

自实现的文本分割器解决了LangChain的依赖问题：

python复制class SimpleTextSplitter:
    def __init__(self, chunk_size=500, chunk_overlap=50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        # 中文友好的分隔符优先级
        self.separators = ["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
    
    def split_text(self, text: str):
        if len(text) <= self.chunk_size:
            return [text]
        
        chunks = []
        start = 0
        while start < len(text):
            end = start + self.chunk_size
            chunk = text[start:end]
            
            # 尝试在最后一个分隔符处分割
            if end < len(text):
                for sep in self.separators:
                    if sep and sep in chunk:
                        last_sep_pos = chunk.rfind(sep)
                        if last_sep_pos > self.chunk_size * 0.5:
                            chunk = chunk[:last_sep_pos + len(sep)]
                            end = start + len(chunk)
                            break
            
            chunks.append(chunk.strip())
            start = end - self.chunk_overlap
        
        return [ch for ch in chunks if ch]

这个实现考虑了中文文本的特点，比简单的固定长度分割效果更好。

5. 性能优化策略

5.1 显存管理

在多模型场景下，显存管理至关重要：

python复制# 策略1：分散设备负载
kb.load_embedding_model(device="cpu")  # Embedding放在CPU
llm.load(device="cuda:0")  # LLM放在GPU

# 策略2：按需加载
if not is_embedding_loaded:
    kb.load_embedding_model()
# 使用后及时卸载
kb.unload_embedding_model()

5.2 批量处理

批量处理可以显著提升效率：

python复制# 批量编码文档
embeddings = embedding_model.encode_documents(chunks, batch_size=32)

# 批量重排序
reranked = reranker_model.rerank(query, documents, batch_size=16)

5.3 缓存机制

对频繁查询使用缓存：

python复制from functools import lru_cache

@lru_cache(maxsize=100)
def cached_encode_query(query: str):
    return embedding_model.encode_query(query)

6. 部署与使用

6.1 环境准备

推荐使用conda管理环境：

bash复制conda create -n rag python=3.9
conda activate rag
pip install torch transformers chromadb sentencepiece

6.2 模型下载

使用ModelScope加速下载：

bash复制pip install modelscope
python -c "from modelscope import snapshot_download; snapshot_download('Qwen/Qwen3-Embedding-0.6B', cache_dir='./models')"

6.3 启动服务

Gradio界面启动示例：

python复制import gradio as gr
from knowledge_base import KnowledgeBase

kb = KnowledgeBase()
llm = QwenChatModel()

def chat_fn(message, use_kb):
    if use_kb:
        results = kb.search(message, top_k=3)
        context = "\n".join([r["text"] for r in results])
        prompt = f"Context:\n{context}\nQuestion:{message}\nAnswer:"
    else:
        prompt = message
    
    return llm.chat(prompt)

demo = gr.ChatInterface(fn=chat_fn)
demo.launch()

7. 常见问题与解决方案

7.1 模型加载失败

问题现象：模型下载后仍提示找不到
解决方案：

检查模型路径是否正确
确保有足够的磁盘空间
尝试指定绝对路径：

python复制kb = KnowledgeBase(
    embedding_model_path="/absolute/path/to/Qwen3-Embedding-0.6B"
)

7.2 检索效果不佳

可能原因：

文本分割不合理
Embedding模型不适合当前领域
Reranker阈值设置不当

优化方法：

调整分割参数：

python复制text_splitter = SimpleTextSplitter(chunk_size=300, chunk_overlap=30)

尝试不同的prompt模板
调整Reranker的top_k参数

7.3 显存不足

优化策略：

使用4bit量化：

python复制llm.load(load_in_4bit=True)

启用CPU卸载：

python复制kb.load_embedding_model(device="cpu")

限制并发请求数

8. 扩展与定制

8.1 支持更多文件格式

可以通过添加文件解析器来扩展支持范围：

python复制def parse_file(file_path):
    if file_path.endswith('.pdf'):
        import pdfplumber
        with pdfplumber.open(file_path) as pdf:
            return "\n".join([page.extract_text() for page in pdf.pages])
    elif file_path.endswith('.docx'):
        from docx import Document
        doc = Document(file_path)
        return "\n".join([para.text for para in doc.paragraphs])
    # 其他格式处理...

8.2 增量更新支持

实现知识库的增量更新：

python复制def update_document(self, doc_id: str, new_text: str):
    # 先删除旧内容
    self.collection.delete(where={"doc_id": doc_id})
    # 添加新内容
    self.add_document(new_text, doc_id)

8.3 混合检索策略

结合关键词和向量检索：

python复制def hybrid_search(self, query: str, top_k=5, alpha=0.3):
    # 向量检索
    vector_results = self.vector_search(query, top_k*2)
    
    # 关键词检索（需要提前构建倒排索引）
    keyword_results = self.keyword_search(query, top_k*2)
    
    # 混合评分
    combined = []
    for doc in set(vector_results + keyword_results):
        score = alpha*doc['vector_score'] + (1-alpha)*doc['keyword_score']
        combined.append({**doc, 'combined_score': score})
    
    return sorted(combined, key=lambda x: -x['combined_score'])[:top_k]

这个本地RAG系统的实现展示了如何在不依赖高级框架的情况下构建一个功能完整的智能体系统。通过模块化设计和精细的性能优化，它既适合学习RAG原理，也能满足实际生产环境的需求。系统最大的优势在于完全可控，开发者可以根据具体需求灵活调整每个组件的实现细节。