在日常开发中，我们经常会遇到需要处理各种数据格式和文件类型的场景。特别是随着大语言模型（LLM）技术的快速发展，如何有效地将文档内容传递给LLM进行智能问答成为了一个热门话题。最近在项目中遇到了一个具体需求：需要将单个PDF文件的内容传递给LLM模型进行问答交互。这个需求看似简单，但实际操作中却涉及文件解析、文本处理、上下文构建等多个技术环节。

本文将围绕"如何将单个PDF文件传递给LLM进行问答"这一主题，从技术原理到完整实现，为开发者提供一套可落地的解决方案。无论你是刚接触LLM的新手，还是有一定经验的开发者，都能从本文中获得实用的技术指导。

1. LLM与文档问答的技术背景

1.1 什么是LLM及其在文档处理中的应用

大语言模型（Large Language Model，简称LLM）是一种基于深度学习的人工智能模型，能够理解和生成人类语言。近年来，随着GPT、LLaMA等模型的兴起，LLM在自然语言处理领域展现出了强大的能力。

在文档处理场景中，LLM可以用于：

• 文档内容摘要和总结
• 基于文档的智能问答
• 文档内容分析和提取
• 多文档信息整合

1.2 PDF文档处理的特殊性

PDF（Portable Document Format）作为一种常见的文档格式，具有以下特点：

• 格式固定，保持原始布局
• 可能包含文本、图片、表格等混合内容
• 文本可能以非连续方式存储
• 支持加密和权限控制

这些特性使得PDF文档的解析比普通文本文件更加复杂，需要专门的工具和技术来处理。

1.3 文档问答的技术架构

一个完整的文档问答系统通常包含以下组件：

1. 文档解析模块：负责提取PDF中的文本内容
2. 文本预处理模块：清理和标准化提取的文本
3. 向量化模块：将文本转换为数值向量
4. 检索模块：根据问题检索相关文本片段
5. LLM推理模块：基于检索结果生成答案

2. 环境准备与工具选择

2.1 核心工具库介绍

要实现PDF到LLM的完整流程，我们需要以下几个关键工具：

PDF解析工具：

• PyPDF2：轻量级的PDF处理库
• pdfplumber：更强大的PDF文本提取工具
• Tika：Apache的文档内容提取工具

文本处理工具：

• NLTK：自然语言处理工具包
• spaCy：工业级NLP库
• LangChain：LLM应用开发框架

LLM接入工具：

• OpenAI API
• Hugging Face Transformers
• 本地部署的开源模型

2.2 环境配置要求

# requirements.txt # PDF处理相关 PyPDF2==3.0.1 pdfplumber==0.10.3 # 文本处理相关 nltk==3.8.1 spacy==3.7.2 langchain==0.1.0 langchain-community==0.0.10 # LLM相关 openai==1.3.0 transformers==4.35.0 torch==2.1.0 # 其他工具 numpy==1.24.0 pandas==2.0.0

2.3 开发环境搭建

# 创建虚拟环境 python -m venv pdf_llm_env source pdf_llm_env/bin/activate # Linux/Mac # 或 pdf_llm_env\Scripts\activate # Windows # 安装依赖 pip install -r requirements.txt # 下载spaCy模型 python -m spacy download en_core_web_sm

3. PDF文档解析技术详解

3.1 使用pdfplumber进行文本提取

pdfplumber是目前最优秀的PDF文本提取库之一，它能够准确保持文本的原始顺序和布局。

import pdfplumber import re from typing import List, Dict class PDFParser: def __init__(self, pdf_path: str): self.pdf_path = pdf_path self.text_content = "" def extract_text(self) -> str: """提取PDF中的文本内容""" try: with pdfplumber.open(self.pdf_path) as pdf: full_text = [] for page in pdf.pages: # 提取页面文本 page_text = page.extract_text() if page_text: full_text.append(page_text) self.text_content = "\n".join(full_text) return self.text_content except Exception as e: print(f"PDF解析错误: {e}") return "" def clean_text(self, text: str) -> str: """清理提取的文本""" # 移除多余的换行和空格 text = re.sub(r'\n+', '\n', text) text = re.sub(r' +', ' ', text) # 移除特殊字符但保留标点 text = re.sub(r'[^\w\s\.\,\!\?\-\:\(\)]', '', text) return text.strip() def get_structured_content(self) -> Dict[int, str]: """获取分页的结构化内容""" structured_content = {} try: with pdfplumber.open(self.pdf_path) as pdf: for page_num, page in enumerate(pdf.pages, 1): page_text = page.extract_text() if page_text: cleaned_text = self.clean_text(page_text) structured_content[page_num] = cleaned_text except Exception as e: print(f"结构化提取错误: {e}") return structured_content # 使用示例 if __name__ == "__main__": parser = PDFParser("sample.pdf") text = parser.extract_text() print(f"提取的文本长度: {len(text)} 字符") structured = parser.get_structured_content() for page, content in structured.items(): print(f"第{page}页: {content[:100]}...")

3.2 处理复杂PDF布局

对于包含表格、多栏布局的复杂PDF，需要更精细的处理：

def extract_complex_pdf(pdf_path: str) -> Dict[str, List]: """处理复杂布局的PDF""" result = { 'text': [], 'tables': [], 'images': [] } with pdfplumber.open(pdf_path) as pdf: for page_num, page in enumerate(pdf.pages): # 提取文本（尝试保持阅读顺序） text = page.extract_text(x_tolerance=3, y_tolerance=3) if text: result['text'].append({ 'page': page_num + 1, 'content': text }) # 提取表格 tables = page.extract_tables() for table in tables: if table: result['tables'].append({ 'page': page_num + 1, 'table': table }) return result

3.3 文本分块与预处理

将长文档分割成适合LLM处理的片段：

from langchain.text_splitter import RecursiveCharacterTextSplitter class TextProcessor: def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200): self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, length_function=len, separators=["\n\n", "\n", "。", "！", "？", "．", ".", "!", "?"] ) def split_text(self, text: str) -> List[str]: """将文本分割成块""" return self.text_splitter.split_text(text) def preprocess_text(self, text: str) -> str: """文本预处理""" # 移除URL text = re.sub(r'http\S+', '', text) # 标准化空格 text = re.sub(r'\s+', ' ', text) # 处理特殊字符 text = text.encode('utf-8', 'ignore').decode('utf-8') return text.strip() # 使用示例 processor = TextProcessor() pdf_parser = PDFParser("document.pdf") raw_text = pdf_parser.extract_text() cleaned_text = processor.preprocess_text(raw_text) chunks = processor.split_text(cleaned_text) print(f"原始文本分割为 {len(chunks)} 个块") for i, chunk in enumerate(chunks[:3]): # 显示前3个块 print(f"块 {i+1}: {chunk[:100]}...")

4. LLM集成与问答系统实现

4.1 基于LangChain的问答系统架构

LangChain提供了构建LLM应用的完整框架，特别适合文档问答场景。

from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings from langchain.vectorstores import FAISS from langchain.chains import RetrievalQA from langchain.llms import OpenAI from langchain.prompts import PromptTemplate import os class PDFQASystem: def __init__(self, api_key: str = None, model_name: str = "gpt-3.5-turbo"): # 设置API密钥 if api_key: os.environ["OPENAI_API_KEY"] = api_key # 初始化嵌入模型 self.embeddings = OpenAIEmbeddings() # 或者使用本地模型 # self.embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2") # 初始化LLM self.llm = OpenAI(temperature=0, model_name=model_name) self.vector_store = None self.qa_chain = None def create_vector_store(self, text_chunks: List[str]): """创建向量数据库""" self.vector_store = FAISS.from_texts(text_chunks, self.embeddings) def setup_qa_chain(self): """设置问答链""" # 自定义提示模板 prompt_template = """基于以下上下文信息，请回答问题。如果你不确定答案，请说不知道，不要编造信息。 上下文： {context} 问题：{question} 答案：""" PROMPT = PromptTemplate( template=prompt_template, input_variables=["context", "question"] ) self.qa_chain = RetrievalQA.from_chain_type( llm=self.llm, chain_type="stuff", retriever=self.vector_store.as_retriever(search_kwargs={"k": 3}), chain_type_kwargs={"prompt": PROMPT}, return_source_documents=True ) def ask_question(self, question: str) -> Dict: """提问并获取答案""" if not self.qa_chain: raise ValueError("请先调用setup_qa_chain()设置问答链") result = self.qa_chain({"query": question}) return { "question": question, "answer": result["result"], "source_documents": result["source_documents"] } # 完整使用示例 def main(): # 1. 解析PDF parser = PDFParser("technical_document.pdf") text = parser.extract_text() # 2. 处理文本 processor = TextProcessor() cleaned_text = processor.preprocess_text(text) chunks = processor.split_text(cleaned_text) # 3. 初始化QA系统 qa_system = PDFQASystem(api_key="your-openai-api-key") qa_system.create_vector_store(chunks) qa_system.setup_qa_chain() # 4. 进行问答 questions = [ "文档的主要内容包括什么？", "文档中提到了哪些关键技术？", "作者的主要观点是什么？" ] for question in questions: result = qa_system.ask_question(question) print(f"问题: {result['question']}") print(f"答案: {result['answer']}") print(f"来源: {len(result['source_documents'])} 个相关文档块") print("-" * 50) if __name__ == "__main__": main()

4.2 使用本地LLM模型

对于数据敏感或需要离线使用的场景，可以使用本地部署的LLM：

from langchain.llms import LlamaCpp from langchain.embeddings import HuggingFaceEmbeddings class LocalPDFQASystem: def __init__(self, model_path: str): # 使用本地嵌入模型 self.embeddings = HuggingFaceEmbeddings( model_name="sentence-transformers/all-MiniLM-L6-v2" ) # 加载本地LLM self.llm = LlamaCpp( model_path=model_path, temperature=0.1, max_tokens=2000, top_p=1, verbose=True, n_ctx=2048 ) self.vector_store = None self.qa_chain = None # 其他方法与PDFQASystem类似 def create_vector_store(self, text_chunks: List[str]): self.vector_store = FAISS.from_texts(text_chunks, self.embeddings) def setup_qa_chain(self): self.qa_chain = RetrievalQA.from_chain_type( llm=self.llm, chain_type="stuff", retriever=self.vector_store.as_retriever(search_kwargs={"k": 3}), return_source_documents=True ) # 使用示例（需要先下载模型） local_system = LocalPDFQASystem("path/to/llama-model.gguf")

4.3 高级检索策略

为了提高问答质量，可以实现更复杂的检索策略：

from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain.vectorstores import FAISS class AdvancedRetrievalQA: def __init__(self, embeddings, llm): self.embeddings = embeddings self.llm = llm def create_ensemble_retriever(self, texts: List[str]): """创建混合检索器""" # 向量检索器 vector_store = FAISS.from_texts(texts, self.embeddings) vector_retriever = vector_store.as_retriever(search_kwargs={"k": 3}) # BM25检索器 bm25_retriever = BM25Retriever.from_texts(texts) bm25_retriever.k = 3 # 混合检索器 ensemble_retriever = EnsembleRetriever( retrievers=[vector_retriever, bm25_retriever], weights=[0.5, 0.5] ) return ensemble_retriever def setup_advanced_qa(self, texts: List[str]): """设置高级问答系统""" retriever = self.create_ensemble_retriever(texts) self.qa_chain = RetrievalQA.from_chain_type( llm=self.llm, chain_type="map_reduce", # 处理长文档 retriever=retriever, return_source_documents=True, chain_type_kwargs={ "question_prompt": PromptTemplate( template="问题：{question}\n上下文：{context}\n答案：", input_variables=["question", "context"] ) } )

5. 完整实战案例：技术文档问答系统

5.1 项目结构设计

pdf_qa_system/ ├── src/ │ ├── __init__.py │ ├── pdf_parser.py # PDF解析模块 │ ├── text_processor.py # 文本处理模块 │ ├── qa_system.py # 问答系统模块 │ └── utils.py # 工具函数 ├── data/ │ ├── input/ # 输入PDF文件 │ └── processed/ # 处理后的文本 ├── tests/ # 测试文件 ├── config/ │ └── settings.py # 配置文件 ├── requirements.txt └── main.py # 主程序

5.2 配置文件设计

# config/settings.py import os from typing import Dict, Any class Settings: # PDF处理配置 PDF_CONFIG = { "chunk_size": 1000, "chunk_overlap": 200, "max_pages": None, # None表示处理所有页面 } # 模型配置 MODEL_CONFIG = { "embedding_model": "text-embedding-ada-002", # 或 "all-MiniLM-L6-v2" "llm_model": "gpt-3.5-turbo", "temperature": 0.1, "max_tokens": 1000, } # 检索配置 RETRIEVAL_CONFIG = { "search_kwargs": {"k": 3}, "score_threshold": 0.7, } @classmethod def get_api_key(cls) -> str: return os.getenv("OPENAI_API_KEY", "") @classmethod def get_model_path(cls) -> str: return os.getenv("LOCAL_MODEL_PATH", "./models") settings = Settings()

5.3 主程序实现

# main.py import argparse import json from datetime import datetime from src.pdf_parser import PDFParser from src.text_processor import TextProcessor from src.qa_system import PDFQASystem from config.settings import settings class PDFQAAplication: def __init__(self): self.parser = PDFParser() self.processor = TextProcessor( chunk_size=settings.PDF_CONFIG["chunk_size"], chunk_overlap=settings.PDF_CONFIG["chunk_overlap"] ) self.qa_system = None def process_pdf(self, pdf_path: str) -> Dict[str, Any]: """处理PDF文件并返回处理结果""" print(f"开始处理PDF文件: {pdf_path}") # 解析PDF start_time = datetime.now() raw_text = self.parser.extract_text(pdf_path) parsing_time = datetime.now() - start_time if not raw_text: raise ValueError(f"无法从PDF文件中提取文本: {pdf_path}") # 处理文本 start_time = datetime.now() cleaned_text = self.processor.preprocess_text(raw_text) chunks = self.processor.split_text(cleaned_text) processing_time = datetime.now() - start_time result = { "pdf_path": pdf_path, "original_text_length": len(raw_text), "cleaned_text_length": len(cleaned_text), "number_of_chunks": len(chunks), "parsing_time": str(parsing_time), "processing_time": str(processing_time), "sample_chunks": chunks[:2] # 保存前两个块作为样本 } return result, chunks def initialize_qa_system(self, api_key: str = None, use_local: bool = False): """初始化问答系统""" if use_local: from src.qa_system import LocalPDFQASystem model_path = settings.get_model_path() self.qa_system = LocalPDFQASystem(model_path) else: api_key = api_key or settings.get_api_key() if not api_key: raise ValueError("请提供OpenAI API密钥或设置环境变量") self.qa_system = PDFQASystem(api_key=api_key) def interactive_qa(self, chunks: List[str]): """交互式问答模式""" if not self.qa_system: raise ValueError("请先初始化问答系统") # 创建向量存储 self.qa_system.create_vector_store(chunks) self.qa_system.setup_qa_chain() print("问答系统已就绪，输入'quit'退出") while True: question = input("\n请输入问题: ").strip() if question.lower() in ['quit', 'exit', '退出']: break if not question: continue try: result = self.qa_system.ask_question(question) print(f"\n答案: {result['answer']}") print(f"参考来源: {len(result['source_documents'])} 个相关段落") # 显示来源信息 for i, doc in enumerate(result['source_documents'][:2], 1): print(f"来源 {i}: {doc.page_content[:100]}...") except Exception as e: print(f"回答问题时出错: {e}") def main(): parser = argparse.ArgumentParser(description="PDF文档问答系统") parser.add_argument("--pdf", required=True, help="PDF文件路径") parser.add_argument("--api-key", help="OpenAI API密钥") parser.add_argument("--local", action="store_true", help="使用本地模型") parser.add_argument("--questions", help="问题文件路径（JSON格式）") args = parser.parse_args() # 创建应用实例 app = PDFQAAplication() try: # 处理PDF processing_result, chunks = app.process_pdf(args.pdf) print("PDF处理完成:") print(json.dumps(processing_result, indent=2, ensure_ascii=False)) # 初始化QA系统 app.initialize_qa_system(api_key=args.api_key, use_local=args.local) # 批量问答或交互式问答 if args.questions: with open(args.questions, 'r', encoding='utf-8') as f: questions = json.load(f) app.qa_system.create_vector_store(chunks) app.qa_system.setup_qa_chain() for question in questions: result = app.qa_system.ask_question(question) print(f"\n问题: {question}") print(f"答案: {result['answer']}") else: app.interactive_qa(chunks) except Exception as e: print(f"程序执行出错: {e}") if __name__ == "__main__": main()

5.4 测试用例

# tests/test_pdf_qa.py import unittest import tempfile import os from src.pdf_parser import PDFParser from src.text_processor import TextProcessor class TestPDFQA(unittest.TestCase): def setUp(self): # 创建测试PDF文件（简单文本） self.test_pdf_content = """ 这是一个测试文档。 文档包含多个段落的信息。 第一节：介绍 这是第一节的内容，包含一些技术术语和概念说明。 第二节：实现 描述具体的实现方法和步骤。 """ # 在实际测试中，这里应该创建一个真实的PDF文件 # 为简化示例，我们直接使用文本测试 def test_text_extraction(self): """测试文本提取功能""" # 这里应该使用真实的PDF文件进行测试 # 简化示例中直接测试文本处理 processor = TextProcessor() chunks = processor.split_text(self.test_pdf_content) self.assertGreater(len(chunks), 0) self.assertTrue(all(len(chunk) <= 1000 for chunk in chunks)) def test_text_cleaning(self): """测试文本清理功能""" processor = TextProcessor() dirty_text = "这是 有多个空格 和\n换行\n的文本。" cleaned = processor.preprocess_text(dirty_text) self.assertNotIn(" ", cleaned) # 不应该有多个连续空格 self.assertNotIn("\n\n", cleaned) # 不应该有多个连续换行 if __name__ == "__main__": unittest.main()

6. 性能优化与最佳实践

6.1 处理大型PDF文档

对于超过100页的大型PDF文档，需要特殊处理：

class LargePDFProcessor: def __init__(self, max_memory_chunks: int = 1000): self.max_memory_chunks = max_memory_chunks def process_large_pdf(self, pdf_path: str, output_dir: str): """处理大型PDF，分块保存到磁盘""" os.makedirs(output_dir, exist_ok=True) with pdfplumber.open(pdf_path) as pdf: total_pages = len(pdf.pages) chunks_collected = [] chunk_file_count = 0 for page_num, page in enumerate(pdf.pages, 1): text = page.extract_text() if text: chunks_collected.append({ 'page': page_num, 'content': text }) # 达到内存限制时保存到文件 if len(chunks_collected) >= self.max_memory_chunks: self._save_chunks(chunks_collected, output_dir, chunk_file_count) chunks_collected = [] chunk_file_count += 1 print(f"已处理 {page_num}/{total_pages} 页") # 保存剩余块 if chunks_collected: self._save_chunks(chunks_collected, output_dir, chunk_file_count) def _save_chunks(self, chunks: List[Dict], output_dir: str, file_count: int): """保存块到文件""" filename = os.path.join(output_dir, f"chunks_{file_count:04d}.json") with open(filename, 'w', encoding='utf-8') as f: json.dump(chunks, f, ensure_ascii=False, indent=2)

6.2 缓存机制实现

为了避免重复处理相同的PDF文件，可以实现缓存机制：

import hashlib import pickle from pathlib import Path class CacheManager: def __init__(self, cache_dir: str = "./cache"): self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(exist_ok=True) def get_file_hash(self, file_path: str) -> str: """计算文件哈希值""" hasher = hashlib.md5() with open(file_path, 'rb') as f: for chunk in iter(lambda: f.read(4096), b""): hasher.update(chunk) return hasher.hexdigest() def get_cached_result(self, file_path: str, cache_key: str) -> Any: """获取缓存结果""" file_hash = self.get_file_hash(file_path) cache_file = self.cache_dir / f"{file_hash}_{cache_key}.pkl" if cache_file.exists(): with open(cache_file, 'rb') as f: return pickle.load(f) return None def save_to_cache(self, file_path: str, cache_key: str, data: Any): """保存到缓存""" file_hash = self.get_file_hash(file_path) cache_file = self.cache_dir / f"{file_hash}_{cache_key}.pkl" with open(cache_file, 'wb') as f: pickle.dump(data, f)

6.3 异步处理优化

对于需要处理多个PDF文件的场景，可以使用异步处理：

import asyncio from concurrent.futures import ThreadPoolExecutor class AsyncPDFProcessor: def __init__(self, max_workers: int = 4): self.executor = ThreadPoolExecutor(max_workers=max_workers) async def process_multiple_pdfs(self, pdf_paths: List[str]) -> List[Dict]: """异步处理多个PDF文件""" loop = asyncio.get_event_loop() tasks = [] for pdf_path in pdf_paths: task = loop.run_in_executor( self.executor, self._process_single_pdf, pdf_path ) tasks.append(task) results = await asyncio.gather(*tasks, return_exceptions=True) return results def _process_single_pdf(self, pdf_path: str) -> Dict: """处理单个PDF文件（同步方法）""" parser = PDFParser(pdf_path) text = parser.extract_text() processor = TextProcessor() chunks = processor.split_text(text) return { 'file_path': pdf_path, 'chunks': chunks, 'chunk_count': len(chunks) } # 使用示例 async def main_async(): pdf_paths = ["doc1.pdf", "doc2.pdf", "doc3.pdf"] processor = AsyncPDFProcessor() results = await processor.process_multiple_pdfs(pdf_paths) for result in results: if not isinstance(result, Exception): print(f"处理完成: {result['file_path']}, 块数: {result['chunk_count']}")

7. 常见问题与解决方案

7.1 PDF解析相关问题

问题1：PDF文本提取不完整或乱序

解决方案：

• 尝试不同的PDF解析库（pdfplumber、PyMuPDF、Tika）
• 调整提取参数（x_tolerance、y_tolerance）
• 对于扫描版PDF，使用OCR技术

def improve_text_extraction(pdf_path: str): """改进文本提取质量""" import pymupdf # PyMuPDF doc = pymupdf.open(pdf_path) text = "" for page in doc: # 尝试不同的提取策略 text += page.get_text("text", sort=True) # 排序文本 text += page.get_text("words", sort=True) # 按单词排序 return text

问题2：包含图片的PDF处理

解决方案：

• 使用OCR识别图片中的文字
• 结合多个提取方法

def extract_text_with_ocr(pdf_path: str): """使用OCR提取文本""" try: import pytesseract from pdf2image import convert_from_path images = convert_from_path(pdf_path) text = "" for image in images: text += pytesseract.image_to_string(image, lang='chi_sim+eng') return text except ImportError: print("请安装pytesseract和pdf2image") return ""

7.2 LLM问答质量问题

问题1：答案不准确或幻觉

解决方案：

• 改进检索策略，增加检索数量
• 添加答案验证机制
• 使用更具体的提示模板

def create_verified_qa_prompt(): """创建带验证的提示模板""" return """ 请基于以下上下文信息回答问题。如果上下文中的信息不足以回答问题，请明确说明"根据提供的上下文，无法回答这个问题"。 上下文： {context} 问题：{question} 请确保答案严格基于上下文信息，不要添加外部知识。 答案： """

问题2：处理长文档时性能问题

解决方案：

• 使用map-reduce方法处理长文档
• 实现分级检索策略
• 优化文本分块大小

7.3 系统部署问题

问题1：内存使用过高

解决方案：

• 使用磁盘缓存替代内存存储
• 实现流式处理
• 限制同时处理的文档数量

问题2：API调用限制

解决方案：

• 实现请求限流和重试机制
• 使用本地模型减少API依赖
• 批量处理请求

import time from tenacity import retry, stop_after_attempt, wait_exponential class RateLimitedQA: def __init__(self, qa_system, requests_per_minute: int = 60): self.qa_system = qa_system self.requests_per_minute = requests_per_minute self.last_request_time = 0 self.min_interval = 60.0 / requests_per_minute @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def ask_question_with_retry(self, question: str): """带重试和限流的提问方法""" current_time = time.time() time_since_last = current_time - self.last_request_time if time_since_last < self.min_interval: time.sleep(self.min_interval - time_since_last) self.last_request_time = time.time() return self.qa_system.ask_question(question)

8. 生产环境部署建议

8.1 安全考虑

API密钥管理：

• 使用环境变量或密钥管理服务
• 定期轮换密钥
• 限制API密钥权限

数据安全：

• 敏感文档本地处理
• 传输加密
• 访问日志记录

8.2 性能监控

实现系统性能监控：

import time import logging from dataclasses import dataclass from typing import Dict, Any @dataclass class PerformanceMetrics: pdf_parsing_time: float text_processing_time: float embedding_time: float query_time: float total_tokens_used: int class MonitoringSystem: def __init__(self): self.metrics = {} def start_timer(self, operation: str): self.metrics[operation] = {'start': time.time()} def end_timer(self, operation: str): if operation in self.metrics: self.metrics[operation]['end'] = time.time() self.metrics[operation]['duration'] = ( self.metrics[operation]['end'] - self.metrics[operation]['start'] ) def log_metrics(self, pdf_path: str, question: str = None): """记录性能指标""" metrics_data = { 'pdf_path': pdf_path, 'question': question, 'timings': {op: data.get('duration', 0) for op, data in self.metrics.items()}, 'timestamp': time.time() } logging.info(f"性能指标: {metrics_data}") return metrics_data

8.3 可扩展架构

对于企业级应用，建议采用微服务架构：

前端界面 → API网关 → PDF处理服务 → 向量数据库 → LLM服务

每个服务独立部署，通过消息队列进行通信，支持水平扩展。

本文详细介绍了如何将单个PDF文件传递给LLM进行问答的完整技术方案。从PDF解析、文本处理到LLM集成，每个环节都提供了具体的代码实现和最佳实践。在实际项目中，建议根据具体需求调整参数和架构，特别是对于敏感数据，优先考虑本地模型部署方案。