最近在尝试将企业文档、个人笔记等非结构化数据接入大模型时发现单纯的向量检索Naive RAG效果总是不尽如人意要么答非所问要么“幻觉”严重要么无法追溯答案来源。经过一系列项目实践和技术选型我发现“编译式RAG”是解决这些痛点的关键路径。本文将为你完整拆解一套基于“三层架构”的编译式RAG实战方案涵盖从知识库构建、智能问答到自动更新的全流程并深入对比Naive RAG、高级RAG和编译式RAG的核心差异帮你做出最适合自己场景的技术选型。1. RAG技术演进与核心概念从检索到编译在深入实战之前我们有必要厘清RAGRetrieval-Augmented Generation检索增强生成技术的演进脉络这直接决定了我们架构设计的出发点。1.1 RAG的三种范式如何选择1. Naive RAG传统/朴素RAG这是最常见的入门方案流程可概括为“索引-检索-生成”。流程将文档切块 - 向量化存入向量数据库 - 用户提问时进行语义检索 - 将检索到的文本块直接拼接到提示词中交给大模型生成答案。优点实现简单快速上手。缺点检索精度低简单的语义相似度检索容易引入无关内容。上下文碎片化检索到的文本块可能丢失原文的章节结构、逻辑关系。幻觉风险高当检索到不相关或冲突信息时模型容易产生错误答案。无法溯源答案可能综合了多个片段难以精确定位到原文出处。2. Advanced RAG高级RAG为了克服Naive RAG的缺陷Advanced RAG在检索前后增加了优化步骤。核心改进检索前对文档进行更精细的预处理如重写用户查询Query Rewriting、假设性文档嵌入HyDE。检索中尝试混合检索结合关键词与向量、多向量检索、递归检索等。检索后对检索结果进行重排序Reranking、过滤或压缩。优点显著提升了检索质量减少了无关信息干扰。缺点本质上仍是“检索-拼接”模式对于复杂、需要深度推理和多步处理的问题其能力仍有天花板。3. Compilational RAG编译式RAG这是本文的重点它借鉴了程序编译的思想将知识库视为“源代码”通过一个“编译”过程将其转化为更结构化、更易于模型理解和推理的中间表示。核心思想不是简单地把文档片段扔给模型而是先对知识库进行深度加工、索引和重构生成一个富含元数据、关系、摘要的“知识图谱”或“结构化索引”。问答时系统在这个高级索引上进行“查询”和“推理”而不仅仅是“检索”。关键特征深度索引不仅存储文本块和向量还提取实体、关系、摘要、章节大纲等。查询编译将用户自然语言问题“编译”成对结构化索引的查询指令。动态知识整合能够综合多个来源、不同粒度的信息构建连贯的答案。精准溯源答案可以与索引中的特定节点如某章节、某实体精确关联。选择建议需求简单、文档规整、问答直接- 从Naive RAG开始验证可行性。追求更高答案质量、减少幻觉、文档类型多样- 必须采用Advanced RAG。构建企业级、生产环境知识库要求高准确性、强溯源、复杂推理- 编译式RAG是更优解。1.2 为什么需要“三层架构”一个健壮的编译式RAG系统不能是“一次性”的管道而应该是一个可持续运营的“知识系统”。三层架构为此而生数据层负责原始知识的接入、存储与管理。处理不同格式PDF、Word、Markdown、网页、不同来源的数据。索引层核心负责“编译”过程。对原始数据进行解析、清洗、切分、向量化、元数据提取、关系构建形成结构化的“编译后知识索引”。这是编译式RAG区别于前两者的核心。应用层面向用户的接口。接收问题利用索引层进行智能查询与推理生成答案并附带溯源同时管理对话会话。2. 环境准备与项目初始化我们将使用Python生态中成熟的工具链来构建这个系统。2.1 技术栈选型与版本说明编程语言Python 3.9核心框架LangChain / LlamaIndex。本文示例将侧重思路部分代码使用LangChain因其生态丰富但LlamaIndex在索引深度上更有特色可根据项目偏好选择。文本嵌入模型text-embedding-ada-002(OpenAI API) 或BAAI/bge-large-zh-v1.5(本地部署)。前者效果稳定后者数据隐私性好。向量数据库Chroma轻量开发友好或 Milvus / Weaviate生产级性能强。本文用Chroma演示。大语言模型GPT-4/3.5-Turbo (OpenAI API) 或 ChatGLM3、Qwen等开源模型本地/API。其他工具PyPDF2/pdfplumber解析PDFpython-docx解析WordBeautifulSoup4解析HTMLsentence-transformers本地嵌入模型。重要提示以下版本仅为示例请根据你的实际环境调整依赖版本。2.2 创建项目并安装依赖# 创建项目目录 mkdir llm-wiki-compilational-rag cd llm-wiki-compilational-rag # 创建虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 创建 requirements.txt 文件requirements.txt内容示例langchain0.1.0 langchain-community0.0.10 langchain-openai0.0.5 chromadb0.4.22 pypdf23.0.1 pdfplumber0.10.3 python-docx1.1.0 beautifulsoup44.12.2 sentence-transformers2.2.2 openai1.6.1 tiktoken0.5.2 streamlit1.29.0 # 用于构建简单Web界面安装依赖pip install -r requirements.txt2.3 项目结构设计llm-wiki-compilational-rag/ ├── config/ # 配置文件 │ └── settings.py ├── data_layer/ # 数据层 │ ├── connectors/ # 数据连接器PDF、Web等 │ ├── storage/ # 原始文件存储 │ └── data_manager.py ├── index_layer/ # 索引层核心 │ ├── processors/ # 文档处理器 │ ├── compilers/ # 知识编译器 │ ├── vector_store/ # 向量存储操作 │ └── index_manager.py ├── app_layer/ # 应用层 │ ├── agents/ # 智能体/问答链 │ ├── web_ui/ # Web界面 │ └── api/ # API接口 ├── knowledge_base/ # 知识库目录 │ ├── raw_docs/ # 存放原始文档 │ └── compiled_index/ # 存放编译后的索引 ├── main.py # 主程序入口 └── requirements.txt3. 数据层实现多源知识接入与管理数据层是知识库的源头需要稳定可靠。3.1 实现通用文档加载器我们创建一个基类和针对不同格式的加载器。# data_layer/connectors/base_loader.py from abc import ABC, abstractmethod from typing import List, Dict, Any import hashlib class BaseDocumentLoader(ABC): 文档加载器基类 def __init__(self, file_path: str): self.file_path file_path self.metadata { source: file_path, file_hash: self._calculate_file_hash(file_path) } def _calculate_file_hash(self, file_path: str) - str: 计算文件哈希值用于内容变更检测 hash_md5 hashlib.md5() with open(file_path, rb) as f: for chunk in iter(lambda: f.read(4096), b): hash_md5.update(chunk) return hash_md5.hexdigest() abstractmethod def load(self) - List[Dict[str, Any]]: 加载文档并返回字典列表。 每个字典应包含: {content: str, metadata: dict} metadata至少包含: source, page (如果有) pass # data_layer/connectors/pdf_loader.py import pdfplumber from .base_loader import BaseDocumentLoader class PDFLoader(BaseDocumentLoader): def load(self) - List[Dict[str, Any]]: documents [] try: with pdfplumber.open(self.file_path) as pdf: for page_num, page in enumerate(pdf.pages, start1): text page.extract_text() if text.strip(): # 忽略空白页 doc_metadata self.metadata.copy() doc_metadata.update({page: page_num, type: pdf}) documents.append({ content: text, metadata: doc_metadata }) except Exception as e: print(f加载PDF文件 {self.file_path} 失败: {e}) return documents # data_layer/connectors/markdown_loader.py import markdown from bs4 import BeautifulSoup from .base_loader import BaseDocumentLoader class MarkdownLoader(BaseDocumentLoader): def load(self) - List[Dict[str, Any]]: documents [] try: with open(self.file_path, r, encodingutf-8) as f: md_content f.read() # 将Markdown转换为纯文本同时可保留结构 html markdown.markdown(md_content) soup BeautifulSoup(html, html.parser) text soup.get_text(separator\n, stripTrue) doc_metadata self.metadata.copy() doc_metadata.update({type: markdown}) # 可以进一步解析标题作为元数据 documents.append({ content: text, metadata: doc_metadata }) except Exception as e: print(f加载Markdown文件 {self.file_path} 失败: {e}) return documents3.2 实现数据管理器负责调度不同的加载器并管理原始文档的存储状态。# data_layer/data_manager.py import os from typing import List, Dict, Any import json from .connectors.pdf_loader import PDFLoader from .connectors.markdown_loader import MarkdownLoader class DataManager: 数据管理器统一处理知识源 def __init__(self, raw_docs_dir: str ./knowledge_base/raw_docs): self.raw_docs_dir raw_docs_dir os.makedirs(self.raw_docs_dir, exist_okTrue) self._loaders { .pdf: PDFLoader, .md: MarkdownLoader, .txt: None, # 可扩展 .docx: None, # 可扩展 } self._state_file os.path.join(self.raw_docs_dir, _data_state.json) self._load_state() def _load_state(self): 加载已处理文件的状态记录 if os.path.exists(self._state_file): with open(self._state_file, r, encodingutf-8) as f: self.processed_files json.load(f) else: self.processed_files {} # {file_path: file_hash} def _save_state(self): 保存状态 with open(self._state_file, w, encodingutf-8) as f: json.dump(self.processed_files, f, ensure_asciiFalse, indent2) def add_document(self, file_path: str) - List[Dict[str, Any]]: 添加单个文档到知识库。 返回加载的文档内容列表。 if not os.path.exists(file_path): raise FileNotFoundError(f文件不存在: {file_path}) # 检查文件类型 ext os.path.splitext(file_path)[1].lower() loader_class self._loaders.get(ext) if not loader_class: raise ValueError(f不支持的文件类型: {ext}) # 实例化加载器并加载 loader loader_class(file_path) loaded_docs loader.load() if loaded_docs: # 复制文件到知识库目录可选便于集中管理 import shutil target_path os.path.join(self.raw_docs_dir, os.path.basename(file_path)) shutil.copy2(file_path, target_path) # 更新状态 self.processed_files[target_path] loader.metadata[file_hash] self._save_state() print(f文档已添加: {os.path.basename(file_path)}) return loaded_docs def scan_and_load_new(self, source_dir: str) - List[Dict[str, Any]]: 扫描目录仅加载新增或修改过的文件。 返回所有新加载的文档内容。 all_new_docs [] for filename in os.listdir(source_dir): file_path os.path.join(source_dir, filename) if os.path.isfile(file_path): ext os.path.splitext(filename)[1].lower() if ext not in self._loaders: continue # 计算当前哈希 loader self._loaders[ext](file_path) current_hash loader.metadata[file_hash] # 检查是否为新文件或已修改 target_path_in_kb os.path.join(self.raw_docs_dir, filename) if target_path_in_kb not in self.processed_files or \ self.processed_files[target_path_in_kb] ! current_hash: print(f检测到新/已修改文件: {filename}) docs self.add_document(file_path) all_new_docs.extend(docs) else: print(f文件无变化跳过: {filename}) return all_new_docs4. 索引层实现编译式RAG的核心这是编译式RAG的“编译”环节我们将对原始文本进行深度处理构建结构化索引。4.1 文档处理器超越简单切分传统RAG的简单文本切分RecursiveCharacterTextSplitter会破坏文档结构。我们需要更智能的处理器。# index_layer/processors/smart_text_splitter.py from langchain.text_splitter import RecursiveCharacterTextSplitter import re class SmartTextSplitter: 智能文本分割器。 1. 尝试按章节标题分割保留结构。 2. 对于无结构的文本再按长度分割但尽量保证句子完整。 def __init__(self, chunk_size: int 1000, chunk_overlap: int 200): self.chunk_size chunk_size self.chunk_overlap chunk_overlap self.fallback_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , , , , ] ) def split_by_headings(self, text: str, metadata: dict) - List[Dict[str, Any]]: 尝试根据Markdown或常见标题格式分割 chunks [] # 匹配不同层级的标题 heading_pattern r(^#{1,6}\s.$|^第[一二三四五六七八九十][章|节|条].$) lines text.split(\n) current_chunk [] current_heading 起始 heading_level 0 for line in lines: line_stripped line.strip() # 判断是否为标题行 if re.match(heading_pattern, line_stripped): # 保存上一个块 if current_chunk: chunk_text \n.join(current_chunk) chunk_meta metadata.copy() chunk_meta.update({heading: current_heading, heading_level: heading_level}) chunks.append({content: chunk_text, metadata: chunk_meta}) # 开始新块 current_chunk [line_stripped] current_heading line_stripped # 简单判断标题级别 if line_stripped.startswith(#): heading_level line_stripped.count(#) else: heading_level 7 # 非Markdown标题给一个默认高级别 else: current_chunk.append(line) # 处理最后一个块 if current_chunk: chunk_text \n.join(current_chunk) chunk_meta metadata.copy() chunk_meta.update({heading: current_heading, heading_level: heading_level}) chunks.append({content: chunk_text, metadata: chunk_meta}) # 如果按标题没分割出内容或者块太大则用fallback分割器 final_chunks [] for chunk in chunks: if len(chunk[content]) self.chunk_size * 1.5: # 如果块太大 sub_chunks self.fallback_splitter.split_text(chunk[content]) for sub in sub_chunks: sub_meta chunk[metadata].copy() final_chunks.append({content: sub, metadata: sub_meta}) else: final_chunks.append(chunk) return final_chunks if final_chunks else self.fallback_split(text, metadata) def fallback_split(self, text: str, metadata: dict) - List[Dict[str, Any]]: 后备分割方案 texts self.fallback_splitter.split_text(text) return [{content: t, metadata: metadata.copy()} for t in texts] def split_documents(self, documents: List[Dict[str, Any]]) - List[Dict[str, Any]]: 分割文档列表 all_chunks [] for doc in documents: content doc[content] metadata doc[metadata] chunks self.split_by_headings(content, metadata) all_chunks.extend(chunks) return all_chunks4.2 知识编译器构建结构化索引这是编译式RAG的灵魂。我们将为每个文本块生成摘要、关键词并尝试提取实体和关系简化版。# index_layer/compilers/knowledge_compiler.py from langchain.chat_models import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage from typing import List, Dict, Any import json class KnowledgeCompiler: 知识编译器。 对文本块进行深度处理提取结构化信息丰富元数据。 def __init__(self, llm_api_key: str, llm_model: str gpt-3.5-turbo): self.llm ChatOpenAI( openai_api_keyllm_api_key, model_namellm_model, temperature0.1 # 低温度保证稳定性 ) def compile_chunk(self, chunk: Dict[str, Any]) - Dict[str, Any]: 编译单个文本块。 返回增强后的块包含摘要、关键词、实体等。 content chunk[content] metadata chunk[metadata] # 1. 生成摘要 summary_prompt f 请为以下文本生成一个简洁的摘要不超过100字 {content[:1500]} # 限制输入长度 summary self._call_llm(summary_prompt) # 2. 提取关键词 keyword_prompt f 请从以下文本中提取3-5个核心关键词或短语 {content[:1500]} 请以JSON列表格式返回例如[关键词1, 关键词2] keyword_response self._call_llm(keyword_prompt) try: keywords json.loads(keyword_response) except: keywords [] # 3. 提取实体简化示例生产环境可用NER模型 entity_prompt f 识别以下文本中提到的主要实体如人名、组织名、产品名、技术术语等 {content[:1500]} 请以JSON列表格式返回每个实体是一个对象包含name和type字段。 例如[{{name: Python, type: 技术}}, {{name: 张三, type: 人名}}] entity_response self._call_llm(entity_prompt) try: entities json.loads(entity_response) except: entities [] # 构建增强后的元数据 enhanced_metadata metadata.copy() enhanced_metadata.update({ summary: summary, keywords: keywords, entities: entities, compiled: True # 标记已编译 }) return { content: content, metadata: enhanced_metadata } def _call_llm(self, prompt: str) - str: 调用LLM的辅助函数 try: response self.llm([HumanMessage(contentprompt)]) return response.content.strip() except Exception as e: print(fLLM调用失败: {e}) return def compile_batch(self, chunks: List[Dict[str, Any]]) - List[Dict[str, Any]]: 批量编译生产环境应考虑异步和限流 compiled_chunks [] for i, chunk in enumerate(chunks): print(f编译进度: {i1}/{len(chunks)}) compiled self.compile_chunk(chunk) compiled_chunks.append(compiled) return compiled_chunks4.3 向量化与存储管理将编译后的知识存入向量数据库并维护元数据索引。# index_layer/vector_store/vector_manager.py import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import numpy as np from typing import List, Dict, Any import uuid class VectorStoreManager: 向量存储管理器 def __init__(self, persist_dir: str ./knowledge_base/compiled_index/chroma_db, embedding_model_name: str BAAI/bge-large-zh-v1.5): self.persist_dir persist_dir # 初始化Chroma客户端 self.client chromadb.PersistentClient( pathpersist_dir, settingsSettings(anonymized_telemetryFalse) ) # 加载本地嵌入模型或使用OpenAI Embeddings self.embedding_model SentenceTransformer(embedding_model_name) self.collection self.client.get_or_create_collection( namecompiled_knowledge, metadata{hnsw:space: cosine} # 使用余弦相似度 ) def _generate_embedding(self, text: str) - List[float]: 生成文本向量 # 这里使用本地模型也可替换为OpenAIEmbeddings embedding self.embedding_model.encode(text) return embedding.tolist() def add_documents(self, compiled_chunks: List[Dict[str, Any]]): 将编译后的文档块添加到向量库 ids [] embeddings [] metadatas [] documents [] for chunk in compiled_chunks: chunk_id str(uuid.uuid4()) content chunk[content] metadata chunk[metadata] # 生成向量 embedding self._generate_embedding(content) ids.append(chunk_id) embeddings.append(embedding) metadatas.append(metadata) documents.append(content) # 批量添加到Chroma if ids: self.collection.add( idsids, embeddingsembeddings, metadatasmetadatas, documentsdocuments ) print(f成功添加 {len(ids)} 个文档块到向量库。) def similarity_search(self, query: str, k: int 5, filter_conditions: dict None) - List[Dict[str, Any]]: 相似度搜索 query_embedding self._generate_embedding(query) results self.collection.query( query_embeddings[query_embedding], n_resultsk, wherefilter_conditions # 可进行元数据过滤如 where{source: xxx.pdf} ) # 格式化结果 formatted_results [] if results[ids]: for i in range(len(results[ids][0])): formatted_results.append({ id: results[ids][0][i], content: results[documents][0][i], metadata: results[metadatas][0][i], distance: results[distances][0][i] }) return formatted_results4.4 索引管理器串联整个流程# index_layer/index_manager.py from .processors.smart_text_splitter import SmartTextSplitter from .compilers.knowledge_compiler import KnowledgeCompiler from .vector_store.vector_manager import VectorStoreManager class IndexManager: 索引管理器协调处理、编译、存储全流程 def __init__(self, llm_api_key: str, embedding_model: str BAAI/bge-large-zh-v1.5): self.splitter SmartTextSplitter(chunk_size800, chunk_overlap150) self.compiler KnowledgeCompiler(llm_api_keyllm_api_key) self.vector_manager VectorStoreManager(embedding_model_nameembedding_model) def build_index(self, raw_documents: List[Dict[str, Any]]): 从原始文档构建完整索引 print(步骤1/4: 智能分割文档...) chunks self.splitter.split_documents(raw_documents) print(f分割为 {len(chunks)} 个块。) print(步骤2/4: 编译知识块提取摘要、关键词等...) compiled_chunks self.compiler.compile_batch(chunks) print(知识编译完成。) print(步骤3/4: 生成向量并存入数据库...) self.vector_manager.add_documents(compiled_chunks) print(索引构建完成) def search(self, query: str, k: int 5, **filters) - List[Dict[str, Any]]: 在索引中搜索 return self.vector_manager.similarity_search(query, kk, filter_conditionsfilters) def get_collection_stats(self): 获取索引统计信息 collection self.vector_manager.collection count collection.count() return {total_chunks: count}5. 应用层实现智能问答与溯源应用层负责利用构建好的“编译后索引”进行智能问答。5.1 实现带溯源的问答链# app_layer/agents/qa_agent.py from langchain.chat_models import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.runnable import RunnablePassthrough from langchain.schema.output_parser import StrOutputParser from typing import List, Dict, Any class QAAgentWithCitation: 支持溯源引用的问答智能体 def __init__(self, llm_api_key: str, index_manager): self.llm ChatOpenAI( openai_api_keyllm_api_key, model_namegpt-3.5-turbo-16k, # 使用长上下文模型 temperature0.1 ) self.index_manager index_manager # 定义提示词模板 self.prompt_template ChatPromptTemplate.from_messages([ (system, 你是一个专业的知识库助手请根据提供的参考信息回答问题。 如果参考信息不足以回答问题请明确告知“根据现有知识无法回答”。 你的回答必须严格基于参考信息不要编造信息。 在回答的最后请列出你所参考的信息来源出处。), (human, 请回答以下问题 问题{question} 参考信息 {context} 请给出答案) ]) # 构建链 self.qa_chain ( {context: self._retrieve_context, question: RunnablePassthrough()} | self.prompt_template | self.llm | StrOutputParser() ) def _retrieve_context(self, question: str) - str: 检索与问题相关的上下文并格式化 search_results self.index_manager.search(question, k4) if not search_results: return 没有找到相关参考信息。 # 格式化上下文附上来源 context_parts [] for i, res in enumerate(search_results): source res[metadata].get(source, 未知来源) page res[metadata].get(page, N/A) heading res[metadata].get(heading, 无标题) # 使用摘要如果存在或截取内容 summary res[metadata].get(summary, res[content][:300] ...) context_parts.append( f[参考片段 {i1}来源{source}页码{page}章节{heading}]\n f摘要{summary}\n f详细内容{res[content][:500]}\n f--- ) return \n.join(context_parts) def ask(self, question: str) - Dict[str, Any]: 提问并返回答案及溯源信息 # 先进行检索 search_results self.index_manager.search(question, k4) # 生成答案 answer self.qa_chain.invoke(question) # 准备溯源信息 citations [] for res in search_results: citations.append({ content_snippet: res[content][:200], source: res[metadata].get(source), page: res[metadata].get(page), heading: res[metadata].get(heading), score: 1 - res[distance] # 近似相关度分数 }) return { question: question, answer: answer, citations: citations, retrieved_count: len(search_results) }5.2 构建简单的Web界面Streamlit# app_layer/web_ui/app.py import streamlit as st import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from index_layer.index_manager import IndexManager from app_layer.agents.qa_agent import QAAgentWithCitation from data_layer.data_manager import DataManager def main(): st.set_page_config(page_titleLLM Wiki - 编译式RAG知识库, layoutwide) st.title( LLM Wiki - 编译式RAG知识库系统) # 侧边栏 - 知识库管理 with st.sidebar: st.header(知识库管理) uploaded_file st.file_uploader(上传文档PDF/Markdown, type[pdf, md]) if uploaded_file is not None: # 保存上传的文件 save_path os.path.join(./knowledge_base/raw_docs, uploaded_file.name) with open(save_path, wb) as f: f.write(uploaded_file.getbuffer()) st.success(f已上传: {uploaded_file.name}) # 添加到数据管理器并构建索引 if st.button(构建/更新索引): with st.spinner(正在处理文档并构建索引...): data_mgr DataManager() raw_docs data_mgr.add_document(save_path) if raw_docs: # 初始化索引管理器需配置你的API Key index_mgr IndexManager(llm_api_keyst.secrets[OPENAI_API_KEY]) index_mgr.build_index(raw_docs) st.success(索引构建成功) else: st.error(文档加载失败请检查文件格式。) st.divider() st.header(系统状态) # 这里可以显示索引统计等信息 # 主界面 - 问答 st.header(智能问答) question st.text_input(请输入你的问题, placeholder例如RAG的核心思想是什么) if question: with st.spinner(正在检索并生成答案...): # 初始化问答智能体需提前构建好索引 index_mgr IndexManager(llm_api_keyst.secrets[OPENAI_API_KEY]) qa_agent QAAgentWithCitation( llm_api_keyst.secrets[OPENAI_API_KEY], index_managerindex_mgr ) result qa_agent.ask(question) # 显示答案 st.subheader(答案) st.write(result[answer]) # 显示溯源 st.subheader(参考来源) for i, cite in enumerate(result[citations]): with st.expander(f来源 {i1}: {cite[source]} - {cite[heading]}): st.caption(f相关度: {cite[score]:.2%}) st.write(cite[content_snippet]) st.text(f页码: {cite[page]}) if __name__ __main__: # 需要先设置OPENAI_API_KEY环境变量或在secrets.toml中配置 main()6. 自动知识更新与维护一个实用的知识库必须支持内容更新。我们可以在数据层的基础上增加定时扫描和增量索引构建。# main.py 中的更新逻辑示例 import schedule import time from data_layer.data_manager import DataManager from index_layer.index_manager import IndexManager def scheduled_update(): 定时任务扫描目录更新索引 print(开始执行定时知识库更新检查...) data_mgr DataManager() index_mgr IndexManager(llm_api_keyyour-api-key) # 假设我们监控一个源目录 source_dir ./source_docs new_docs data_mgr.scan_and_load_new(source_dir) if new_docs: print(f发现 {len(new_docs)} 个新文档/更新开始增量构建索引...) # 注意这里简化处理实际增量更新需要更精细的管理如删除旧版本块 index_mgr.build_index(new_docs) print(增量索引更新完成。) else: print(未发现变更。) # 每天凌晨2点执行更新 schedule.every().day.at(02:00).do(scheduled_update) if __name__ __main__: print(知识库更新服务启动...) while True: schedule.run_pending() time.sleep(60)7. 常见问题与排查思路在构建和运行编译式RAG系统时你可能会遇到以下典型问题。问题现象可能原因排查思路与解决方案文档加载失败或乱码1. 文件格式不支持。2. 文件编码问题。3. 文件损坏或受密码保护。1. 检查文件扩展名实现对应的加载器。2. 指定正确的编码如utf-8,gbk。3. 尝试用其他软件打开文件确认是否正常。向量检索结果不相关1. 文本分割不合理上下文碎片化。2. 嵌入模型不匹配如中英文。3. 检索参数k值不合适。1. 优化SmartTextSplitter尝试按标题、段落分割。2. 为中文文本选择BAAI/bge-*系列模型。3. 调整k值如从5调到10或启用重排序Reranker。LLM回答出现“幻觉”1. 检索到的上下文不足或无关。2. 提示词Prompt未强制要求基于上下文。3. 模型温度Temperature设置过高。1. 检查检索结果质量优化索引。2. 在系统提示词中明确指令“严格基于以下参考信息”。3. 将temperature调低如0.1。知识编译摘要/提取速度慢1. 串行调用LLM。2. 模型响应慢或网络延迟。3. 文本块过大或过多。1. 实现异步批量处理asyncio。2. 考虑使用更快的模型如gpt-3.5-turbo或本地轻量模型进行初步提取。3. 调整chunk_size避免单个块过长。溯源信息不准确1. 元数据如页码、标题在分割时丢失。2. 答案综合了多个片段难以对应单一来源。1. 确保分割器正确传递并保留原始元数据。2. 在答案中声明“综合以下信息”并列出所有相关片段来源。系统内存/磁盘占用过高1. 向量数据库未做持久化或分片。2. 原始文档和索引未定期清理。1. 对于大规模知识库使用Milvus、Weaviate等支持集群和磁盘索引的数据库。2. 制定数据归档策略移除过时文档及其索引。8. 最佳实践与工程建议将编译式RAG投入生产环境需要关注以下工程细节。1. 索引构建优化分层索引不仅存储文本块向量还可以构建文档级摘要向量、章节标题向量实现从粗到细的多级检索。混合检索结合语义向量检索和关键词BM25检索提升召回率。可以使用langchain.retrievers中的EnsembleRetriever。元数据过滤为每个文档块添加丰富的元数据如文档类型、创建日期、部门、权限等级在检索时支持高效的过滤。2. 知识编译策略成本与质量权衡为每个文本块调用LLM生成摘要和关键词成本较高。可考虑仅对重要文档或核心段落进行深度编译。使用小型、高效的本地模型如all-MiniLM-L6-v2进行初步特征提取。缓存编译结果避免重复处理。结构化提取针对特定领域如技术API、产品手册可以定义固定的Schema如“功能”、“参数”、“示例”让LLM进行结构化信息抽取存入图数据库实现更精确的查询。3. 问答链增强查询重写在检索前使用LLM对原始用户查询进行扩展或重写使其更符合检索需求。例如将“怎么用”重写为“安装步骤、使用方法、示例代码”。重排序Reranking在初步检索出大量相关片段后使用一个更精细的交叉编码器模型如bge-reranker对结果进行重排序将最相关的片段排在最前。答案验证生成答案后可以设计一个验证步骤让模型判断答案是否能在提供的上下文中找到支持减少幻觉。4. 系统运维与监控版本化管理对知识库文档和对应的索引进行版本化管理。当文档更新时能追溯旧版本并实现索引的增量更新或回滚。效果评估建立评估体系包括检索评估命中率、MRR平均倒数排名。答案评估人工或利用LLM评估答案的准确性、相关性和完整性。溯源评估检查提供的引用是否真实支持答案。日志与审计记录所有用户查询、检索结果、生成答案及所用模型便于问题排查和效果分析。5. 安全与权限数据脱敏在构建索引前对文档中的敏感信息如手机号、身份证号进行自动识别和脱敏处理。访问控制在应用层实现基于用户角色的知识访问控制。可以在元数据中标记文档权限检索时进行过滤。这套三层架构的编译式RAG系统通过将原始知识“编译”成富含语义和结构的高级索引从根本上提升了问答的准确性、可解释性和可维护性。它不再是简单的关键词匹配和文本拼接而是迈向了一个真正的“知识系统”。你可以从本文提供的核心模块开始根据具体业务需求在每一层进行深化和扩展例如引入图数据库存储实体关系或集成Agent实现多步推理。
最近在尝试将企业文档、个人笔记等非结构化数据接入大模型时发现单纯的向量检索Naive RAG效果总是不尽如人意要么答非所问要么“幻觉”严重要么无法追溯答案来源。经过一系列项目实践和技术选型我发现“编译式RAG”是解决这些痛点的关键路径。本文将为你完整拆解一套基于“三层架构”的编译式RAG实战方案涵盖从知识库构建、智能问答到自动更新的全流程并深入对比Naive RAG、高级RAG和编译式RAG的核心差异帮你做出最适合自己场景的技术选型。1. RAG技术演进与核心概念从检索到编译在深入实战之前我们有必要厘清RAGRetrieval-Augmented Generation检索增强生成技术的演进脉络这直接决定了我们架构设计的出发点。1.1 RAG的三种范式如何选择1. Naive RAG传统/朴素RAG这是最常见的入门方案流程可概括为“索引-检索-生成”。流程将文档切块 - 向量化存入向量数据库 - 用户提问时进行语义检索 - 将检索到的文本块直接拼接到提示词中交给大模型生成答案。优点实现简单快速上手。缺点检索精度低简单的语义相似度检索容易引入无关内容。上下文碎片化检索到的文本块可能丢失原文的章节结构、逻辑关系。幻觉风险高当检索到不相关或冲突信息时模型容易产生错误答案。无法溯源答案可能综合了多个片段难以精确定位到原文出处。2. Advanced RAG高级RAG为了克服Naive RAG的缺陷Advanced RAG在检索前后增加了优化步骤。核心改进检索前对文档进行更精细的预处理如重写用户查询Query Rewriting、假设性文档嵌入HyDE。检索中尝试混合检索结合关键词与向量、多向量检索、递归检索等。检索后对检索结果进行重排序Reranking、过滤或压缩。优点显著提升了检索质量减少了无关信息干扰。缺点本质上仍是“检索-拼接”模式对于复杂、需要深度推理和多步处理的问题其能力仍有天花板。3. Compilational RAG编译式RAG这是本文的重点它借鉴了程序编译的思想将知识库视为“源代码”通过一个“编译”过程将其转化为更结构化、更易于模型理解和推理的中间表示。核心思想不是简单地把文档片段扔给模型而是先对知识库进行深度加工、索引和重构生成一个富含元数据、关系、摘要的“知识图谱”或“结构化索引”。问答时系统在这个高级索引上进行“查询”和“推理”而不仅仅是“检索”。关键特征深度索引不仅存储文本块和向量还提取实体、关系、摘要、章节大纲等。查询编译将用户自然语言问题“编译”成对结构化索引的查询指令。动态知识整合能够综合多个来源、不同粒度的信息构建连贯的答案。精准溯源答案可以与索引中的特定节点如某章节、某实体精确关联。选择建议需求简单、文档规整、问答直接- 从Naive RAG开始验证可行性。追求更高答案质量、减少幻觉、文档类型多样- 必须采用Advanced RAG。构建企业级、生产环境知识库要求高准确性、强溯源、复杂推理- 编译式RAG是更优解。1.2 为什么需要“三层架构”一个健壮的编译式RAG系统不能是“一次性”的管道而应该是一个可持续运营的“知识系统”。三层架构为此而生数据层负责原始知识的接入、存储与管理。处理不同格式PDF、Word、Markdown、网页、不同来源的数据。索引层核心负责“编译”过程。对原始数据进行解析、清洗、切分、向量化、元数据提取、关系构建形成结构化的“编译后知识索引”。这是编译式RAG区别于前两者的核心。应用层面向用户的接口。接收问题利用索引层进行智能查询与推理生成答案并附带溯源同时管理对话会话。2. 环境准备与项目初始化我们将使用Python生态中成熟的工具链来构建这个系统。2.1 技术栈选型与版本说明编程语言Python 3.9核心框架LangChain / LlamaIndex。本文示例将侧重思路部分代码使用LangChain因其生态丰富但LlamaIndex在索引深度上更有特色可根据项目偏好选择。文本嵌入模型text-embedding-ada-002(OpenAI API) 或BAAI/bge-large-zh-v1.5(本地部署)。前者效果稳定后者数据隐私性好。向量数据库Chroma轻量开发友好或 Milvus / Weaviate生产级性能强。本文用Chroma演示。大语言模型GPT-4/3.5-Turbo (OpenAI API) 或 ChatGLM3、Qwen等开源模型本地/API。其他工具PyPDF2/pdfplumber解析PDFpython-docx解析WordBeautifulSoup4解析HTMLsentence-transformers本地嵌入模型。重要提示以下版本仅为示例请根据你的实际环境调整依赖版本。2.2 创建项目并安装依赖# 创建项目目录 mkdir llm-wiki-compilational-rag cd llm-wiki-compilational-rag # 创建虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 创建 requirements.txt 文件requirements.txt内容示例langchain0.1.0 langchain-community0.0.10 langchain-openai0.0.5 chromadb0.4.22 pypdf23.0.1 pdfplumber0.10.3 python-docx1.1.0 beautifulsoup44.12.2 sentence-transformers2.2.2 openai1.6.1 tiktoken0.5.2 streamlit1.29.0 # 用于构建简单Web界面安装依赖pip install -r requirements.txt2.3 项目结构设计llm-wiki-compilational-rag/ ├── config/ # 配置文件 │ └── settings.py ├── data_layer/ # 数据层 │ ├── connectors/ # 数据连接器PDF、Web等 │ ├── storage/ # 原始文件存储 │ └── data_manager.py ├── index_layer/ # 索引层核心 │ ├── processors/ # 文档处理器 │ ├── compilers/ # 知识编译器 │ ├── vector_store/ # 向量存储操作 │ └── index_manager.py ├── app_layer/ # 应用层 │ ├── agents/ # 智能体/问答链 │ ├── web_ui/ # Web界面 │ └── api/ # API接口 ├── knowledge_base/ # 知识库目录 │ ├── raw_docs/ # 存放原始文档 │ └── compiled_index/ # 存放编译后的索引 ├── main.py # 主程序入口 └── requirements.txt3. 数据层实现多源知识接入与管理数据层是知识库的源头需要稳定可靠。3.1 实现通用文档加载器我们创建一个基类和针对不同格式的加载器。# data_layer/connectors/base_loader.py from abc import ABC, abstractmethod from typing import List, Dict, Any import hashlib class BaseDocumentLoader(ABC): 文档加载器基类 def __init__(self, file_path: str): self.file_path file_path self.metadata { source: file_path, file_hash: self._calculate_file_hash(file_path) } def _calculate_file_hash(self, file_path: str) - str: 计算文件哈希值用于内容变更检测 hash_md5 hashlib.md5() with open(file_path, rb) as f: for chunk in iter(lambda: f.read(4096), b): hash_md5.update(chunk) return hash_md5.hexdigest() abstractmethod def load(self) - List[Dict[str, Any]]: 加载文档并返回字典列表。 每个字典应包含: {content: str, metadata: dict} metadata至少包含: source, page (如果有) pass # data_layer/connectors/pdf_loader.py import pdfplumber from .base_loader import BaseDocumentLoader class PDFLoader(BaseDocumentLoader): def load(self) - List[Dict[str, Any]]: documents [] try: with pdfplumber.open(self.file_path) as pdf: for page_num, page in enumerate(pdf.pages, start1): text page.extract_text() if text.strip(): # 忽略空白页 doc_metadata self.metadata.copy() doc_metadata.update({page: page_num, type: pdf}) documents.append({ content: text, metadata: doc_metadata }) except Exception as e: print(f加载PDF文件 {self.file_path} 失败: {e}) return documents # data_layer/connectors/markdown_loader.py import markdown from bs4 import BeautifulSoup from .base_loader import BaseDocumentLoader class MarkdownLoader(BaseDocumentLoader): def load(self) - List[Dict[str, Any]]: documents [] try: with open(self.file_path, r, encodingutf-8) as f: md_content f.read() # 将Markdown转换为纯文本同时可保留结构 html markdown.markdown(md_content) soup BeautifulSoup(html, html.parser) text soup.get_text(separator\n, stripTrue) doc_metadata self.metadata.copy() doc_metadata.update({type: markdown}) # 可以进一步解析标题作为元数据 documents.append({ content: text, metadata: doc_metadata }) except Exception as e: print(f加载Markdown文件 {self.file_path} 失败: {e}) return documents3.2 实现数据管理器负责调度不同的加载器并管理原始文档的存储状态。# data_layer/data_manager.py import os from typing import List, Dict, Any import json from .connectors.pdf_loader import PDFLoader from .connectors.markdown_loader import MarkdownLoader class DataManager: 数据管理器统一处理知识源 def __init__(self, raw_docs_dir: str ./knowledge_base/raw_docs): self.raw_docs_dir raw_docs_dir os.makedirs(self.raw_docs_dir, exist_okTrue) self._loaders { .pdf: PDFLoader, .md: MarkdownLoader, .txt: None, # 可扩展 .docx: None, # 可扩展 } self._state_file os.path.join(self.raw_docs_dir, _data_state.json) self._load_state() def _load_state(self): 加载已处理文件的状态记录 if os.path.exists(self._state_file): with open(self._state_file, r, encodingutf-8) as f: self.processed_files json.load(f) else: self.processed_files {} # {file_path: file_hash} def _save_state(self): 保存状态 with open(self._state_file, w, encodingutf-8) as f: json.dump(self.processed_files, f, ensure_asciiFalse, indent2) def add_document(self, file_path: str) - List[Dict[str, Any]]: 添加单个文档到知识库。 返回加载的文档内容列表。 if not os.path.exists(file_path): raise FileNotFoundError(f文件不存在: {file_path}) # 检查文件类型 ext os.path.splitext(file_path)[1].lower() loader_class self._loaders.get(ext) if not loader_class: raise ValueError(f不支持的文件类型: {ext}) # 实例化加载器并加载 loader loader_class(file_path) loaded_docs loader.load() if loaded_docs: # 复制文件到知识库目录可选便于集中管理 import shutil target_path os.path.join(self.raw_docs_dir, os.path.basename(file_path)) shutil.copy2(file_path, target_path) # 更新状态 self.processed_files[target_path] loader.metadata[file_hash] self._save_state() print(f文档已添加: {os.path.basename(file_path)}) return loaded_docs def scan_and_load_new(self, source_dir: str) - List[Dict[str, Any]]: 扫描目录仅加载新增或修改过的文件。 返回所有新加载的文档内容。 all_new_docs [] for filename in os.listdir(source_dir): file_path os.path.join(source_dir, filename) if os.path.isfile(file_path): ext os.path.splitext(filename)[1].lower() if ext not in self._loaders: continue # 计算当前哈希 loader self._loaders[ext](file_path) current_hash loader.metadata[file_hash] # 检查是否为新文件或已修改 target_path_in_kb os.path.join(self.raw_docs_dir, filename) if target_path_in_kb not in self.processed_files or \ self.processed_files[target_path_in_kb] ! current_hash: print(f检测到新/已修改文件: {filename}) docs self.add_document(file_path) all_new_docs.extend(docs) else: print(f文件无变化跳过: {filename}) return all_new_docs4. 索引层实现编译式RAG的核心这是编译式RAG的“编译”环节我们将对原始文本进行深度处理构建结构化索引。4.1 文档处理器超越简单切分传统RAG的简单文本切分RecursiveCharacterTextSplitter会破坏文档结构。我们需要更智能的处理器。# index_layer/processors/smart_text_splitter.py from langchain.text_splitter import RecursiveCharacterTextSplitter import re class SmartTextSplitter: 智能文本分割器。 1. 尝试按章节标题分割保留结构。 2. 对于无结构的文本再按长度分割但尽量保证句子完整。 def __init__(self, chunk_size: int 1000, chunk_overlap: int 200): self.chunk_size chunk_size self.chunk_overlap chunk_overlap self.fallback_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , , , , ] ) def split_by_headings(self, text: str, metadata: dict) - List[Dict[str, Any]]: 尝试根据Markdown或常见标题格式分割 chunks [] # 匹配不同层级的标题 heading_pattern r(^#{1,6}\s.$|^第[一二三四五六七八九十][章|节|条].$) lines text.split(\n) current_chunk [] current_heading 起始 heading_level 0 for line in lines: line_stripped line.strip() # 判断是否为标题行 if re.match(heading_pattern, line_stripped): # 保存上一个块 if current_chunk: chunk_text \n.join(current_chunk) chunk_meta metadata.copy() chunk_meta.update({heading: current_heading, heading_level: heading_level}) chunks.append({content: chunk_text, metadata: chunk_meta}) # 开始新块 current_chunk [line_stripped] current_heading line_stripped # 简单判断标题级别 if line_stripped.startswith(#): heading_level line_stripped.count(#) else: heading_level 7 # 非Markdown标题给一个默认高级别 else: current_chunk.append(line) # 处理最后一个块 if current_chunk: chunk_text \n.join(current_chunk) chunk_meta metadata.copy() chunk_meta.update({heading: current_heading, heading_level: heading_level}) chunks.append({content: chunk_text, metadata: chunk_meta}) # 如果按标题没分割出内容或者块太大则用fallback分割器 final_chunks [] for chunk in chunks: if len(chunk[content]) self.chunk_size * 1.5: # 如果块太大 sub_chunks self.fallback_splitter.split_text(chunk[content]) for sub in sub_chunks: sub_meta chunk[metadata].copy() final_chunks.append({content: sub, metadata: sub_meta}) else: final_chunks.append(chunk) return final_chunks if final_chunks else self.fallback_split(text, metadata) def fallback_split(self, text: str, metadata: dict) - List[Dict[str, Any]]: 后备分割方案 texts self.fallback_splitter.split_text(text) return [{content: t, metadata: metadata.copy()} for t in texts] def split_documents(self, documents: List[Dict[str, Any]]) - List[Dict[str, Any]]: 分割文档列表 all_chunks [] for doc in documents: content doc[content] metadata doc[metadata] chunks self.split_by_headings(content, metadata) all_chunks.extend(chunks) return all_chunks4.2 知识编译器构建结构化索引这是编译式RAG的灵魂。我们将为每个文本块生成摘要、关键词并尝试提取实体和关系简化版。# index_layer/compilers/knowledge_compiler.py from langchain.chat_models import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage from typing import List, Dict, Any import json class KnowledgeCompiler: 知识编译器。 对文本块进行深度处理提取结构化信息丰富元数据。 def __init__(self, llm_api_key: str, llm_model: str gpt-3.5-turbo): self.llm ChatOpenAI( openai_api_keyllm_api_key, model_namellm_model, temperature0.1 # 低温度保证稳定性 ) def compile_chunk(self, chunk: Dict[str, Any]) - Dict[str, Any]: 编译单个文本块。 返回增强后的块包含摘要、关键词、实体等。 content chunk[content] metadata chunk[metadata] # 1. 生成摘要 summary_prompt f 请为以下文本生成一个简洁的摘要不超过100字 {content[:1500]} # 限制输入长度 summary self._call_llm(summary_prompt) # 2. 提取关键词 keyword_prompt f 请从以下文本中提取3-5个核心关键词或短语 {content[:1500]} 请以JSON列表格式返回例如[关键词1, 关键词2] keyword_response self._call_llm(keyword_prompt) try: keywords json.loads(keyword_response) except: keywords [] # 3. 提取实体简化示例生产环境可用NER模型 entity_prompt f 识别以下文本中提到的主要实体如人名、组织名、产品名、技术术语等 {content[:1500]} 请以JSON列表格式返回每个实体是一个对象包含name和type字段。 例如[{{name: Python, type: 技术}}, {{name: 张三, type: 人名}}] entity_response self._call_llm(entity_prompt) try: entities json.loads(entity_response) except: entities [] # 构建增强后的元数据 enhanced_metadata metadata.copy() enhanced_metadata.update({ summary: summary, keywords: keywords, entities: entities, compiled: True # 标记已编译 }) return { content: content, metadata: enhanced_metadata } def _call_llm(self, prompt: str) - str: 调用LLM的辅助函数 try: response self.llm([HumanMessage(contentprompt)]) return response.content.strip() except Exception as e: print(fLLM调用失败: {e}) return def compile_batch(self, chunks: List[Dict[str, Any]]) - List[Dict[str, Any]]: 批量编译生产环境应考虑异步和限流 compiled_chunks [] for i, chunk in enumerate(chunks): print(f编译进度: {i1}/{len(chunks)}) compiled self.compile_chunk(chunk) compiled_chunks.append(compiled) return compiled_chunks4.3 向量化与存储管理将编译后的知识存入向量数据库并维护元数据索引。# index_layer/vector_store/vector_manager.py import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import numpy as np from typing import List, Dict, Any import uuid class VectorStoreManager: 向量存储管理器 def __init__(self, persist_dir: str ./knowledge_base/compiled_index/chroma_db, embedding_model_name: str BAAI/bge-large-zh-v1.5): self.persist_dir persist_dir # 初始化Chroma客户端 self.client chromadb.PersistentClient( pathpersist_dir, settingsSettings(anonymized_telemetryFalse) ) # 加载本地嵌入模型或使用OpenAI Embeddings self.embedding_model SentenceTransformer(embedding_model_name) self.collection self.client.get_or_create_collection( namecompiled_knowledge, metadata{hnsw:space: cosine} # 使用余弦相似度 ) def _generate_embedding(self, text: str) - List[float]: 生成文本向量 # 这里使用本地模型也可替换为OpenAIEmbeddings embedding self.embedding_model.encode(text) return embedding.tolist() def add_documents(self, compiled_chunks: List[Dict[str, Any]]): 将编译后的文档块添加到向量库 ids [] embeddings [] metadatas [] documents [] for chunk in compiled_chunks: chunk_id str(uuid.uuid4()) content chunk[content] metadata chunk[metadata] # 生成向量 embedding self._generate_embedding(content) ids.append(chunk_id) embeddings.append(embedding) metadatas.append(metadata) documents.append(content) # 批量添加到Chroma if ids: self.collection.add( idsids, embeddingsembeddings, metadatasmetadatas, documentsdocuments ) print(f成功添加 {len(ids)} 个文档块到向量库。) def similarity_search(self, query: str, k: int 5, filter_conditions: dict None) - List[Dict[str, Any]]: 相似度搜索 query_embedding self._generate_embedding(query) results self.collection.query( query_embeddings[query_embedding], n_resultsk, wherefilter_conditions # 可进行元数据过滤如 where{source: xxx.pdf} ) # 格式化结果 formatted_results [] if results[ids]: for i in range(len(results[ids][0])): formatted_results.append({ id: results[ids][0][i], content: results[documents][0][i], metadata: results[metadatas][0][i], distance: results[distances][0][i] }) return formatted_results4.4 索引管理器串联整个流程# index_layer/index_manager.py from .processors.smart_text_splitter import SmartTextSplitter from .compilers.knowledge_compiler import KnowledgeCompiler from .vector_store.vector_manager import VectorStoreManager class IndexManager: 索引管理器协调处理、编译、存储全流程 def __init__(self, llm_api_key: str, embedding_model: str BAAI/bge-large-zh-v1.5): self.splitter SmartTextSplitter(chunk_size800, chunk_overlap150) self.compiler KnowledgeCompiler(llm_api_keyllm_api_key) self.vector_manager VectorStoreManager(embedding_model_nameembedding_model) def build_index(self, raw_documents: List[Dict[str, Any]]): 从原始文档构建完整索引 print(步骤1/4: 智能分割文档...) chunks self.splitter.split_documents(raw_documents) print(f分割为 {len(chunks)} 个块。) print(步骤2/4: 编译知识块提取摘要、关键词等...) compiled_chunks self.compiler.compile_batch(chunks) print(知识编译完成。) print(步骤3/4: 生成向量并存入数据库...) self.vector_manager.add_documents(compiled_chunks) print(索引构建完成) def search(self, query: str, k: int 5, **filters) - List[Dict[str, Any]]: 在索引中搜索 return self.vector_manager.similarity_search(query, kk, filter_conditionsfilters) def get_collection_stats(self): 获取索引统计信息 collection self.vector_manager.collection count collection.count() return {total_chunks: count}5. 应用层实现智能问答与溯源应用层负责利用构建好的“编译后索引”进行智能问答。5.1 实现带溯源的问答链# app_layer/agents/qa_agent.py from langchain.chat_models import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.runnable import RunnablePassthrough from langchain.schema.output_parser import StrOutputParser from typing import List, Dict, Any class QAAgentWithCitation: 支持溯源引用的问答智能体 def __init__(self, llm_api_key: str, index_manager): self.llm ChatOpenAI( openai_api_keyllm_api_key, model_namegpt-3.5-turbo-16k, # 使用长上下文模型 temperature0.1 ) self.index_manager index_manager # 定义提示词模板 self.prompt_template ChatPromptTemplate.from_messages([ (system, 你是一个专业的知识库助手请根据提供的参考信息回答问题。 如果参考信息不足以回答问题请明确告知“根据现有知识无法回答”。 你的回答必须严格基于参考信息不要编造信息。 在回答的最后请列出你所参考的信息来源出处。), (human, 请回答以下问题 问题{question} 参考信息 {context} 请给出答案) ]) # 构建链 self.qa_chain ( {context: self._retrieve_context, question: RunnablePassthrough()} | self.prompt_template | self.llm | StrOutputParser() ) def _retrieve_context(self, question: str) - str: 检索与问题相关的上下文并格式化 search_results self.index_manager.search(question, k4) if not search_results: return 没有找到相关参考信息。 # 格式化上下文附上来源 context_parts [] for i, res in enumerate(search_results): source res[metadata].get(source, 未知来源) page res[metadata].get(page, N/A) heading res[metadata].get(heading, 无标题) # 使用摘要如果存在或截取内容 summary res[metadata].get(summary, res[content][:300] ...) context_parts.append( f[参考片段 {i1}来源{source}页码{page}章节{heading}]\n f摘要{summary}\n f详细内容{res[content][:500]}\n f--- ) return \n.join(context_parts) def ask(self, question: str) - Dict[str, Any]: 提问并返回答案及溯源信息 # 先进行检索 search_results self.index_manager.search(question, k4) # 生成答案 answer self.qa_chain.invoke(question) # 准备溯源信息 citations [] for res in search_results: citations.append({ content_snippet: res[content][:200], source: res[metadata].get(source), page: res[metadata].get(page), heading: res[metadata].get(heading), score: 1 - res[distance] # 近似相关度分数 }) return { question: question, answer: answer, citations: citations, retrieved_count: len(search_results) }5.2 构建简单的Web界面Streamlit# app_layer/web_ui/app.py import streamlit as st import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from index_layer.index_manager import IndexManager from app_layer.agents.qa_agent import QAAgentWithCitation from data_layer.data_manager import DataManager def main(): st.set_page_config(page_titleLLM Wiki - 编译式RAG知识库, layoutwide) st.title( LLM Wiki - 编译式RAG知识库系统) # 侧边栏 - 知识库管理 with st.sidebar: st.header(知识库管理) uploaded_file st.file_uploader(上传文档PDF/Markdown, type[pdf, md]) if uploaded_file is not None: # 保存上传的文件 save_path os.path.join(./knowledge_base/raw_docs, uploaded_file.name) with open(save_path, wb) as f: f.write(uploaded_file.getbuffer()) st.success(f已上传: {uploaded_file.name}) # 添加到数据管理器并构建索引 if st.button(构建/更新索引): with st.spinner(正在处理文档并构建索引...): data_mgr DataManager() raw_docs data_mgr.add_document(save_path) if raw_docs: # 初始化索引管理器需配置你的API Key index_mgr IndexManager(llm_api_keyst.secrets[OPENAI_API_KEY]) index_mgr.build_index(raw_docs) st.success(索引构建成功) else: st.error(文档加载失败请检查文件格式。) st.divider() st.header(系统状态) # 这里可以显示索引统计等信息 # 主界面 - 问答 st.header(智能问答) question st.text_input(请输入你的问题, placeholder例如RAG的核心思想是什么) if question: with st.spinner(正在检索并生成答案...): # 初始化问答智能体需提前构建好索引 index_mgr IndexManager(llm_api_keyst.secrets[OPENAI_API_KEY]) qa_agent QAAgentWithCitation( llm_api_keyst.secrets[OPENAI_API_KEY], index_managerindex_mgr ) result qa_agent.ask(question) # 显示答案 st.subheader(答案) st.write(result[answer]) # 显示溯源 st.subheader(参考来源) for i, cite in enumerate(result[citations]): with st.expander(f来源 {i1}: {cite[source]} - {cite[heading]}): st.caption(f相关度: {cite[score]:.2%}) st.write(cite[content_snippet]) st.text(f页码: {cite[page]}) if __name__ __main__: # 需要先设置OPENAI_API_KEY环境变量或在secrets.toml中配置 main()6. 自动知识更新与维护一个实用的知识库必须支持内容更新。我们可以在数据层的基础上增加定时扫描和增量索引构建。# main.py 中的更新逻辑示例 import schedule import time from data_layer.data_manager import DataManager from index_layer.index_manager import IndexManager def scheduled_update(): 定时任务扫描目录更新索引 print(开始执行定时知识库更新检查...) data_mgr DataManager() index_mgr IndexManager(llm_api_keyyour-api-key) # 假设我们监控一个源目录 source_dir ./source_docs new_docs data_mgr.scan_and_load_new(source_dir) if new_docs: print(f发现 {len(new_docs)} 个新文档/更新开始增量构建索引...) # 注意这里简化处理实际增量更新需要更精细的管理如删除旧版本块 index_mgr.build_index(new_docs) print(增量索引更新完成。) else: print(未发现变更。) # 每天凌晨2点执行更新 schedule.every().day.at(02:00).do(scheduled_update) if __name__ __main__: print(知识库更新服务启动...) while True: schedule.run_pending() time.sleep(60)7. 常见问题与排查思路在构建和运行编译式RAG系统时你可能会遇到以下典型问题。问题现象可能原因排查思路与解决方案文档加载失败或乱码1. 文件格式不支持。2. 文件编码问题。3. 文件损坏或受密码保护。1. 检查文件扩展名实现对应的加载器。2. 指定正确的编码如utf-8,gbk。3. 尝试用其他软件打开文件确认是否正常。向量检索结果不相关1. 文本分割不合理上下文碎片化。2. 嵌入模型不匹配如中英文。3. 检索参数k值不合适。1. 优化SmartTextSplitter尝试按标题、段落分割。2. 为中文文本选择BAAI/bge-*系列模型。3. 调整k值如从5调到10或启用重排序Reranker。LLM回答出现“幻觉”1. 检索到的上下文不足或无关。2. 提示词Prompt未强制要求基于上下文。3. 模型温度Temperature设置过高。1. 检查检索结果质量优化索引。2. 在系统提示词中明确指令“严格基于以下参考信息”。3. 将temperature调低如0.1。知识编译摘要/提取速度慢1. 串行调用LLM。2. 模型响应慢或网络延迟。3. 文本块过大或过多。1. 实现异步批量处理asyncio。2. 考虑使用更快的模型如gpt-3.5-turbo或本地轻量模型进行初步提取。3. 调整chunk_size避免单个块过长。溯源信息不准确1. 元数据如页码、标题在分割时丢失。2. 答案综合了多个片段难以对应单一来源。1. 确保分割器正确传递并保留原始元数据。2. 在答案中声明“综合以下信息”并列出所有相关片段来源。系统内存/磁盘占用过高1. 向量数据库未做持久化或分片。2. 原始文档和索引未定期清理。1. 对于大规模知识库使用Milvus、Weaviate等支持集群和磁盘索引的数据库。2. 制定数据归档策略移除过时文档及其索引。8. 最佳实践与工程建议将编译式RAG投入生产环境需要关注以下工程细节。1. 索引构建优化分层索引不仅存储文本块向量还可以构建文档级摘要向量、章节标题向量实现从粗到细的多级检索。混合检索结合语义向量检索和关键词BM25检索提升召回率。可以使用langchain.retrievers中的EnsembleRetriever。元数据过滤为每个文档块添加丰富的元数据如文档类型、创建日期、部门、权限等级在检索时支持高效的过滤。2. 知识编译策略成本与质量权衡为每个文本块调用LLM生成摘要和关键词成本较高。可考虑仅对重要文档或核心段落进行深度编译。使用小型、高效的本地模型如all-MiniLM-L6-v2进行初步特征提取。缓存编译结果避免重复处理。结构化提取针对特定领域如技术API、产品手册可以定义固定的Schema如“功能”、“参数”、“示例”让LLM进行结构化信息抽取存入图数据库实现更精确的查询。3. 问答链增强查询重写在检索前使用LLM对原始用户查询进行扩展或重写使其更符合检索需求。例如将“怎么用”重写为“安装步骤、使用方法、示例代码”。重排序Reranking在初步检索出大量相关片段后使用一个更精细的交叉编码器模型如bge-reranker对结果进行重排序将最相关的片段排在最前。答案验证生成答案后可以设计一个验证步骤让模型判断答案是否能在提供的上下文中找到支持减少幻觉。4. 系统运维与监控版本化管理对知识库文档和对应的索引进行版本化管理。当文档更新时能追溯旧版本并实现索引的增量更新或回滚。效果评估建立评估体系包括检索评估命中率、MRR平均倒数排名。答案评估人工或利用LLM评估答案的准确性、相关性和完整性。溯源评估检查提供的引用是否真实支持答案。日志与审计记录所有用户查询、检索结果、生成答案及所用模型便于问题排查和效果分析。5. 安全与权限数据脱敏在构建索引前对文档中的敏感信息如手机号、身份证号进行自动识别和脱敏处理。访问控制在应用层实现基于用户角色的知识访问控制。可以在元数据中标记文档权限检索时进行过滤。这套三层架构的编译式RAG系统通过将原始知识“编译”成富含语义和结构的高级索引从根本上提升了问答的准确性、可解释性和可维护性。它不再是简单的关键词匹配和文本拼接而是迈向了一个真正的“知识系统”。你可以从本文提供的核心模块开始根据具体业务需求在每一层进行深化和扩展例如引入图数据库存储实体关系或集成Agent实现多步推理。
避坑指南)
