企业官网建设流程全解析

摘要：本文围绕 RAG 系统的文档处理接口展开，重点讲解上传文档接口的前后端实现。后端部分详细说明了 FastAPI 路由中 upload_docs 接口如何接收文件、调用 RAGService 完成文档摄取，并逐层剖析 RAGService、RAGApplication 与 DocumentIngestionPipeline 的初始化流程，包括模型配置、句子分割器、ChromaDB 向量存储以及文档分片、建索引和持久化的完整链路；前端部分则介绍了 file 类型 input 标签绑定 change 事件、通过 handleFileUpload 函数获取文件并发送给后端的实现方式。

内容参考于：图灵AI大模型全栈

文档处理接口

上传文档接口的前缀

处理上传文档的接口和代码（前端和后端代码）

代码说明：

# 导入类型提示模块，用于声明变量类型 from typing import List 从 FastAPI 导入所需组件： APIRouter：用于创建路由对象 Depends：依赖注入工具，用于声明路由处理函数依赖的其他函数/服务 File：用于声明请求中必须包含文件参数 HTTPException：用于抛出 HTTP 错误 UploadFile：表示上传的文件对象 from fastapi import APIRouter, Depends, File, HTTPException, UploadFile 导入 RAG 服务类以及获取该服务实例的依赖函数 from app.rag_service import RAGService, get_rag_service 导入用户模型以及获取当前活跃用户的依赖函数（通常用于鉴权） from app.routers.users import User, get_current_active_user 导入已经定义好的 Pydantic 响应模型，用于格式化接口返回数据 from app.schemas import CommonResponse, DocsListResponse, UploadResponse 创建一个 APIRouter 实例，所有在该实例上定义的路由都会挂载到应用中 router = APIRouter() 定义 POST 请求处理函数，路径为 /upload，响应模型为 UploadResponse @router.post("/upload", response_model=UploadResponse) async def upload_docs( # files 参数：接收多个上传文件，File(...) 表示该参数是必需的，不能省略 files: List[UploadFile] = File(...), # current_user 参数：通过依赖注入获取当前已认证的用户，用于身份验证 current_user: User = Depends(get_current_active_user), # svc 参数：通过依赖注入获取 RAG 服务实例，负责文档处理 svc: RAGService = Depends(get_rag_service), ) -> UploadResponse: """上传文档并交给 LlamaIndex 摄取流水线处理。""" try: # 异步读取所有上传文件的内容，得到字节列表 contents = [await file.read() for file in files] # 提取所有文件的文件名，组成字符串列表 filenames = [file.filename for file in files] # 调用 RAG 服务的 upload_and_process_files 方法， # 传入文件内容和文件名，返回状态描述文本和已处理的文件信息 status_text, processed = svc.upload_and_process_files(contents, filenames) # 根据返回的状态文本判断本次操作结果 # 如果状态文本中包含“失败”或 error 等关键词，设置状态为 failed if any(keyword in status_text for keyword in ["失败", "error", "Error", "ERROR"]): status = "failed" # 如果包含“成功”“完成”或 success 等关键词，设置状态为 success elif any(keyword in status_text for keyword in ["成功", "完成", "success", "done"]): status = "success" # 否则视为仍在处理中 else: status = "processing" # 构造并返回 UploadResponse 响应对象 return UploadResponse(status=status, message=status_text, processed_files=processed) # 捕获所有异常，避免程序崩溃，并将异常信息通过 HTTP 500 返回给客户端 except Exception as exc: # 抛出 HTTP 500 错误，并在 detail 中附带异常的具体信息 raise HTTPException(status_code=500, detail=str(exc)) 定义另一个 POST 请求处理函数，路径为 /reset，响应模型为 CommonResponse @router.post("/reset", response_model=CommonResponse) async def reset_system( # 同样需要用户认证 current_user: User = Depends(get_current_active_user), # 同样需要 RAG 服务实例 svc: RAGService = Depends(get_rag_service), ) -> CommonResponse: """清空当前进程内的聊天记忆和缓存检索器。""" # 调用 RAG 服务的 reset_system 方法，重置系统状态（如清空对话记忆和缓存） svc.reset_system() # 返回一个表示操作成功的通用响应 return CommonResponse(status="success", message="系统已重置")

它通过下图红框的函数处理的文件，这个svc的值

安装CTRL鼠标左键单击下图红框

创建了一个 RAGService 类型的实例

RAGService的构造方法里，创建了一个 RAGApplication 类型

RAGApplication 构造方法里如下图红框

首先是 DocumentIngestionPipeline() 它是用来创建捨取管道的，如下图红框，它的构造方法里

这里给默认是一个空索引

初始化模型

初始分割器

创建存储管理器

如下图存储管理器的内容，后期就可以修改下图框框里的代码更换存储位置

如下图 RAGApplication 的构造方法描述，除了 DocumentIngestionPipeline 没啥好写的

elf.chat_store = self._load_chat_store()也是用的下图红框的代码

现在就初始化完成了，就可以调用 svc.upload_and_process_files(contents, filenames) 了，它又调用了下图红框代码

然后又调用了ingest_documents，如下图红框代码

最终在ingest_documents中，对文档进行了分片获取节点、构建索引、向量、存储，如下图红框

rag_service.py文件，RAGService代码的说明

import os import shutil import tempfile from typing import Dict, List, Optional, Tuple 导入核心 RAG 应用类 from core.application import RAGApplication 导入日志设置函数 from utils.logger import setup_logger 初始化本模块的日志记录器 logger = setup_logger(name) class RAGService: """HTTP 层适配器：处理上传临时文件，并调用核心 RAG 应用。""" def __init__(self) -> None: # 实例化核心 RAG 应用 self.app = RAGApplication() def upload_and_process_files(self, files: List[bytes], filenames: List[str]) -> Tuple[str, List[str]]: """ 接收上传的文件内容（字节列表）和文件名列表， 将它们临时写入磁盘，然后调用核心应用的文档处理功能。 返回处理结果信息和处理过的文件名列表。 """ # 创建临时目录，前缀为 "rag_upload_" tmpdir = tempfile.mkdtemp(prefix="rag_upload_") paths = [] # 存储临时文件的完整路径 try: # 将每个文件内容写入临时目录中的文件 for name, content in zip(filenames, files): # 构建完整路径 path = os.path.join(tmpdir, name) # 以二进制写模式打开文件，写入内容 with open(path, "wb") as file: file.write(content) # 记录路径 paths.append(path) # 记录上传的文件路径 logger.info("上传的文档: %s", paths) # 调用核心应用的 upload_and_process_files 处理这些文件，并返回结果 # 注意：这里返回的第二个值是原始文件名列表，而不是实际处理的文件路径 return self.app.upload_and_process_files(paths), filenames finally: # 无论是否发生异常，都删除临时目录及其内容 shutil.rmtree(tmpdir, ignore_errors=True) async def query_stream( self, session_id: str, query: str, model: str, knowledge_bool: bool, temperature: float, max_tokens: int, ): """ 异步流式查询接口。 首先更新模型配置，然后将查询转发给核心应用的流式处理方法，并逐块产出结果。 """ # 更新模型配置（用户指定的模型、温度、最大 token 数） self.app.update_model_config(model, temperature, max_tokens) # 异步迭代核心应用的流式输出 async for chunk in self.app.query_documents_stream( session_id=session_id, query=query, knowledge_bool=knowledge_bool, ): # 将每个块 yield 给调用者 yield chunk def get_session(self, session_id: str): """获取指定会话的历史记录。""" return self.app.get_session_history(session_id) def clear_session(self, session_id: str) -> None: """清空指定会话的历史和记忆。""" self.app.clear_session(session_id) def reset_system(self) -> None: """重置整个系统（清空所有会话和索引等）。""" # 注意：RAGApplication 类中并未定义 reset 方法，此处可能存在遗漏或需要补全 self.app.reset() 全局 RAGService 实例（单例） _rag_service: Optional[RAGService] = None def get_rag_service() -> RAGService: """获取全局 RAGService 单例实例，若不存在则创建。""" global _rag_service if _rag_service is None: _rag_service = RAGService() return _rag_service

application.py的代码，RAGApplication的代码就在里面

import traceback from pathlib import Path from typing import Any, AsyncGenerator, Dict, List, Optional import chromadb from llama_index.core import ( Settings, SimpleDirectoryReader, StorageContext, VectorStoreIndex, load_index_from_storage, ) from llama_index.core.chat_engine.context import ContextChatEngine from llama_index.core.chat_engine.simple import SimpleChatEngine from llama_index.core.extractors import TitleExtractor from llama_index.core.ingestion import IngestionPipeline from llama_index.core.memory import ChatMemoryBuffer from llama_index.core.node_parser import SentenceSplitter from llama_index.core.postprocessor import SentenceTransformerRerank from llama_index.core.retrievers import QueryFusionRetriever, VectorIndexRetriever from llama_index.core.storage.docstore import SimpleDocumentStore from llama_index.core.storage.chat_store import SimpleChatStore from llama_index.core.storage.index_store import SimpleIndexStore from llama_index.embeddings.huggingface import HuggingFaceEmbedding from llama_index.llms.dashscope import DashScope from llama_index.retrievers.bm25 import BM25Retriever from llama_index.vector_stores.chroma import ChromaVectorStore from config.settings import Settings as AppSettings from utils.logger import setup_logger from llama_index.llms.openai_like import OpenAILike 创建本模块的日志记录器 logger = setup_logger(name) class DocumentIngestionPipeline: """文档摄取：把读取、切分、向量化和持久化交给 LlamaIndex。""" def __init__(self) -> None: # 索引对象，初始为 None，后续加载或创建 self.index: Optional[VectorStoreIndex] = None # 1. 配置基础模型（LLM 和 Embedding） self._setup_models() # 2. 创建摄取管道（包含句子分割器和向量化器） self._create_pipeline() # 3. 初始化存储上下文（文档存储、索引存储、向量存储） self._initialize_storage() def update_model_config(self, model_name: str, temperature: float, max_tokens: int) -> None: """ 动态更新 LLM 模型配置，使用 OpenAILike 接口连接自定义模型。 通常用于用户在运行时切换模型。 """ Settings.llm = OpenAILike( model=model_name, # 注意参数名是 model 而不是 model_name api_key=AppSettings.API_KEY, api_base=AppSettings.API_BASE_URL, temperature=temperature, max_tokens=max_tokens, context_window=32768, # 上下文窗口大小，需根据实际模型调整 is_chat_model=True, # 标记为聊天模型 ) logger.info("模型已更新: model=%s, temperature=%s", model_name, temperature) def ingest_documents(self, file_paths: List[str]) -> str: """ 接收文件路径列表，读取文档并处理成索引节点，然后更新索引。 返回处理结果的状态信息字符串。 """ # 过滤出存在的文件路径 valid_paths = [str(Path(path)) for path in file_paths if Path(path).exists()] if not valid_paths: return "没有找到有效的文档" try: # 使用 SimpleDirectoryReader 读取指定文件，返回 Document 对象列表 documents = SimpleDirectoryReader(input_files=valid_paths).load_data() # 通过管道处理文档，生成节点列表 nodes = self.pipeline.run(documents=documents) if not nodes: return "没有生成有效的文档节点" # 如果索引尚未加载，尝试从存储加载 if self.index is None: try: self.load_index() except RuntimeError: # 加载失败说明还没有索引，忽略异常继续创建新索引 pass if self.index is None: # 索引仍为空，说明是第一次上传文档 # 清空向量数据库（防止旧数据干扰） self._clear_vector_store() # 使用节点和存储上下文创建新的 VectorStoreIndex self.index = VectorStoreIndex( nodes, storage_context=self.storage_context, store_nodes_override=True, ) else: # 索引已存在，将新节点插入现有索引 self.index.insert_nodes(nodes) # 持久化存储（将 docstore、index_store 等写入磁盘） self._persist_storage() result = f"成功摄取了 {len(valid_paths)} 个文档，生成了 {len(nodes)} 个节点" logger.info(result) return result except Exception as exc: error_msg = f"文档摄取失败: {exc}" logger.error(error_msg) return error_msg def load_index(self) -> None: """从持久化存储加载已有索引，若不存在则抛出 RuntimeError。""" logger.info("从持久化存储加载知识库索引") try: # 使用存储上下文加载索引 self.index = load_index_from_storage( self.storage_context, store_nodes_override=True, ) except ValueError as exc: # 捕获“存储上下文中没有索引”的特定异常并转换为 RuntimeError if "No index in storage context" in str(exc): raise RuntimeError("知识库索引未初始化，请先上传文档") from exc raise def _setup_models(self) -> None: """初始化全局 LLM 和 Embedding 模型配置。""" # 配置 LLM 使用 DashScope（阿里云百炼平台） Settings.llm = DashScope( api_key=AppSettings.API_KEY, api_base=AppSettings.API_BASE_URL, model_name=AppSettings.MODEL, temperature=AppSettings.TEMPERATURE, ) # 配置 Embedding 使用 HuggingFace 模型（本地加载） Settings.embed_model = HuggingFaceEmbedding(model_name=AppSettings.EMBEDDING_MODEL_PATH) def _create_pipeline(self) -> None: """创建 IngestionPipeline，定义文档转换步骤。""" self.pipeline = IngestionPipeline( transformations=[ # 句子分割器：将文档切分为固定大小的块，并带有重叠 SentenceSplitter( chunk_size=AppSettings.CHUNK_SIZE, chunk_overlap=AppSettings.CHUNK_OVERLAP, ), # 可选的标题提取器（此处注释掉） # TitleExtractor(nodes=AppSettings.TITLE_EXTRACTOR_NODES), # 嵌入模型：将文本块转换为向量 Settings.embed_model, ] ) def _initialize_storage(self) -> None: """初始化存储上下文：文档存储、索引存储、向量存储（ChromaDB）。""" # 确保持久化目录存在 Path(AppSettings.DEFAULT_PERSIST_DIR).mkdir(parents=True, exist_ok=True) # 创建 ChromaDB 持久化客户端并获取或创建集合 chroma_client = chromadb.PersistentClient(AppSettings.CHROMA_PERSIST_DIR) self.chroma_collection = chroma_client.get_or_create_collection(AppSettings.CHROMA_COLLECTION) vector_store = ChromaVectorStore(chroma_collection=self.chroma_collection) # 构建 StorageContext，包含文档存储、索引存储和向量存储 self.storage_context = StorageContext.from_defaults( docstore=self._load_docstore(), index_store=self._load_index_store(), vector_store=vector_store, ) def _load_docstore(self) -> SimpleDocumentStore: """从磁盘加载文档存储，若不存在则创建空存储。""" docstore_path = Path(AppSettings.DOCSTORE_PATH) if docstore_path.exists(): return SimpleDocumentStore.from_persist_path(str(docstore_path)) return SimpleDocumentStore() def _load_index_store(self) -> SimpleIndexStore: """从磁盘加载索引存储，若不存在则创建空存储。""" index_store_path = Path(AppSettings.INDEX_STORE_PATH) if index_store_path.exists(): return SimpleIndexStore.from_persist_path(str(index_store_path)) return SimpleIndexStore() def _persist_storage(self) -> None: """将存储上下文中的所有数据持久化到指定目录。""" self.storage_context.persist(persist_dir=AppSettings.DEFAULT_PERSIST_DIR) def _clear_vector_store(self) -> None: """清空 ChromaDB 集合中的所有向量数据。""" ids = self.chroma_collection.get().get("ids") or [] if ids: self.chroma_collection.delete(ids=ids) class RAGApplication: """RAG 应用入口：用 LlamaIndex 管理摄取、聊天、记忆和知识库问答。""" def __init__(self) -> None: # 创建文档摄取管道实例 self.ingestion_pipeline = DocumentIngestionPipeline() # 聊天记录存储路径 self.chat_store_path = Path(AppSettings.CHAT_STORE_PATH) # 确保存储目录存在 self.chat_store_path.parent.mkdir(parents=True, exist_ok=True) # 加载聊天记录存储 self.chat_store = self._load_chat_store() # 保存会话 ID 到 ChatMemoryBuffer 的映射 self.memories: Dict[str, ChatMemoryBuffer] = {} # 多路检索器（混合检索），延迟初始化 self.rag_retriever: Optional[QueryFusionRetriever] = None # 重排序器，延迟初始化 self.reranker: Optional[SentenceTransformerRerank] = None def update_model_config(self, model_name: str, temperature: float, max_tokens: int) -> None: """转发模型配置更新到摄取管道。也就是修改使用的模型信息""" self.ingestion_pipeline.update_model_config(model_name, temperature, max_tokens) def upload_and_process_files(self, file_paths: List[str]) -> str: """处理上传文件：调用摄取管道进行文档处理。""" if not file_paths: return "请上传至少一个文件" result = self.ingestion_pipeline.ingest_documents(file_paths) # 上传新文档后，需要重建检索器（因为索引可能已变化） self.rag_retriever = None return result async def query_documents_stream( self, session_id: str, query: str, knowledge_bool: bool, ) -> AsyncGenerator[Dict[str, Any], None]: """ 流式处理用户查询。根据 knowledge_bool 决定是否使用 RAG 检索。 生成器会逐步产出文本块、来源文档和完成信号。 """ try: # 获取聊天引擎（根据是否使用知识库） chat_engine = self._get_chat_engine(session_id, knowledge_bool) # 调用流式聊天接口，返回响应对象 stream_response = await chat_engine.astream_chat(query) # 迭代响应中的令牌（异步生成器） async for token in stream_response.async_response_gen(): if token: # 令牌可能是字符串或对象，统一提取文本内容 if isinstance(token, str): yield {"type": "text", "finished": False, "content": token} else: # 若为块对象，则提取其 text 属性 text = getattr(token, "text", str(token)) yield {"type": "text", "finished": False, "content": text} # 获取检索到的源节点（仅 RAG 引擎有） sources = self._source_nodes_to_dicts(getattr(stream_response, "source_nodes", [])) if sources: # 如果有来源，则产出 sources 类型的消息 yield { "type": "sources", "finished": False, "content": sources, "sources_data": sources, } # 持久化聊天记录 self._persist_chat_store() # 最后产出完成信号，附带最终响应文本 yield {"type": "complete", "finished": True, "content": stream_response.response} except Exception as exc: # 捕获异常，输出完整堆栈信息 error_trace = traceback.format_exc() error_msg = f"查询失败: {exc}\n详细堆栈:\n{error_trace}" print(error_msg) logger.error(error_msg) yield {"type": "error", "content": error_msg, "finished": True} def get_session_history(self, session_id: str) -> List[Dict[str, Any]]: """获取指定会话的历史消息列表。""" # 从记忆缓冲区获取消息 messages = self._get_memory(session_id).get() # 转换为可序列化的字典格式 return [ { "role": message.role.value if hasattr(message.role, "value") else str(message.role), "content": message.content or "", "sources": [], # 历史消息不包含来源信息 } for message in messages ] def clear_session(self, session_id: str) -> None: """清空指定会话的记忆和聊天记录。""" if session_id in self.memories: # 重置内存中的记忆缓冲区 self.memories[session_id].reset() # 从聊天存储中删除该会话的所有消息 self.chat_store.delete_messages(session_id) # 持久化聊天存储 self._persist_chat_store() def _get_memory(self, session_id: str) -> ChatMemoryBuffer: """获取或创建指定会话的记忆缓冲区。""" if session_id not in self.memories: # 若不存在，则基于当前 LLM 和聊天存储创建新的缓冲区 self.memories[session_id] = ChatMemoryBuffer.from_defaults( llm=Settings.llm, chat_store=self.chat_store, chat_store_key=session_id, ) return self.memories[session_id] def _load_chat_store(self) -> SimpleChatStore: """从磁盘加载聊天记录存储，若不存在则创建空存储。""" if self.chat_store_path.exists(): return SimpleChatStore.from_persist_path(str(self.chat_store_path)) return SimpleChatStore() def _persist_chat_store(self) -> None: """将聊天记录存储持久化到磁盘。""" self.chat_store.persist(str(self.chat_store_path)) def _get_chat_engine(self, session_id: str, knowledge_bool: bool): """ 根据是否使用知识库返回不同的聊天引擎。 - 若 knowledge_bool 为 True，则返回 ContextChatEngine（带检索器） - 否则返回 SimpleChatEngine（仅基于 LLM 和记忆） """ memory = self._get_memory(session_id) if knowledge_bool: return ContextChatEngine.from_defaults( retriever=self._get_rag_retriever(), memory=memory, llm=Settings.llm, node_postprocessors=[self._get_reranker()], # 使用重排序器优化结果 ) return SimpleChatEngine.from_defaults(memory=memory, llm=Settings.llm) def _get_rag_retriever(self) -> QueryFusionRetriever: """获取混合检索器（向量检索 + BM25），若不存在则创建。""" # 确保索引已加载 self._ensure_index_loaded() index = self.ingestion_pipeline.index if index is None: raise RuntimeError("知识库索引未初始化，请先上传文档") # 如果混合检索器尚未创建，则现在创建 if self.rag_retriever is None: # 创建向量检索器 vector_retriever = VectorIndexRetriever( index=index, similarity_top_k=AppSettings.SIMILARITY_TOP_K, ) # 获取索引中的所有节点，用于 BM25 检索器 bm25_nodes = [ node for node in index.docstore.docs.values() if node.get_content() ] if not bm25_nodes: raise RuntimeError("本地文档存储为空，请重新上传文档") # 创建 BM25 检索器 bm25_retriever = BM25Retriever.from_defaults( nodes=bm25_nodes, similarity_top_k=AppSettings.SIMILARITY_TOP_K, ) # 创建 QueryFusionRetriever，融合向量和 BM25 的检索结果 self.rag_retriever = QueryFusionRetriever( [vector_retriever, bm25_retriever], num_queries=1, # 融合查询数（这里为1，相当于简单融合） use_async=True, ) return self.rag_retriever def _get_reranker(self) -> SentenceTransformerRerank: """获取重排序器，用于对检索结果重新排序。""" if self.reranker is None: # 使用 SentenceTransformer 模型进行重排序 self.reranker = SentenceTransformerRerank( model=AppSettings.RERANK_MODEL_PATH, top_n=AppSettings.RERANK_TOP_K, ) return self.reranker def _ensure_index_loaded(self) -> None: """确保索引已加载，若没有则尝试从存储加载。""" if self.ingestion_pipeline.index is not None: return # 加载索引 self.ingestion_pipeline.load_index() # 加载后再次检查 if self.ingestion_pipeline.index is None: raise RuntimeError("知识库索引未初始化，请先上传文档") @staticmethod def _source_nodes_to_dicts(source_nodes) -> List[Dict[str, Any]]: """将 LlamaIndex 的 NodeWithScore 列表转换为可序列化的字典列表。""" return [ { "content": node.node.get_content(), # 节点文本内容 "score": float(node.score) if node.score is not None else None, # 相关度分数 "metadata": node.node.metadata or {}, # 节点元数据 } for node in source_nodes or [] ]

前端代码，如下图红框，它就是一个file类型的input标签，它绑定了change事件，也就是当内容发生变化了就会执行handleFileUpload函数

如下图红框，handleFileUpload函数中就是获取input里面的文件，然后发送给后端