4.1 本地知识库：基于RAG的问答系统

本节导读：学习如何使用Llamafile构建本地知识库问答系统，通过RAG技术实现精准的本地知识检索和问答功能。

学习目标

• 掌握基于Llamafile的RAG系统构建方法
• 学会文档分块和向量化处理技术
• 实现本地知识库的问答功能
• 优化检索效果和问答准确度

核心概念

本节将介绍如何使用Llamafile结合向量数据库技术，构建高效的本地知识库问答系统。通过文档分块、向量化检索和上下文增强等技术，实现对本地文档的精准问答。

环境准备 / 前置知识

• Python 3.8+
• Llamafile基础使用经验（第1章）
• 向量数据库基本概念
• 文本处理基础

分步实战

步骤 1：安装必要依赖

# 安装向量数据库
pip install chromadb sentence-transformers faiss-cpu

# 安装文档处理工具
pip install PyPDF2 python-docx markdown

# 安装Llamafile相关依赖
pip install llama-cpp-python
​

步骤 2：文档预处理模块

import os
import json
import hashlib
from typing import List, Dict, Any
from pathlib import Path
import PyPDF2
import docx
import markdown
from sentence_transformers import SentenceTransformer

class DocumentProcessor:
    """文档预处理类，支持PDF、Word、Markdown等格式"""
    
    def __init__(self, embedding_model='all-MiniLM-L6-v2'):
        self.embedding_model = SentenceTransformer(embedding_model)
        self.chunk_size = 500
        self.chunk_overlap = 50
    
    def extract_text_from_pdf(self, pdf_path: str) -> str:
        """从PDF文件提取文本"""
        text = ""
        with open(pdf_path, 'rb') as file:
            pdf_reader = PyPDF2.PdfReader(file)
            for page in pdf_reader.pages:
                text += page.extract_text() + "\n"
        return text
    
    def extract_text_from_word(self, docx_path: str) -> str:
        """从Word文档提取文本"""
        doc = docx.Document(docx_path)
        return '\n'.join([paragraph.text for paragraph in doc.paragraphs])
    
    def extract_text_from_markdown(self, md_path: str) -> str:
        """从Markdown文件提取文本"""
        with open(md_path, 'r', encoding='utf-8') as file:
            md_content = file.read()
            html_content = markdown.markdown(md_content)
            # 简单提取文本（实际应用中需要更复杂的解析）
            return ' '.join(html_content.split())
    
    def chunk_text(self, text: str) -> List[str]:
        """将文本分割成小块"""
        sentences = text.split('。')
        chunks = []
        current_chunk = ""
        
        for sentence in sentences:
            if len(current_chunk) + len(sentence) <= self.chunk_size:
                current_chunk += sentence + "。"
            else:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                current_chunk = sentence + "。"
        
        if current_chunk:
            chunks.append(current_chunk.strip())
        
        return chunks
    
    def process_document(self, file_path: str) -> List[Dict[str, Any]]:
        """处理单个文档，返回分块后的数据"""
        file_extension = Path(file_path).suffix.lower()
        
        if file_extension == '.pdf':
            text = self.extract_text_from_pdf(file_path)
        elif file_extension == '.docx':
            text = self.extract_text_from_word(file_path)
        elif file_extension == '.md':
            text = self.extract_text_from_markdown(file_path)
        else:
            raise ValueError(f"不支持的文件格式: {file_extension}")
        
        chunks = self.chunk_text(text)
        
        documents = []
        for i, chunk in enumerate(chunks):
            doc_id = hashlib.md5(f"{file_path}_{i}".encode()).hexdigest()
            embedding = self.embedding_model.encode(chunk)
            
            documents.append({
                'id': doc_id,
                'content': chunk,
                'embedding': embedding.tolist(),
                'metadata': {
                    'source': file_path,
                    'chunk_index': i,
                    'total_chunks': len(chunks)
                }
            })
        
        return documents
​

步骤 3：向量数据库管理

import chromadb
import numpy as np
from typing import List, Dict, Any, Optional

class LocalKnowledgeBase:
    """本地知识库管理系统"""
    
    def __init__(self, collection_name: str = "llamafile_knowledge"):
        self.client = chromadb.Client()
        self.collection = self.client.create_collection(name=collection_name)
        self.documents = []
    
    def add_documents(self, documents: List[Dict[str, Any]]) -> None:
        """添加文档到知识库"""
        ids = [doc['id'] for doc in documents]
        embeddings = [doc['embedding'] for doc in documents]
        metadatas = [doc['metadata'] for doc in documents]
        contents = [doc['content'] for doc in documents]
        
        self.collection.add(
            embeddings=embeddings,
            documents=contents,
            metadatas=metadatas,
            ids=ids
        )
        
        self.documents.extend(documents)
    
    def search(self, query: str, top_k: int = 5) -> List[Dict[str, Any]]:
        """搜索相关知识"""
        query_embedding = SentenceTransformer('all-MiniLM-L6-v2').encode(query)
        
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k
        )
        
        search_results = []
        for i, (doc_id, distance, metadata) in enumerate(zip(
            results['ids'][0], 
            results['distances'][0], 
            results['metadatas'][0]
        )):
            search_results.append({
                'id': doc_id,
                'content': results['documents'][0][i],
                'distance': distance,
                'metadata': metadata
            })
        
        return search_results
    
    def get_relevant_context(self, query: str, max_length: int = 2000) -> str:
        """获取相关上下文"""
        search_results = self.search(query, top_k=3)
        
        context_parts = []
        total_length = 0
        
        for result in search_results:
            content = result['content']
            if total_length + len(content) <= max_length:
                context_parts.append(f"文档片段 {result['metadata']['chunk_index']+1}:\n{content}")
                total_length += len(content)
            else:
                break
        
        return "\n\n".join(context_parts)
​

步骤 4：Llamafile RAG问答系统

import time
from typing import Dict, Any, Optional

class LlamafileRAG:
    """基于Llamafile的RAG问答系统"""
    
    def __init__(self, llamafile_path: str, knowledge_base: LocalKnowledgeBase):
        self.llamafile_path = llamafile_path
        self.knowledge_base = knowledge_base
        self.llm = None
        self.initialize_llm()
    
    def initialize_llm(self):
        """初始化Llamafile模型"""
        try:
            from llama_cpp import Llama
            self.llm = Llama(
                model_path=self.llamafile_path,
                n_ctx=4096,
                n_threads=4,
                n_batch=512,
                use_mmap=True,
                use_mlock=False,
                verbose=False
            )
            print("Llamafile模型初始化成功")
        except Exception as e:
            print(f"Llamafile模型初始化失败: {e}")
            self.llm = None
    
    def generate_prompt(self, query: str, context: str) -> str:
        """生成RAG提示词"""
        prompt = f"""基于以下本地知识库信息回答问题：

**相关知识：**
{context}

**问题：** {query}

**回答要求：**
1. 基于提供的本地知识库信息回答
2. 如果知识库中没有相关信息，请明确说明
3. 回答要准确、简洁、有针对性
4. 如果适用，请引用具体的文档片段

**回答：**
"""
        return prompt
    
    def ask(self, query: str, temperature: float = 0.7, max_tokens: int = 1000) -> Dict[str, Any]:
        """执行RAG问答"""
        if not self.llm:
            return {
                'success': False,
                'error': 'Llamafile模型未初始化',
                'answer': None
            }
        
        # 获取相关上下文
        context = self.knowledge_base.get_relevant_context(query)
        
        # 生成提示词
        prompt = self.generate_prompt(query, context)
        
        try:
            # 使用Llamafile生成回答
            start_time = time.time()
            response = self.llm(
                prompt=prompt,
                max_tokens=max_tokens,
                temperature=temperature,
                stop=["\n\n", "**"],
                echo=False
            )
            end_time = time.time()
            
            answer = response['choices'][0]['text'].strip()
            
            return {
                'success': True,
                'answer': answer,
                'context': context,
                'response_time': end_time - start_time,
                'tokens_used': len(prompt.split()) + len(answer.split())
            }
        
        except Exception as e:
            return {
                'success': False,
                'error': str(e),
                'answer': None,
                'context': context
            }

# 使用示例
def setup_knowledge_base():
    """设置知识库"""
    processor = DocumentProcessor()
    knowledge_base = LocalKnowledgeBase()
    
    # 处理示例文档（这里用示例文本代替）
    sample_docs = [
        "Python是一种高级编程语言，以其简洁的语法和强大的功能而闻名。",
        "机器学习是人工智能的一个分支，专注于让计算机系统从数据中学习。",
        "Llamafile是一个单文件的大模型运行时工具，无需复杂环境配置即可运行开源大模型。"
    ]
    
    for doc in sample_docs:
        chunks = processor.chunk_text(doc)
        documents = []
        for i, chunk in enumerate(chunks):
            doc_id = hashlib.md5(f"sample_{i}".encode()).hexdigest()
            embedding = processor.embedding_model.encode(chunk)
            
            documents.append({
                'id': doc_id,
                'content': chunk,
                'embedding': embedding.tolist(),
                'metadata': {
                    'source': 'sample',
                    'chunk_index': i,
                    'total_chunks': len(chunks)
                }
            })
        
        knowledge_base.add_documents(documents)
    
    return knowledge_base

def demo_rag_system():
    """演示RAG系统"""
    knowledge_base = setup_knowledge_base()
    
    # 创建RAG系统（实际使用时需要指定llamafile_path）
    rag_system = LlamafileRAG("path/to/your/model.gguf", knowledge_base)
    
    # 示例问答
    questions = [
        "什么是Python？",
        "机器学习是什么？",
        "Llamafile有什么特点？"
    ]
    
    for question in questions:
        result = rag_system.ask(question)
        if result['success']:
            print(f"问题: {question}")
            print(f"回答: {result['answer']}")
            print(f"响应时间: {result['response_time']:.2f}秒")
            print("-" * 50)
        else:
            print(f"问题: {question}")
            print(f"错误: {result['error']}")
            print("-" * 50)
​

完整示例

"""
完整的Llamafile RAG知识库问答系统
作者：Llamafile教程
日期：2026年
"""

import os
import sys
import logging
from pathlib import Path

# 设置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class RAGApplication:
    """RAG应用程序主类"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.knowledge_base = None
        self.rag_system = None
        self.setup_logging()
    
    def setup_logging(self):
        """设置日志配置"""
        log_file = self.config.get('log_file', 'rag_system.log')
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(log_file),
                logging.StreamHandler()
            ]
        )
    
    def initialize_system(self):
        """初始化RAG系统"""
        try:
            # 初始化知识库
            from document_processor import DocumentProcessor
            from knowledge_base import LocalKnowledgeBase
            
            processor = DocumentProcessor(
                chunk_size=self.config.get('chunk_size', 500),
                chunk_overlap=self.config.get('chunk_overlap', 50)
            )
            
            self.knowledge_base = LocalKnowledgeBase(
                collection_name=self.config.get('collection_name', 'llamafile_knowledge')
            )
            
            # 处理文档
            doc_dir = Path(self.config.get('document_directory', './documents'))
            if doc_dir.exists():
                self.process_documents(doc_dir, processor)
            
            # 初始化RAG系统
            from llamafile_rag import LlamafileRAG
            self.rag_system = LlamafileRAG(
                llamafile_path=self.config.get('llamafile_path'),
                knowledge_base=self.knowledge_base
            )
            
            logger.info("RAG系统初始化成功")
            
        except Exception as e:
            logger.error(f"系统初始化失败: {e}")
            raise
    
    def process_documents(self, doc_dir: Path, processor: DocumentProcessor):
        """处理目录中的所有文档"""
        processed_count = 0
        
        for file_path in doc_dir.glob('*'):
            if file_path.is_file() and file_path.suffix.lower() in ['.pdf', '.docx', '.md']:
                try:
                    logger.info(f"处理文档: {file_path}")
                    documents = processor.process_document(str(file_path))
                    self.knowledge_base.add_documents(documents)
                    processed_count += 1
                except Exception as e:
                    logger.error(f"处理文档 {file_path} 失败: {e}")
        
        logger.info(f"成功处理 {processed_count} 个文档")
    
    def ask_question(self, question: str, **kwargs) -> Dict[str, Any]:
        """问答接口"""
        if not self.rag_system:
            return {'success': False, 'error': '系统未初始化'}
        
        return self.rag_system.ask(question, **kwargs)

# 配置示例
config = {
    'llamafile_path': './models/llama-2-7b-chat.gguf',
    'document_directory': './knowledge_base',
    'chunk_size': 500,
    'chunk_overlap': 50,
    'collection_name': 'company_knowledge',
    'log_file': 'rag_system.log'
}

# 使用示例
if __name__ == "__main__":
    app = RAGApplication(config)
    app.initialize_system()
    
    # 问答示例
    questions = [
        "公司的主要产品是什么？",
        "如何申请年假？",
        "技术支持联系方式？"
    ]
    
    for question in questions:
        result = app.ask_question(question)
        if result['success']:
            print(f"Q: {question}")
            print(f"A: {result['answer']}")
            print(f"响应时间: {result['response_time']:.2f}秒")
            print("-" * 60)
        else:
            print(f"Q: {question}")
            print(f"错误: {result['error']}")
            print("-" * 60)
​

常见问题 FAQ

Q1：如何选择合适的分块大小？

A：分块大小需要根据文档类型和问答需求来确定：

• 技术文档：300-500字，保持概念的完整性
• 通用文档：500-800字，提高召回率
• 法律文档：200-300字，确保准确性
• 一般建议：从500字开始，根据效果调整

Q2：向量数据库选择哪个比较好？

A：根据需求选择：

• ChromaDB：轻量级，适合本地部署，本教程使用
• FAISS：Facebook开源，适合大规模向量搜索
• Pinecone：云服务，适合生产环境
• Weaviate：开源向量数据库，支持GraphQL

Q3：如何优化RAG系统的准确性？

A：从以下几个方面优化：

1. 文档分块策略：调整chunk_size和chunk_overlap
2. 向量模型选择：使用更适合语义相似度的模型
3. 提示词优化：设计更清晰的提示词模板
4. 重排序机制：对检索结果进行重新排序
5. 多轮对话：实现多轮问答和上下文管理

Q4：Llamafile与RAG结合有什么优势？

A：

1. 本地部署：数据完全本地化，保护隐私
2. 低延迟：本地推理速度快，响应及时
3. 可扩展：可以集成多种向量数据库
4. 成本效益：无需云服务费用
5. 定制化：可以根据需求定制功能

最佳实践与避坑

• 实践1：在添加新文档后，建议重新生成向量和测试问答效果
• 实践2：定期清理过期的文档和冗余的向量数据
• 实践3：设置监控机制，记录问答质量和响应时间
• 坑点1：避免分块过大导致上下文信息丢失
• 坑点2：注意内存使用，大量文档可能导致内存溢出
• 坑点3：向量模型选择不当会影响检索效果

本节小结

通过本节的学习，我们掌握了基于Llamafile构建本地知识库问答系统的完整流程。从文档预处理、向量数据库管理到RAG问答实现，形成了一个完整的解决方案。这个系统可以用于企业内部知识管理、文档问答、智能客服等多种场景，具有很强的实用价值。

下一节我们将学习如何使用Llamafile构建代码助手工具，进一步提升开发效率。

关键词：Llamafile, RAG, 知识库, 向量数据库, 本地部署, 问答系统, 文档检索, 语义搜索
难度：进阶
预计阅读：45分钟