1. 项目概述最近在做一个挺有意思的项目用Chainlit、Langchain和Elasticsearch搭建了一个PDF文档查询系统。这个组合真的让我眼前一亮特别是当你手头有一堆技术文档、论文或者报告需要快速查找特定内容时传统的关键词搜索往往不够精准。这套方案不仅能实现全文检索还能理解查询意图返回更相关的结果。我最初是在处理公司积累的几百份产品文档时想到这个方案的。每次要找某个功能的详细说明都得一个个PDF翻效率实在太低。现在用这个系统输入自然语言问题就能直接定位到相关内容开发效率提升了不少。下面我就把这个方案的实现细节完整分享出来包括一些踩过的坑和优化技巧。2. 技术栈解析2.1 Chainlit的核心作用Chainlit在这个方案里扮演着文档预处理的关键角色。它最厉害的地方在于能处理PDF这种非结构化数据——不仅能提取普通文本还能识别文档中的表格、图表标注等元素。我测试过几种常见的Python PDF库发现Chainlit在保持原文结构方面做得最好。安装很简单pip install chainlit但实际使用时有个重要细节Chainlit默认使用PDFMiner作为后端对于复杂版式的文档建议改用pdf2textimport chainlit as cl cl.set_pdf_backend(pdf2text) # 显著提升表格提取准确率2.2 Langchain的NLP能力Langchain是这个系统的大脑负责理解查询意图和文档语义。它的TextSplitter功能对长文档处理特别有用from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200, # 防止关键信息被切断 length_functionlen )我推荐使用OpenAI的embeddingstext-embedding-3-small性价比很高但如果对数据隐私要求高HuggingFace的all-MiniLM-L6-v2也是不错的选择from langchain.embeddings import OpenAIEmbeddings embeddings OpenAIEmbeddings( modeltext-embedding-3-small, chunk_size512 # 控制API请求大小 )2.3 Elasticsearch的搜索优势Elasticsearch 8.x版本在语义搜索方面有很大提升。安装时建议直接用Dockerdocker run -p 9200:9200 -p 9300:9300 -e discovery.typesingle-node docker.elastic.co/elasticsearch/elasticsearch:8.12.0创建索引时要特别注意mapping设置{ mappings: { properties: { content: { type: text, analyzer: ik_max_word // 中文分词 }, embedding: { type: dense_vector, dims: 1536 // 匹配embeddings维度 } } } }3. 完整实现流程3.1 文档预处理流水线我设计了一个带错误重试机制的预处理流程def process_pdf(file_path): retry 3 while retry 0: try: doc cl.process_pdf(file_path) texts splitter.split_text(doc.text) return [{ page: doc.pages[i], text: text, metadata: { source: file_path, page_num: i1 } } for i, text in enumerate(texts)] except Exception as e: retry - 1 time.sleep(1) raise Exception(fFailed to process {file_path})重要提示对于扫描版PDF先用OCR工具转换效果会更好。我测试过Tesseract和PaddleOCR后者对中文支持更好。3.2 向量化与索引构建批量处理时要注意API速率限制from tqdm import tqdm def batch_embed(texts, batch_size100): for i in tqdm(range(0, len(texts), batch_size)): batch texts[i:ibatch_size] embeddings embedder.embed_documents(batch) yield [ { content: text, embedding: embedding, metadata: texts[ij][metadata] } for j, (text, embedding) in enumerate(zip(batch, embeddings)) ]Elasticsearch的bulk API使用技巧from elasticsearch.helpers import parallel_bulk def index_documents(docs): actions ({ _op_type: index, _index: pdf_docs, _source: doc } for doc in docs) for success, info in parallel_bulk(es, actions, thread_count4): if not success: logger.error(fFailed to index: {info})3.3 查询接口实现混合搜索关键词语义的实现def hybrid_search(query, top_k5): # 语义搜索 query_embedding embedder.embed_query(query) semantic_query { script_score: { query: {match_all: {}}, script: { source: cosineSimilarity(params.query_vector, embedding) 1.0, params: {query_vector: query_embedding} } } } # 关键词搜索 keyword_query { multi_match: { query: query, fields: [content] } } response es.search( indexpdf_docs, query{ bool: { should: [ {weight: 0.7, query: semantic_query}, {weight: 0.3, query: keyword_query} ] } }, sizetop_k ) return [hit[_source] for hit in response[hits][hits]]4. 性能优化技巧4.1 索引优化分片策略根据数据量调整PUT /pdf_docs/_settings { index.number_of_replicas: 1, index.number_of_shards: 3 # 适合10GB以下数据 }字段映射优化{ content: { type: text, fields: { keyword: { type: keyword, ignore_above: 256 } } } }4.2 查询优化使用search_after分页替代from/sizedef paginated_search(query, last_sortNone): body { query: {match: {content: query}}, size: 50, sort: [{_score: desc}, {_id: asc}] } if last_sort: body[search_after] last_sort return es.search(indexpdf_docs, bodybody)预计算常见查询from functools import lru_cache lru_cache(maxsize1000) def cached_search(query): return hybrid_search(query)5. 常见问题排查5.1 中文分词问题症状搜索中文词无结果 解决方案安装IK分词插件elasticsearch-plugin install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v8.12.0/elasticsearch-analysis-ik-8.12.0.zip创建自定义词典echo 机器学习 config/analysis-ik/custom.dic5.2 内存溢出处理当处理大PDF时可能遇到调整Chainlit内存限制cl.set_memory_limit(4096) # 单位MBElasticsearch JVM设置ES_JAVA_OPTS-Xms4g -Xmx4g ./bin/elasticsearch5.3 精度问题症状语义搜索相关度低 检查点embeddings维度是否匹配索引设置归一化处理import numpy as np def normalize_embedding(embedding): return embedding / np.linalg.norm(embedding)6. 进阶扩展方向多文档关联分析from langchain.chains import QAGenerationChain qa_chain QAGenerationChain.from_llm(llm) cross_doc_qa qa_chain.run(documentsrelated_docs)自动摘要生成from langchain.chains.summarize import load_summarize_chain summary_chain load_summarize_chain(llm, chain_typemap_reduce) document_summary summary_chain.run(split_documents)可视化分析import matplotlib.pyplot as plt from sklearn.manifold import TSNE def plot_embeddings(embeddings): tsne TSNE(n_components2) vis tsne.fit_transform(embeddings) plt.scatter(vis[:, 0], vis[:, 1]) plt.show()这套系统我已经在生产环境跑了半年多处理过10万页的PDF文档。最大的体会是一定要做好文档预处理的质量监控建议对每个PDF提取后做人工抽样检查。另外Elasticsearch的定期优化forcemerge、cache清理等对维持查询性能非常关键。