Chandra OCR实战教程Python批量处理PDF→自动分类→结构化存储至Milvus向量库1. 引言为什么选择Chandra OCR如果你曾经尝试过从PDF或扫描文档中提取文字肯定遇到过这些问题表格提取后格式全乱、数学公式变成乱码、排版信息完全丢失。传统的OCR工具只能给你一堆杂乱无章的文本后续还需要大量人工整理。Chandra OCR彻底改变了这种情况。这个开源模型不仅能准确识别文字还能完整保留文档的排版结构——表格保持表格的样子公式保持公式的格式甚至连复选框状态都能识别。更棒的是它可以直接输出结构化的Markdown、HTML或JSON格式让你的文档数字化一步到位。本教程将手把手教你如何用Python搭建一个完整的文档处理流水线从批量处理PDF文档到自动分类内容最后将结构化数据存储到Milvus向量数据库中为后续的智能检索和分析做好准备。2. 环境准备与快速部署2.1 系统要求与依赖安装首先确保你的系统满足以下要求Python 3.8或更高版本至少4GB显存RTX 3060或同等性能显卡8GB以上内存50GB可用磁盘空间用于模型权重和临时文件安装必要的Python包pip install chandra-ocr milvus pymilvus python-multipart pdf2image pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu1182.2 快速启动Chandra OCR服务Chandra提供了多种使用方式我们选择最方便的vLLM后端# 拉取官方Docker镜像 docker pull datalabto/chandra-ocr-vllm:latest # 启动服务假设使用两张GPU卡 docker run -d --gpus all -p 8000:8000 \ -e NVIDIA_VISIBLE_DEVICES0,1 \ datalabto/chandra-ocr-vllm:latest服务启动后你可以在浏览器中访问http://localhost:8000/docs查看API文档。3. 核心功能与基础概念3.1 Chandra的核心能力Chandra不同于传统OCR工具它具有以下独特优势布局感知能识别文档中的标题、段落、表格、图片等元素多格式输出同时生成Markdown、HTML和JSON三种格式复杂元素处理完美支持表格、数学公式、手写文字、复选框等多语言支持对中英日韩德法西等40多种语言有优秀表现3.2 输出格式解析Chandra的JSON输出包含了丰富的结构化信息{ metadata: {width: 1240, height: 1754}, blocks: [ { type: heading, text: 文档标题, bbox: [100, 120, 400, 150], confidence: 0.98 }, { type: table, html: table.../table, markdown: | 列1 | 列2 |\n|-----|-----|, bbox: [80, 200, 500, 300] } ] }这种结构化数据为我们后续的分类和存储提供了极大便利。4. 批量处理PDF文档4.1 PDF转图片预处理由于Chandra处理的是图像输入我们需要先将PDF转换为图片from pdf2image import convert_from_path import os def pdf_to_images(pdf_path, output_dir): 将PDF转换为图片 os.makedirs(output_dir, exist_okTrue) images convert_from_path(pdf_path, dpi300) image_paths [] for i, image in enumerate(images): image_path os.path.join(output_dir, fpage_{i1:03d}.jpg) image.save(image_path, JPEG, quality95) image_paths.append(image_path) return image_paths4.2 批量调用Chandra OCR编写一个批量处理函数来处理整个目录的PDF文件import requests import json from tqdm import tqdm def process_pdf_with_chandra(pdf_path, output_dir, api_urlhttp://localhost:8000): 使用Chandra处理PDF文件 # 转换PDF为图片 image_dir os.path.join(output_dir, images) image_paths pdf_to_images(pdf_path, image_dir) results [] for image_path in tqdm(image_paths, desc处理PDF页面): with open(image_path, rb) as f: files {image: f} response requests.post(f{api_url}/ocr, filesfiles) if response.status_code 200: result response.json() results.append(result) # 保存结果 base_name os.path.splitext(os.path.basename(image_path))[0] with open(os.path.join(output_dir, f{base_name}.json), w) as f: json.dump(result, f, ensure_asciiFalse, indent2) else: print(f处理失败: {image_path}, 错误: {response.text}) return results5. 自动分类与内容分析5.1 基于内容的文档分类利用Chandra输出的结构化信息我们可以实现智能文档分类from typing import List, Dict import re class DocumentClassifier: def classify_document(self, ocr_results: List[Dict]) - str: 根据OCR结果分类文档 full_text self._extract_text(ocr_results) # 基于关键词和特征的分类逻辑 if self._is_contract(full_text): return contract elif self._is_academic_paper(full_text): return academic elif self._is_form(full_text): return form elif self._is_report(full_text): return report else: return general def _extract_text(self, ocr_results): 提取所有文本内容 text_parts [] for result in ocr_results: for block in result.get(blocks, []): if text in block: text_parts.append(block[text]) return .join(text_parts) def _is_contract(self, text): 判断是否为合同文档 contract_keywords [合同, 协议, 条款, 甲方, 乙方, contract, agreement] return any(keyword in text.lower() for keyword in contract_keywords) def _is_academic_paper(self, text): 判断是否为学术论文 academic_indicators [摘要, 参考文献, 引用, abstract, reference] math_patterns [r\\[a-zA-Z]\{, r\$\$[^$]\$\$] # LaTeX公式 has_academic_words any(indicator in text for indicator in academic_indicators) has_math any(re.search(pattern, text) for pattern in math_patterns) return has_academic_words or has_math def _is_form(self, text): 判断是否为表单 form_indicators [□, ☑, checkbox, 选择, 填写] return any(indicator in text for indicator in form_indicators) def _is_report(self, text): 判断是否为报告 report_indicators [报告, 分析, 总结, report, analysis] has_tables any(table in block.get(type, ) for result in ocr_results for block in result.get(blocks, [])) return any(indicator in text for indicator in report_indicators) and has_tables5.2 关键信息提取针对不同类型的文档提取关键信息class InformationExtractor: def extract_key_info(self, ocr_results, doc_type): 根据文档类型提取关键信息 extractors { contract: self._extract_contract_info, academic: self._extract_academic_info, form: self._extract_form_info, report: self._extract_report_info, general: self._extract_general_info } return extractors.get(doc_type, self._extract_general_info)(ocr_results) def _extract_contract_info(self, ocr_results): 提取合同关键信息 full_text self._extract_full_text(ocr_results) # 提取合同双方、日期、金额等信息 info { parties: self._extract_contract_parties(full_text), effective_date: self._extract_date(full_text), amount: self._extract_monetary_amount(full_text), key_terms: self._extract_key_terms(full_text) } return info def _extract_academic_info(self, ocr_results): 提取学术论文信息 info { title: self._find_title(ocr_results), authors: self._find_authors(ocr_results), abstract: self._find_abstract(ocr_results), references: self._find_references(ocr_results) } return info6. 存储到Milvus向量数据库6.1 Milvus环境搭建首先确保Milvus服务正常运行# 使用Docker启动Milvus docker run -d --name milvus_standalone \ -p 19530:19530 \ -p 9091:9091 \ milvusdb/milvus:latest6.2 创建向量集合根据文档内容特点设计Milvus集合结构from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility class MilvusManager: def __init__(self, hostlocalhost, port19530): self.host host self.port port self.connect() def connect(self): 连接Milvus数据库 connections.connect(aliasdefault, hostself.host, portself.port) def create_collection(self, collection_namedocuments): 创建文档集合 # 定义字段 fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namedoc_id, dtypeDataType.VARCHAR, max_length256), FieldSchema(nametitle, dtypeDataType.VARCHAR, max_length512), FieldSchema(namecontent, dtypeDataType.VARCHAR, max_length65535), FieldSchema(namedoc_type, dtypeDataType.VARCHAR, max_length64), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dim768), FieldSchema(namemetadata, dtypeDataType.JSON), FieldSchema(namecreate_time, dtypeDataType.INT64) ] # 创建集合 schema CollectionSchema(fields, descriptionDocument storage collection) collection Collection(namecollection_name, schemaschema) # 创建索引 index_params { index_type: IVF_FLAT, metric_type: L2, params: {nlist: 128} } collection.create_index(field_nameembedding, index_paramsindex_params) return collection6.3 文档向量化与存储将处理后的文档内容转换为向量并存储import numpy as np from sentence_transformers import SentenceTransformer import time class DocumentStorage: def __init__(self): self.milvus_manager MilvusManager() self.collection self.milvus_manager.create_collection() self.embedding_model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) def store_document(self, doc_id, title, content, doc_type, metadata): 存储文档到Milvus # 生成内容向量 embedding self.embedding_model.encode(content).tolist() # 准备插入数据 data [ [doc_id], # doc_id [title], # title [content], # content [doc_type], # doc_type [embedding], # embedding [metadata], # metadata [int(time.time())] # create_time ] # 插入数据 self.collection.insert(data) self.collection.flush() print(f文档 {doc_id} 存储成功) def batch_store_documents(self, documents): 批量存储文档 for doc in documents: self.store_document(**doc)7. 完整流水线实现7.1 端到端处理流程将各个模块组合成完整的处理流水线import glob from pathlib import Path class DocumentProcessingPipeline: def __init__(self): self.classifier DocumentClassifier() self.extractor InformationExtractor() self.storage DocumentStorage() def process_directory(self, input_dir, output_base_dir): 处理整个目录的PDF文件 pdf_files glob.glob(os.path.join(input_dir, *.pdf)) for pdf_path in pdf_files: try: self.process_single_pdf(pdf_path, output_base_dir) except Exception as e: print(f处理文件 {pdf_path} 时出错: {str(e)}) def process_single_pdf(self, pdf_path, output_base_dir): 处理单个PDF文件 pdf_name Path(pdf_path).stem output_dir os.path.join(output_base_dir, pdf_name) os.makedirs(output_dir, exist_okTrue) print(f开始处理: {pdf_name}) # 1. OCR处理 ocr_results process_pdf_with_chandra(pdf_path, output_dir) # 2. 文档分类 doc_type self.classifier.classify_document(ocr_results) print(f文档分类: {doc_type}) # 3. 信息提取 key_info self.extractor.extract_key_info(ocr_results, doc_type) # 4. 准备存储数据 full_text self._combine_ocr_results(ocr_results) document_data { doc_id: pdf_name, title: self._extract_title(ocr_results) or pdf_name, content: full_text, doc_type: doc_type, metadata: { key_info: key_info, page_count: len(ocr_results), processing_time: time.time(), original_path: pdf_path } } # 5. 存储到Milvus self.storage.store_document(**document_data) print(f完成处理: {pdf_name}) def _combine_ocr_results(self, ocr_results): 合并所有OCR结果 texts [] for result in ocr_results: for block in result.get(blocks, []): if text in block: texts.append(block[text]) return \n.join(texts) def _extract_title(self, ocr_results): 提取文档标题 for result in ocr_results: for block in result.get(blocks, []): if block.get(type) heading and block.get(text): return block[text] return None7.2 使用示例if __name__ __main__: pipeline DocumentProcessingPipeline() # 处理单个文件 pipeline.process_single_pdf(合同文档.pdf, ./output) # 或者处理整个目录 pipeline.process_directory(./pdf_documents, ./output)8. 总结与进阶建议通过本教程你已经学会了如何搭建一个完整的文档智能处理流水线。这个系统能够自动处理PDF文档提取结构化信息智能分类内容并存储到向量数据库中以供后续检索和分析。8.1 核心价值总结自动化处理无需人工干预批量处理大量文档智能分类基于内容特征自动识别文档类型结构化存储保留文档的完整结构和语义信息检索友好向量化存储支持相似性搜索和语义检索8.2 进阶优化建议如果你想要进一步提升系统性能可以考虑以下方向性能优化使用异步处理提高并发性能实现分布式处理支持大规模文档库添加缓存机制减少重复处理功能扩展集成更多文档类型支持Word、Excel等添加版本管理和变更追踪实现实时处理和工作流管理智能增强添加自定义分类规则训练实现更精细的信息提取模型集成问答和摘要生成功能这个流水线为构建企业级文档管理系统提供了坚实基础你可以根据具体需求进行定制和扩展。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。