基于RAG与LLM的智能文档处理系统：从原理到工程实践

1. 项目概述：一个为文档处理而生的智能“机械爪”

如果你经常和一堆杂乱的文档打交道，比如PDF、Word、Excel，需要从中快速提取关键信息、总结内容，或者根据文档内容回答特定问题，那你一定体会过手动翻阅和查找的低效与痛苦。tbdavid2019/openclaw-docs-skill这个项目，就像是为你的电脑装上了一个智能的“机械爪”，它能自动抓取、理解并处理你指定的文档，帮你完成那些重复、繁琐的信息处理工作。

这个项目本质上是一个“文档处理技能”的实现。它不是一个独立的桌面软件，而更像是一个可以被集成到其他自动化流程或智能助手（比如一些RPA工具或AI Agent框架）中的核心能力模块。你可以把它想象成一个专门处理文档的“插件”或“技能包”。当你把一堆文档“喂”给它，并告诉它你想干什么（比如“找出所有合同里的甲方名称”或“总结这份50页报告的核心观点”），它就能调用内置的AI能力去阅读、分析，并给出结构化的答案。

它的核心价值在于将非结构化的文档内容，转化为结构化的、可操作的数据或知识。无论是个人用于快速阅读文献、整理读书笔记，还是企业用于自动化处理合同、票据、报告，这个“技能”都能显著提升信息处理的效率和准确性。接下来，我们就深入拆解这个“机械爪”是如何被设计和制造出来的。

1.1 核心需求与场景解析

为什么我们需要这样一个工具？其需求源于几个普遍的业务痛点：

1. 信息检索与提取的低效：面对上百份投标文件，需要快速找出所有涉及“付款周期”的条款；从一堆年度财报PDF里，提取每家公司的“净利润”和“营业收入”数据。传统方式是肉眼扫描或简单的关键词搜索，后者在格式复杂、表述多样的文档面前显得力不从心。

2. 内容理解与总结的自动化：每天需要阅读大量行业研报，人工总结耗时耗力。或者，客服部门需要从历史工单文档中，自动归纳常见问题类型。这需要工具不仅能“看到”文字，还要能“理解”文字的含义和上下文。

3. 多格式文档的统一处理入口：数据来源五花八门，可能是扫描的PDF图片、可编辑的Word、结构化的Excel，甚至是网页截图。一个理想的工具需要屏蔽这些格式差异，提供一个统一的“文本输入”接口。

4. 与企业现有流程的集成：自动化处理不应该是一个信息孤岛。这个“技能”需要能够被轻松地嵌入到企业现有的OA系统、知识库平台或RPA机器人流程中，作为其中一个智能环节来调用。

因此，openclaw-docs-skill瞄准的正是这些场景。它不适合替代专业的文档管理系统，而是专注于“文档内容的理解与提取”这一细分但高频的自动化需求。它的用户可能是开发者（将其集成到自己的产品中）、数据分析师、法务、财务以及任何需要与大量文档内容打交道的知识工作者。

1.2 技术栈选型与架构总览

要实现上述能力，项目背后必然依赖一套稳健的技术组合。虽然我们无法看到其源码，但根据其项目名和描述，可以推断出它大概率采用了当前处理此类问题的“最佳实践”技术栈。

核心架构分层：

1. 文档接入与解析层：

   • 目标：处理多种格式的原始文档，将其转换为纯文本或带有基础格式（如段落、标题、表格）的结构化数据。
   • 可能的技术选型：
     • PDF处理：PyPDF2、pdfplumber或PyMuPDF。对于扫描件，则需要集成OCR引擎，如Tesseract或云服务（阿里云、百度云OCR的SDK）。pdfplumber在提取表格数据方面优势明显。
     • Office文档处理：python-pptx处理PPT，openpyxl或pandas处理Excel，对于Word，python-docx是标准选择。
     • 通用文本/HTML：标准库足以应对。
   • 设计考量：这一层的关键是鲁棒性。不同的PDF生成方式（扫描、文字型、加密）需要不同的处理策略，必须要有良好的异常处理和回退机制。

2. 文本处理与向量化层：

   • 目标：对解析出的文本进行清洗（去噪、分段）、关键信息增强，并转换为计算机可以高效“理解”和“比对”的格式——即向量（Embedding）。
   • 可能的技术选型：
     • 文本分割：使用langchain的RecursiveCharacterTextSplitter或基于标记（Token）的分割器，将长文本切分成语义连贯的片段（Chunk），以适应大语言模型的上下文长度限制。
     • 向量模型：选用开源的嵌入模型，如text2vec、BGE（BAAI General Embedding）系列，或调用OpenAI、Cohere的Embedding API。选择时需权衡效果、速度和成本。
     • 向量数据库：为了快速检索相关文本片段，必须引入向量数据库。ChromaDB（轻量、易用）、Milvus（高性能、可扩展）或Qdrant是常见选择。这一步是将文档知识“存入大脑”的关键。

3. 智能问答与任务执行层：

   • 目标：接收用户查询，从向量库中召回最相关的文档片段，并驱动大语言模型生成精准答案或执行总结、翻译等任务。
   • 可能的技术选型：
     • 大语言模型（LLM）集成：通过langchain、llama_index等框架方便地接入各类LLM。可能是本地部署的Llama 3、Qwen系列，也可能是调用GPT-4、Claude或国内大厂的API。这是项目的“大脑”。
     • 提示词工程：精心设计系统提示词（System Prompt），引导模型扮演“专业文档分析师”的角色，严格按照提供的上下文回答问题，避免幻觉（Hallucination）。
     • 任务链设计：复杂任务可能被分解为“检索 -> 摘要 -> 格式化输出”等多个步骤，通过langchain的LCEL（LangChain Expression Language）或自定义链（Chain）来编排。

4. 技能封装与接口层：

   • 目标：将以上复杂能力包装成一个简单的、可调用的“技能”。这通常通过一个Web API（如FastAPI、Flask）来实现，提供诸如/process_document、/ask之类的端点。
   • 设计考量：接口设计要简洁明了，输入输出定义清晰（如支持文件上传、文本输入、任务类型参数）。同时，需要考虑异步处理、任务队列（Celery）以应对耗时较长的文档解析任务。

这个分层架构确保了各司其职，也便于后续替换或升级某一层的组件（比如换用更强的嵌入模型或LLM）。

2. 核心模块深度拆解与实现要点

理解了整体架构，我们再来深入看看几个核心模块在实现时有哪些“魔鬼细节”。

2.1 文档解析：从乱码到清晰文本的攻坚战

文档解析是第一步，也是坑最多的一步。处理不当，后续所有高级分析都是空中楼阁。

PDF解析的实战策略：对于文字型PDF，直接使用pdfplumber提取文本和表格通常效果很好。但你必须处理以下问题：

import pdfplumber def extract_text_from_pdf(pdf_path): text_chunks = [] with pdfplumber.open(pdf_path) as pdf: for page in pdf.pages: # 提取文本 page_text = page.extract_text() if page_text: # 简单的段落合并（根据换行符判断） lines = page_text.split('\n') cleaned_text = ' '.join([line.strip() for line in lines if line.strip()]) text_chunks.append(cleaned_text) # 尝试提取表格 tables = page.extract_tables() for table in tables: # 将表格转换为Markdown或结构化字符串 table_str = '\n'.join(['|'.join(row) for row in table]) text_chunks.append(f"[表格]:\n{table_str}") return '\n\n'.join(text_chunks)

注意：pdfplumber提取的文本可能包含多余的空格和换行，需要根据语义进行清洗和重组。对于双栏排版的学术论文，简单的按行提取会导致语句顺序错乱，可能需要更复杂的布局分析算法。

OCR场景的应对：对于扫描件，必须引入OCR。Tesseract是免费首选，但直接使用效果可能不佳。

# 安装Tesseract及中文语言包 # Ubuntu: sudo apt install tesseract-ocr tesseract-ocr-chi-sim # 使用前最好对图像进行预处理：灰度化、二值化、去噪

更稳健的做法是使用云OCR服务（如阿里云、百度AI开放平台），它们对复杂版面和手写体有更好的支持，但会产生费用。一个实用的技巧是：先尝试用普通解析器，如果返回的文本长度极短或乱码，则自动切换到OCR流程。这需要在代码中实现一个解析器的“降级策略”。

Office文档的陷阱：Word文档中的复杂格式（文本框、页眉页脚）、Excel中合并的单元格、PPT中文字在图形里，都是解析的难点。python-docx等库提供了遍历文档对象模型的能力，但你需要编写逻辑来忽略页脚、正确处理列表和标题层级。

实操心得：永远不要相信解析器能100%完美还原文档。必须在解析后加入一个“质量评估”环节，比如检查提取的文本是否包含大量乱码（如“^&*”）、有效句子比例是否过低。对于关键任务，可以设计一个人工复核的入口，或者记录解析失败的文档，后续集中处理。

2.2 文本分割与向量化：为AI准备“记忆面包”

把一本厚厚的书直接塞给AI，它也会“消化不良”。文本分割的目的就是把大文档切成容易“咀嚼”的小块。

分割策略的选择：

• 按固定长度分割：简单粗暴，但可能把一个完整的句子或段落从中间切断，破坏语义。
• 按分隔符递归分割：langchain的RecursiveCharacterTextSplitter是更优选择。它会优先尝试用双换行（\n\n）分割，不行再用单换行，再不行用句号，最后用空格，直到每个片段小于设定长度。这能更好地保持语义完整性。

from langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个片段的大致字符数 chunk_overlap=50, # 片段间重叠字符数，避免上下文断裂 separators=["\n\n", "\n", "。", "？", "！", "，", " ", ""] # 中文分隔符 ) chunks = text_splitter.split_text(long_document)

重叠（Overlap）是关键参数。设置50-100个字符的重叠，能确保关键信息（比如一个概念的定义）不会因为恰好落在两个片段边缘而丢失，让检索更准确。

向量模型的选择与调优：选择嵌入模型时，需要在效果、速度和本地部署成本间权衡。

• 轻量本地模型：all-MiniLM-L6-v2速度快，资源占用小，但语义捕捉能力稍弱。
• 平衡之选：BGE系列（如BGE-base-zh）针对中文优化，在效果和速度上取得了很好的平衡，是中文项目的热门选择。
• 云端重型模型：OpenAI的text-embedding-3-small或-large效果通常最好，但会产生API调用费用和网络延迟。

一个关键技巧是：对嵌入进行归一化（Normalization）。计算余弦相似度时，将所有向量归一化为单位长度，可以提升相似度计算的准确性和稳定性。很多库（如sentence-transformers）默认会做这件事，但自己实现时需要注意。

向量数据库的灌入：将分割后的文本块转换为向量，并存入向量数据库。这里要注意元数据（Metadata）的存储。除了文本本身，应该把片段的来源（文件名、页码、章节标题）也存进去。这样当AI返回答案时，你可以知道它引用了哪份文档的哪一页，方便溯源和验证。

# 伪代码示例 for i, chunk in enumerate(chunks): vector = embedding_model.encode(chunk) metadata = {"source": filename, "page": page_num, "chunk_id": i} vector_db.add(embeddings=vector, documents=chunk, metadatas=metadata)

2.3 提示词工程与问答链设计：引导AI成为专家

这是决定最终输出质量的核心。一个糟糕的提示词会让最强的LLM也表现失常。

系统提示词（System Prompt）的黄金法则：系统提示词定义了AI的角色和行为准则。对于文档QA，一个强大的系统提示词应包含：

1. 明确角色：“你是一个严谨的文档分析助手。”
2. 规定知识范围：“你只能基于用户提供的上下文信息来回答问题。如果上下文没有提供足够信息，请直接说‘根据提供的文档，无法回答此问题’，不要编造信息。”
3. 定义输出格式：“你的回答应当简洁、准确。如果可能，请引用上下文中的原文（注明出处），并以列表或要点形式组织答案。”
4. 安全与边界：“不要回答与文档无关的问题，不要生成任何有害或误导性内容。”

示例：

你是一个专业的文档信息提取助手。你的任务是根据用户提供的文档片段，准确回答用户的问题。 规则： 1. 你的回答必须严格基于提供的<context>内容。 2. 如果<context>中没有相关信息，请回答“文档中未提及相关信息”。 3. 如果问题与文档内容无关，请礼貌拒绝。 4. 回答要简洁、客观，优先引用原文表述。 上下文：<context> 问题：<question>

检索增强生成（RAG）流程的优化：标准的RAG是：用户提问 -> 将问题向量化 -> 从向量库检索Top K个相关片段 -> 将片段和问题一起发给LLM生成答案。 但这里可以优化：

• 重排序（Re-ranking）：初步检索出Top N（比如10个）片段后，使用一个更精细的交叉编码器模型（如bge-reranker）对它们进行重排序，选出最相关的Top K（比如3个）再给LLM，能显著提升答案相关性。
• 混合检索：结合关键词检索（如BM25）和向量检索。先用关键词快速过滤出相关文档，再用向量检索做语义精筛，兼顾召回率和准确率。
• 上下文压缩：如果检索到的片段总长度还是超过了LLM的上下文窗口，可以先用一个较小的LLM对这些片段进行摘要压缩，再将摘要送给主LLM生成最终答案。

任务链扩展：除了问答，这个“技能”还可以被设计成执行多种任务：

• 摘要链：输入文档 -> 分割 -> 为每个片段生成摘要 -> 聚合摘要生成总摘要。
• 翻译链：检索到相关片段 -> 指示LLM进行翻译。
• 信息格式化提取链：例如，“从合同中提取所有日期、金额和甲方乙方名称”，并输出为JSON格式。这需要设计更结构化的输出提示词，并可能用到LLM的函数调用（Function Calling）能力来输出标准JSON。

3. 从零搭建与核心环节实现

假设我们现在要仿照openclaw-docs-skill的思路，从零开始构建一个最小可行产品（MVP），以下是核心步骤和代码示例。

3.1 环境准备与依赖安装

首先，创建一个干净的Python环境（推荐3.9+），并安装核心依赖。

# 创建并激活虚拟环境 python -m venv venv_openclaw source venv_openclaw/bin/activate # Linux/Mac # venv_openclaw\Scripts\activate # Windows # 安装核心库 pip install langchain langchain-community langchain-chroma # LLM框架与向量库客户端 pip install sentence-transformers # 用于本地嵌入模型 pip install pdfplumber python-docx openpyxl # 文档解析 pip install fastapi uvicorn # 构建API接口 pip install pydantic # 数据验证 # 如果需要OCR pip install pytesseract pillow # 如果需要使用OpenAI等在线模型 pip install openai

注意：langchain及其生态包版本迭代很快，建议在项目中固定主要依赖的版本号，避免未来更新导致代码不兼容。

3.2 构建核心文档处理引擎

我们创建一个DocumentProcessor类，封装从文件到文本块的完整流程。

# document_processor.py import os from typing import List, Optional import pdfplumber from docx import Document import openpyxl from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document as LangchainDocument # 注意与docx的Document区分 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_and_split(self, file_path: str) -> List[LangchainDocument]: """根据文件后缀名，调用不同的解析方法，并分割文本""" ext = os.path.splitext(file_path)[1].lower() raw_text = "" if ext == '.pdf': raw_text = self._parse_pdf(file_path) elif ext == '.docx': raw_text = self._parse_docx(file_path) elif ext in ['.xlsx', '.xls']: raw_text = self._parse_excel(file_path) elif ext == '.txt': with open(file_path, 'r', encoding='utf-8') as f: raw_text = f.read() else: raise ValueError(f"Unsupported file type: {ext}") # 使用Langchain的Document对象存储，便于后续与向量库集成 texts = self.text_splitter.split_text(raw_text) docs = [LangchainDocument(page_content=text, metadata={"source": file_path}) for text in texts] return docs def _parse_pdf(self, file_path: str) -> str: full_text = [] try: with pdfplumber.open(file_path) as pdf: for page in pdf.pages: page_text = page.extract_text() if page_text: # 基础清洗 cleaned = ' '.join(page_text.split()) full_text.append(cleaned) except Exception as e: print(f"Error parsing PDF {file_path}: {e}") # 此处可加入OCR降级逻辑 full_text.append(f"[PDF解析异常: {e}]") return '\n'.join(full_text) def _parse_docx(self, file_path: str) -> str: doc = Document(file_path) full_text = [] for para in doc.paragraphs: if para.text.strip(): full_text.append(para.text) return '\n'.join(full_text) def _parse_excel(self, file_path: str) -> str: wb = openpyxl.load_workbook(file_path, data_only=True) # data_only获取计算后的值 all_data = [] for ws in wb.worksheets: for row in ws.iter_rows(values_only=True): # 将一行中非空单元格的值用逗号连接 row_str = ','.join([str(cell) for cell in row if cell is not None]) if row_str: all_data.append(row_str) return '\n'.join(all_data) # 使用示例 if __name__ == "__main__": processor = DocumentProcessor() docs = processor.load_and_split("sample_contract.pdf") print(f"成功将文档分割为 {len(docs)} 个文本块。") for i, doc in enumerate(docs[:2]): # 打印前两个块 print(f"--- Chunk {i} ---") print(doc.page_content[:200] + "...") print(f"Metadata: {doc.metadata}")

3.3 实现向量存储与检索系统

接下来，我们实现一个简单的向量数据库管理类，用于存储文档块和进行语义检索。

# vector_store_manager.py from langchain_chroma import Chroma from langchain_huggingface import HuggingFaceEmbeddings import os from typing import List from langchain.schema import Document as LangchainDocument class VectorStoreManager: def __init__(self, persist_directory="./chroma_db", embedding_model_name="BAAI/bge-small-zh-v1.5"): """ 初始化向量存储管理器。 persist_directory: 向量数据库持久化目录 embedding_model_name: 使用的嵌入模型名称 """ self.persist_directory = persist_directory # 初始化嵌入模型 self.embeddings = HuggingFaceEmbeddings( model_name=embedding_model_name, model_kwargs={'device': 'cpu'}, # 根据环境改为'cuda' encode_kwargs={'normalize_embeddings': True} # 关键：归一化向量 ) # 初始化或加载Chroma向量库 self.vectorstore = Chroma( collection_name="docs_collection", embedding_function=self.embeddings, persist_directory=persist_directory ) def add_documents(self, documents: List[LangchainDocument]): """将文档列表添加到向量库""" # Chroma的add_documents方法会自动将文档文本进行嵌入并存储 self.vectorstore.add_documents(documents) self.vectorstore.persist() # 持久化到磁盘 print(f"已成功添加 {len(documents)} 个文档块到向量库。") def similarity_search(self, query: str, k: int = 4) -> List[LangchainDocument]: """在向量库中进行语义搜索，返回最相关的k个文档块""" results = self.vectorstore.similarity_search(query, k=k) return results def clear(self): """清空向量库（谨慎操作）""" self.vectorstore.delete_collection() print("向量库已清空。") # 集成示例 if __name__ == "__main__": from document_processor import DocumentProcessor # 1. 处理文档 processor = DocumentProcessor() docs = processor.load_and_split("your_document.pdf") # 2. 创建向量库并添加文档 vs_manager = VectorStoreManager() vs_manager.add_documents(docs) # 3. 进行查询 query = "本合同中的违约责任是如何规定的？" relevant_docs = vs_manager.similarity_search(query, k=3) print(f"针对查询 '{query}'， 找到 {len(relevant_docs)} 个相关片段：") for i, doc in enumerate(relevant_docs): print(f"\n--- 相关片段 {i+1} (来源: {doc.metadata['source']}) ---") print(doc.page_content[:300]) # 打印前300字符

3.4 集成大语言模型与构建问答链

最后，我们将检索到的上下文与用户问题结合，发送给LLM生成最终答案。这里以使用本地模型（通过Ollama）和在线API（OpenAI）两种方式为例。

# qa_chain.py from langchain.prompts import ChatPromptTemplate from langchain.chat_models import init_chat_model from langchain.schema import StrOutputParser from langchain_core.runnables import RunnablePassthrough from typing import List class QAChatChain: def __init__(self, llm_type="openai", model_name="gpt-3.5-turbo", base_url=None, api_key=None): """ 初始化问答链。 llm_type: 可以是 'openai', 'ollama', 'zhipu' 等 model_name: 模型名称，如 'gpt-4', 'qwen2:7b' base_url: 对于本地部署的API（如Ollama），需要指定URL api_key: 对于在线API，需要密钥 """ # 根据类型初始化聊天模型 if llm_type == "openai": from langchain_openai import ChatOpenAI self.llm = ChatOpenAI(model=model_name, api_key=api_key, temperature=0.1) # temperature低，输出更确定 elif llm_type == "ollama": from langchain_community.chat_models import ChatOllama self.llm = ChatOllama(base_url=base_url or "http://localhost:11434", model=model_name, temperature=0.1) else: raise ValueError(f"Unsupported LLM type: {llm_type}") # 定义提示词模板 self.prompt_template = ChatPromptTemplate.from_messages([ ("system", """你是一个专业的文档分析助手。请严格根据以下上下文来回答问题。如果上下文不包含回答问题所需的信息，请直接说“根据提供的文档，无法回答此问题”。不要编造信息。回答要简洁、准确，可以引用原文。 上下文：{context} 问题：{question} 答案："""), ]) # 构建链：合并上下文和问题 -> 格式化提示词 -> 调用LLM -> 解析输出 self.chain = ( {"context": RunnablePassthrough(), "question": RunnablePassthrough()} | self.prompt_template | self.llm | StrOutputParser() ) def format_context(self, docs: List) -> str: """将检索到的多个文档片段格式化为一个连续的上下文字符串""" context_parts = [] for i, doc in enumerate(docs): context_parts.append(f"[片段 {i+1}, 来源: {doc.metadata.get('source', 'N/A')}]\n{doc.page_content}") return "\n\n".join(context_parts) def ask(self, question: str, context_docs: List) -> str: """核心问答方法""" formatted_context = self.format_context(context_docs) answer = self.chain.invoke({"context": formatted_context, "question": question}) return answer # 完整流程示例 if __name__ == "__main__": import os from document_processor import DocumentProcessor from vector_store_manager import VectorStoreManager # 0. 假设文档已处理并存入向量库 # processor = DocumentProcessor() # docs = processor.load_and_split("contract.pdf") # vs_manager = VectorStoreManager() # vs_manager.add_documents(docs) # 1. 用户提问 user_question = "本合同约定的服务期限是多久？" # 2. 从向量库检索相关片段 vs_manager = VectorStoreManager() # 加载已有的向量库 retrieved_docs = vs_manager.similarity_search(user_question, k=3) # 3. 初始化问答链（这里以Ollama本地模型为例） # 确保已安装并运行Ollama，并拉取了模型，如：ollama pull qwen2:7b qa_chain = QAChatChain( llm_type="ollama", model_name="qwen2:7b", # 或 "llama3:8b" base_url="http://localhost:11434" ) # 4. 生成答案 if retrieved_docs: answer = qa_chain.ask(user_question, retrieved_docs) print(f"问题：{user_question}") print(f"答案：{answer}") print("\n--- 引用的上下文片段 ---") for doc in retrieved_docs: print(f"... {doc.page_content[:100]}...") else: print("未找到相关文档内容。")

3.5 封装为Web API服务

为了让这个“技能”能被其他系统调用，我们需要用FastAPI将其包装成HTTP接口。

# main.py (FastAPI 应用入口) from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.responses import JSONResponse from pydantic import BaseModel import os import uuid from typing import List, Optional from document_processor import DocumentProcessor from vector_store_manager import VectorStoreManager from qa_chain import QAChatChain app = FastAPI(title="OpenClaw Docs Skill API") # 全局初始化（生产环境应考虑依赖注入和生命周期管理） processor = DocumentProcessor() vs_manager = VectorStoreManager() # 注意：LLM初始化可能需要API KEY，应从环境变量读取 qa_chain = QAChatChain(llm_type="ollama", model_name="qwen2:7b") UPLOAD_DIR = "./uploaded_docs" os.makedirs(UPLOAD_DIR, exist_ok=True) class QueryRequest(BaseModel): question: str top_k: Optional[int] = 4 class AnswerResponse(BaseModel): answer: str sources: List[str] # 可以返回来源文档名或片段ID @app.post("/upload/") async def upload_document(file: UploadFile = File(...)): """上传并处理文档，将其内容存入向量库""" if not file.filename: raise HTTPException(status_code=400, detail="No file provided") # 生成唯一文件名保存 file_ext = os.path.splitext(file.filename)[1] saved_filename = f"{uuid.uuid4().hex}{file_ext}" saved_path = os.path.join(UPLOAD_DIR, saved_filename) try: # 保存文件 contents = await file.read() with open(saved_path, "wb") as f: f.write(contents) # 处理文档 docs = processor.load_and_split(saved_path) # 添加到向量库 vs_manager.add_documents(docs) return JSONResponse(content={ "message": "Document processed successfully", "document_id": saved_filename, "chunks_added": len(docs) }) except Exception as e: raise HTTPException(status_code=500, detail=f"Document processing failed: {str(e)}") @app.post("/ask/", response_model=AnswerResponse) async def ask_question(req: QueryRequest): """根据已处理的文档回答问题""" try: # 1. 检索 relevant_docs = vs_manager.similarity_search(req.question, k=req.top_k) if not relevant_docs: return AnswerResponse(answer="知识库中未找到相关信息。", sources=[]) # 2. 生成答案 answer = qa_chain.ask(req.question, relevant_docs) # 3. 提取来源信息 sources = list(set([doc.metadata.get("source", "Unknown") for doc in relevant_docs])) return AnswerResponse(answer=answer, sources=sources) except Exception as e: raise HTTPException(status_code=500, detail=f"QA process failed: {str(e)}") @app.get("/health") async def health_check(): return {"status": "healthy"} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

运行python main.py后，你就拥有了一个具备/upload/和/ask/端口的本地API服务。其他应用可以通过HTTP请求与之交互，实现文档处理与智能问答能力。

4. 部署、优化与避坑指南

将原型部署到生产环境并稳定运行，会遇到一系列新挑战。

4.1 性能优化与规模化考量

1. 向量检索速度：

• 索引选择：Chroma默认使用HNSW索引，对于千万级以下的数据量表现良好。如果数据量极大（上亿条），需要考虑Milvus或Qdrant，它们对分布式和GPU加速支持更好。
• 量化与降维：对于768维或1024维的向量，可以考虑使用量化技术（如PQ， Product Quantization）在轻微损失精度的情况下大幅减少存储和计算开销。一些嵌入模型也提供降维版本（如text-embedding-3-small可输出256维）。
• 预过滤：如果文档有明确的类别、时间等标签，可以先通过传统数据库（如PostgreSQL）进行标签过滤，缩小向量检索的范围。

2. 大语言模型推理成本与延迟：

• 模型选型：在效果和成本间权衡。对于内部知识库问答，7B-13B参数的量化模型（如Qwen2-7B-Instruct, Llama 3-8B）在消费级GPU上即可运行，且效果足够。对准确性要求极高的场景，再考虑调用GPT-4等大模型API。
• 上下文长度管理：严格控制送入模型的上下文总长度（问题+检索到的片段）。可以设置一个最大Token限制，超过则优先保留相似度最高的片段，或对片段进行二次摘要。
• 流式输出与异步：对于长答案生成，使用流式输出（Server-Sent Events）提升用户体验。耗时的文档解析和向量化操作，应放入Celery等任务队列异步执行，避免阻塞API请求。

3. 系统资源管理：

• 嵌入模型缓存：重复编码相同的文本是浪费。可以建立一个小型缓存（如LRU Cache），对近期处理过的文本块直接返回缓存向量。
• 向量库持久化与备份：Chroma的持久化目录需要定期备份。对于高可用场景，应考虑使用支持客户端-服务器模式的向量数据库。

4.2 效果提升的进阶技巧

1. 提升检索质量：

• 查询扩展（Query Expansion）：在将用户问题向量化前，先用LLM对问题进行改写或扩展。例如，将“违约责任”扩展为“违约责任、赔偿条款、违约金”。这能提高召回率。
• 多向量检索（Multi-Vector）：除了存储文本块的向量，还可以存储其摘要的向量、或提取出的关键实体（人名、组织名、日期）的向量。检索时，综合多个向量的结果。
• 元数据过滤：允许用户在提问时附加过滤器，如“在2023年的报告中找...”。这需要你在存储时丰富元数据。

2. 提升答案准确性：

• 引用溯源（Citation）：强制LLM在答案中引用来源片段的ID或页码。这不仅能增加可信度，也方便用户核查。在提示词中明确要求：“请在你的答案末尾，用【来源X】的形式注明引用。”
• 自洽性校验（Self-Consistency）：对于重要问题，可以让LLM基于不同检索结果或不同思考链多次生成答案，然后选择一个最一致或投票最多的答案。
• 后处理（Post-Processing）：对LLM生成的答案进行基础校验，比如检查是否有明显的矛盾、是否包含“根据上下文”等短语（如果上下文为空，则不应出现此短语）。

4.3 常见问题与排查实录

在实际开发和运维中，你几乎一定会遇到以下问题：

问题1：LLM回答“根据提供的文档，无法回答此问题”，但明明文档里有相关内容。

• 可能原因1：检索失败。用户问题的表述和文档中的表述差异太大，向量检索没找到正确片段。排查：打印出检索到的Top K个片段内容，看是否包含答案。如果没有，考虑优化嵌入模型或引入查询扩展。
• 可能原因2：上下文不足或噪声太大。检索到的片段可能包含了答案，但被大量无关文本淹没，LLM没“看到”。排查：检查文本分割是否合理，重叠是否足够。尝试增加top_k参数，或使用重排序模型筛选出最相关的1-2个片段。
• 可能原因3：提示词限制过死。系统提示词中“如果上下文没有...请直接说无法回答”可能被过度执行。排查：微调提示词语气，或尝试让模型先复述相关上下文再判断。

问题2：处理扫描PDF或图片时，OCR效果差，提取文本乱码。

• 解决方案：
  1. 图像预处理：在使用Tesseract前，先对图像进行灰度化、二值化、降噪、矫正倾斜等操作。OpenCV和PIL库可以完成这些任务。
  2. 调整OCR参数：Tesseract有PSM（页面分割模式）和OEM（OCR引擎模式）参数，针对纯文本、单列、多列等不同版面进行调整。
  3. 使用商用OCR API：对于关键业务，投资阿里云、百度云的高精度OCR服务，它们通常对复杂版面、手写体和低质量图片有更好的识别率。
  4. 人工复核流程：对于识别置信度低的文档，系统将其标记并转入人工复核队列。

问题3：向量数据库占用磁盘空间增长过快。

• 解决方案：
  1. 定期清理：建立文档生命周期管理，对于过期的、临时处理的文档，定期从向量库中删除其对应的向量。Chroma支持按元数据过滤删除。
  2. 向量压缩：采用量化技术存储向量。
  3. 优化文本分割：避免分割得过细。在保持语义完整的前提下，适当增大chunk_size，减少总的向量数量。
  4. 分库分集合：按业务线或文档类型，使用不同的集合（Collection），可以独立管理和清理。

问题4：API并发请求时响应慢或超时。

• 解决方案：
  1. 异步化：将文档解析、向量化等CPU/IO密集型任务全部改为异步任务（使用asyncio或Celery），API接口只负责接收请求和返回任务ID，通过WebSocket或轮询通知结果。
  2. 限流与队列：使用API网关或slowapi等中间件对/ask接口进行限流，防止被高频请求打垮。将请求排队处理。
  3. 模型服务化：将LLM模型和嵌入模型部署为独立的推理服务（如使用Triton Inference Server, vLLM），通过gRPC调用，实现资源复用和负载均衡。
  4. 缓存答案：对于完全相同的提问，可以将答案缓存一段时间（如Redis），直接返回，避免重复检索和推理。

问题5：LLM产生“幻觉”，编造文档中没有的信息。

• 解决方案：
  1. 强化提示词：在系统提示词中多次、强调地要求“严格基于上下文”，并设定惩罚性语句，如“任何编造信息都是不可接受的”。
  2. 提供少量示例（Few-Shot）：在提示词中提供1-2个“基于上下文回答”和“上下文无信息时拒绝回答”的示例，让模型通过示例学习。
  3. 输出格式约束：要求模型以“引用原文：...”的格式输出，这能在心理上约束其编造行为。
  4. 答案验证：对于关键答案，可以设计一个简单的验证步骤，例如，让另一个LLM实例或规则系统判断答案中的关键事实是否能在提供的上下文中找到直接支持。

构建一个像openclaw-docs-skill这样成熟可用的文档处理技能，远不止是拼接几个开源库。它需要你在文档解析的鲁棒性、检索的准确性、提示词的精细度、系统架构的扩展性以及运维的便捷性上持续打磨。每一个环节的细微优化，都可能带来最终用户体验的显著提升。从解决一个具体的文档处理痛点开始，逐步迭代，你也能打造出属于自己的智能“机械爪”。