下面是一个基于 FastAPI 框架使用 MinerU 和本地 Qwen 模型实现文档处理全流程的示例代码。这个方案实现了“文档上传、内容解析、文本分块、向量化与存储ES”的功能闭环。️ 技术架构说明该方案的组件选择和连接方式如下Web框架 (FastAPI)用于接收文档上传请求和管理API。文档解析 (MinerU)核心解析引擎将PDF、Word等文档转换为包含表格、公式的结构化Markdown/JSON数据。嵌入模型 (本地Qwen)使用阿里巴巴的Qwen系列模型通过sentence-transformers在本地将文本块转换为向量。向量数据库 (Elasticsearch)作为统一的存储和检索引擎存储向量、原始文本和元数据并支持高效的向量相似度搜索。 环境准备首先确保已安装以下核心依赖并启动了Elasticsearch服务。pipinstallfastapi uvicorn python-multipart mineru-next-dev elasticsearch sentence-transformers注意根据你的硬件环境sentence-transformers会自动选择CPU或GPU运行。Qwen模型文件较大首次运行时会自动下载。 完整代码实现 (main.py)importosimportuuidimportshutilfromtypingimportListfromfastapiimportFastAPI,File,UploadFile,HTTPExceptionfrompydanticimportBaseModel# MinerU 相关frommineruimportMinerUfrommineru.utilsimportMinerUConfig# 文本分块fromlangchain_text_splittersimportRecursiveCharacterTextSplitter# 本地Embedding模型fromsentence_transformersimportSentenceTransformer# Elasticsearch 向量存储fromelasticsearchimportElasticsearch,helpers appFastAPI(titleRAG Knowledge Base API)# 1. 配置与初始化 # 存储路径UPLOAD_DIR./uploaded_filesOUTPUT_DIR./parsed_outputos.makedirs(UPLOAD_DIR,exist_okTrue)os.makedirs(OUTPUT_DIR,exist_okTrue)# 初始化MinerUminer_configMinerUConfig()minerMinerU(configminer_config)# 初始化本地Embedding模型 (以Qwen系列为例选择合适的模型)# 可选: Qwen/Qwen-1_8B-Embedding, Qwen/Qwen-7B-Embedding 等embedding_model_nameQwen/Qwen-1_8B-Embeddingprint(f正在加载本地Embedding模型:{embedding_model_name}...)embedding_modelSentenceTransformer(embedding_model_name)print(模型加载完成。)# 初始化文本分块器text_splitterRecursiveCharacterTextSplitter(chunk_size800,chunk_overlap150,separators[\n\n,\n,。,,,,, ,],)# 初始化Elasticsearch客户端ES_URLhttp://localhost:9200# 请替换为你的ES地址ES_INDEX_NAMEknowledge_basees_clientElasticsearch(ES_URL)# 检查并创建索引如果不存在ifnotes_client.indices.exists(indexES_INDEX_NAME):# 定义索引映射包含向量字段mapping{mappings:{properties:{text:{type:text},metadata:{type:object,enabled:True},embedding:{type:dense_vector,dims:1536,# Qwen-1.8B-Embedding 的向量维度index:True,similarity:cosine,}}}}es_client.indices.create(indexES_INDEX_NAME,bodymapping)print(fES索引 {ES_INDEX_NAME} 创建成功。)else:print(fES索引 {ES_INDEX_NAME} 已存在。)# 2. 核心处理函数 defparse_document(file_path:str)-List[str]: 使用MinerU解析文档返回文本块列表。 # 设置输出路径output_pathos.path.join(OUTPUT_DIR,f{uuid.uuid4()}.md)# 调用MinerU进行解析# miner.parse 方法会解析文档并返回结果这里假设它生成Markdown文件# 实际API可能略有不同请参考MinerU官方文档miner.parse(file_path,output_pathoutput_path)# 读取解析后的Markdown内容withopen(output_path,r,encodingutf-8)asf:raw_textf.read()# 清理文本并分块# 注意MinerU 输出的是结构良好的Markdown直接分块即可docs[{page_content:raw_text,metadata:{source:file_path}}]chunkstext_splitter.split_documents(docs)# 返回文本内容列表和元数据return[{text:chunk.page_content,metadata:chunk.metadata}forchunkinchunks]defstore_in_es(chunks:List[dict]): 将文本块向量化后批量存入Elasticsearch。 actions[]forchunkinchunks:textchunk[text]metadatachunk[metadata]# 生成向量embeddingembedding_model.encode(text).tolist()# 准备ES文档action{_index:ES_INDEX_NAME,_source:{text:text,metadata:metadata,embedding:embedding,}}actions.append(action)# 批量写入ESifactions:success,failedhelpers.bulk(es_client,actions,stats_onlyTrue)print(f成功写入{success}条记录, 失败{failed}条。)returnsuccessreturn0# 3. API 接口 classProcessResponse(BaseModel):request_id:strfile_name:strchunk_count:intstatus:strmessage:strapp.post(/upload/,response_modelProcessResponse)asyncdefupload_and_process(file:UploadFileFile(...)): 上传文档并自动完成解析、分块、向量化和存储。 # 生成请求IDrequest_idstr(uuid.uuid4())# 1. 保存上传文件file_pathos.path.join(UPLOAD_DIR,f{request_id}_{file.filename})try:withopen(file_path,wb)asbuffer:shutil.copyfileobj(file.file,buffer)exceptExceptionase:raiseHTTPException(status_code500,detailf文件保存失败:{str(e)})# 2. 调用MinerU进行解析和分块try:chunksparse_document(file_path)exceptExceptionase:# 清理已上传文件os.remove(file_path)raiseHTTPException(status_code500,detailf文档解析失败:{str(e)})ifnotchunks:os.remove(file_path)raiseHTTPException(status_code400,detail文档解析后未产生任何文本块。)# 3. 向量化并存入EStry:stored_countstore_in_es(chunks)exceptExceptionase:# 即使存储失败也保留解析结果供排查raiseHTTPException(status_code500,detailf数据存储失败:{str(e)})# 清理临时文件可选# os.remove(file_path)returnProcessResponse(request_idrequest_id,file_namefile.filename,chunk_countstored_count,statussuccess,message文档处理完成已存入知识库。)# 4. 启动脚本 if__name____main__:importuvicorn uvicorn.run(app,host0.0.0.0,port8000) 关键实现要点MinerU 集成代码中通过miner.parse(file_path, output_pathoutput_path)调用MinerU进行解析。MinerU支持多种格式PDF、DOCX、PPTX等并会输出结构化的Markdown保留了表格、标题等关键信息。本地Embedding模型选择在初始化SentenceTransformer时你可以选择不同的Qwen模型。Qwen/Qwen-1_8B-Embedding是一个轻量级选择适合在消费级GPU上运行如果硬件允许Qwen/Qwen-7B-Embedding能提供更好的向量表示质量。注意需要根据选择的模型在ES索引映射中正确设置dims向量维度。Elasticsearch索引配置代码中为embedding字段设置了dense_vector类型并指定了相似度算法为cosine。这为后续进行语义检索奠定了基础。生产环境中你可以根据数据量级调整分片和副本数。 运行与测试启动服务python main.py服务将在http://127.0.0.1:8000启动自动生成的API文档在http://127.0.0.1:8000/docs。测试API使用curl或Postman发送POST请求curl-XPOST-Ffile/path/to/your/document.pdfhttp://127.0.0.1:8000/upload/成功后会返回处理结果包括存储的文本块数量。这个示例提供了一个可直接运行的起点。并可以根据实际的文档类型如扫描件、复杂表格调整MinerU的解析参数或根据业务需求优化分块策略。部署时建议参考MinerU官方文档进行更详细的配置例如使用precision模式处理长文档。