AI文档智能与数据提取完全教程
全面讲解AI文档智能与数据提取的核心技术，从OCR识别到LLM结构化提取，帮助开发者构建智能文档处理系统。
目录
- 文档智能技术全景
- OCR技术演进与实战
- PDF解析与版面分析
- 表格识别与结构化提取
- 多模态文档理解
- LLM结构化数据提取
- 发票/合同/报告场景实战
- 文档分类与信息抽取Pipeline
- 文档Chunking策略
- 多语言文档处理
- 文档质量评估与纠错
- 批量文档处理工作流
- 文档智能API服务搭建
- 最佳实践与总结
1. 文档智能技术全景
1.1 什么是文档智能
文档智能（Document Intelligence）是人工智能的一个重要分支，专注于让机器理解、解析和提取文档中的信息。它结合了计算机视觉、自然语言处理和机器学习等多种技术，能够自动处理各类文档，包括扫描件、PDF、图片、表格等。
文档智能的核心目标是将非结构化或半结构化的文档数据转化为结构化信息，从而实现自动化处理、信息检索和知识管理。
1.2 技术架构全景
一个完整的文档智能系统通常包含以下层次：
┌─────────────────────────────────────────────┐
│ 应用层 (Application) │
│ 发票处理 | 合同审查 | 报告分析 | 知识管理 │
├─────────────────────────────────────────────┤
│ 理解层 (Understanding) │
│ 文档分类 | 实体提取 | 关系抽取 | 语义理解 │
├─────────────────────────────────────────────┤
│ 解析层 (Parsing) │
│ 版面分析 | 表格识别 | 公式检测 | 图表解析 │
├─────────────────────────────────────────────┤
│ 识别层 (Recognition) │
│ OCR | 手写识别 | 印章检测 | 签名识别 │
├─────────────────────────────────────────────┤
│ 预处理层 (Preprocessing) │
│ 图像增强 | 去噪 | 矫正 | 分割 │
└─────────────────────────────────────────────┘
1.3 核心技术栈概览
| 技术领域 | 代表工具/框架 | 主要用途 |
|---|---|---|
| OCR引擎 | PaddleOCR, EasyOCR, Surya, Tesseract | 文字识别 |
| PDF解析 | PyMuPDF, pdfplumber, Unstructured | 文档解析 |
| 版面分析 | LayoutParser, Detectron2, Surya | 文档结构理解 |
| 表格识别 | TableTransformer, img2table | 表格提取 |
| 多模态理解 | GPT-4V, Qwen-VL, InternVL | 文档理解 |
| 结构化提取 | Instructor, LangChain, LlamaIndex | 信息提取 |
| NER/RE | spaCy, Hugging Face Transformers | 命名实体识别 |
2. OCR技术演进与实战
2.1 OCR技术发展简史
OCR（Optical Character Recognition，光学字符识别）技术经历了三个主要发展阶段：
- 传统方法（1990s-2010s）：基于模板匹配和特征工程，代表工具如Tesseract
- 深度学习方法（2015-2020）：基于CNN和RNN的端到端识别，如CRNN
- 大模型方法（2020-至今）：基于Transformer和多模态预训练，如Surya
2.2 PaddleOCR实战
PaddleOCR是百度开源的超轻量级OCR工具库，支持80+语言识别，在中文场景表现尤为出色。
# 安装: pip install paddlepaddle paddleocr
from paddleocr import PaddleOCR
import cv2
# 初始化OCR引擎
# lang='ch' 支持中英文混合识别
ocr = PaddleOCR(
use_angle_cls=True, # 启用方向分类
lang='ch', # 中文模型
use_gpu=True, # 使用GPU加速
det_model_dir=None, # 使用默认检测模型
rec_model_dir=None # 使用默认识别模型
)
# 对图片进行OCR识别
img_path = 'document_scan.jpg'
result = ocr.ocr(img_path, cls=True)
# 解析结果
for line in result[0]:
bbox = line[0] # 文字边界框坐标 [[x1,y1],[x2,y2],[x3,y3],[x4,y4]]
text = line[1][0] # 识别的文字内容
confidence = line[1][1] # 置信度分数
print(f"文字: {text}")
print(f"置信度: {confidence:.4f}")
print(f"位置: {bbox}")
print("-" * 50)
# 批量处理多张图片
def batch_ocr(image_paths):
all_results = {}
for path in image_paths:
result = ocr.ocr(path, cls=True)
texts = []
for line in result[0]:
texts.append({
'text': line[1][0],
'confidence': line[1][1],
'bbox': line[0]
})
all_results[path] = texts
return all_results
# 可视化OCR结果
def visualize_ocr(img_path, result):
img = cv2.imread(img_path)
for line in result[0]:
bbox = line[0]
text = line[1][0]
# 绘制边界框
pts = [(int(p[0]), int(p[1])) for p in bbox]
cv2.polylines(img, [np.array(pts)], True, (0, 255, 0), 2)
# 标注文字
cv2.putText(img, text, (pts[0][0], pts[0][1] - 10),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 1)
cv2.imwrite('ocr_result.jpg', img)
2.3 EasyOCR实战
EasyOCR是一个基于PyTorch的OCR库，支持80+语言，API设计简洁。
# 安装: pip install easyocr
import easyocr
# 初始化Reader，支持多语言
reader = easyocr.Reader(['ch_sim', 'en'], gpu=True)
# 识别图片中的文字
results = reader.readtext('receipt.jpg')
for (bbox, text, prob) in results:
print(f"文字: {text}")
print(f"置信度: {prob:.4f}")
print(f"位置: {bbox}")
print("-" * 50)
# 仅返回文字列表（适合简单场景）
texts = [text for _, text, _ in results]
full_text = '\n'.join(texts)
print(full_text)
# 对PDF页面进行OCR
import fitz # PyMuPDF
def ocr_pdf_page(pdf_path, page_num=0):
doc = fitz.open(pdf_path)
page = doc[page_num]
# 将PDF页面渲染为图片
pix = page.get_pixmap(dpi=300)
img_path = f'page_{page_num}.png'
pix.save(img_path)
# 对图片进行OCR
results = reader.readtext(img_path)
return results
2.4 Surya OCR实战
Surya是一个全新的开源OCR工具，基于Transformer架构，在复杂文档场景下表现优异。
# 安装: pip install surya-ocr
from surya.ocr import run_ocr
from surya.model.detection.model import load_model as load_det_model
from surya.model.detection.processor import load_processor as load_det_processor
from surya.model.recognition.model import load_model as load_rec_model
from surya.model.recognition.processor import load_processor as load_rec_processor
from PIL import Image
# 加载模型
det_model = load_det_model()
det_processor = load_det_processor()
rec_model = load_rec_model()
rec_processor = load_rec_processor()
# 读取图片
image = Image.open('complex_document.jpg')
# 运行OCR
results = run_ocr(
[image],
[['zh', 'en']], # 语言列表
det_model, det_processor,
rec_model, rec_processor
)
# 输出结果
for text_line in results[0].text_lines:
print(f"文字: {text_line.text}")
print(f"置信度: {text_line.confidence:.4f}")
2.5 OCR引擎选择指南
def choose_ocr_engine(requirements):
"""
根据需求选择最佳OCR引擎
Args:
requirements: dict，包含场景需求
Returns:
推荐的OCR引擎名称
"""
recommendations = {
'chinese_heavy': 'PaddleOCR', # 中文为主场景
'multilingual': 'EasyOCR', # 多语言混合场景
'complex_layout': 'Surya', # 复杂版面场景
'speed_priority': 'PaddleOCR', # 速度优先
'accuracy_priority': 'Surya', # 精度优先
'lightweight': 'PaddleOCR', # 轻量部署
'printed_only': 'Tesseract', # 纯印刷体
'handwriting': 'Surya', # 手写识别
}
scenario = requirements.get('scenario', 'general')
return recommendations.get(scenario, 'PaddleOCR')
3. PDF解析与版面分析
3.1 PDF文档类型
PDF文档可分为三大类：
- 文本型PDF：包含可选择的文字层，直接提取文本即可
- 扫描型PDF：本质是图片，需要OCR才能获取文字
- 混合型PDF：部分区域有文字层，部分为扫描图片
3.2 PyMuPDF（fitz）实战
PyMuPDF是功能强大的PDF处理库，支持文本提取、图片提取、页面渲染等。
# 安装: pip install PyMuPDF
import fitz # PyMuPDF
def extract_pdf_content(pdf_path):
"""提取PDF文档的全部内容"""
doc = fitz.open(pdf_path)
content = {
'metadata': doc.metadata,
'page_count': len(doc),
'pages': []
}
for page_num in range(len(doc)):
page = doc[page_num]
# 提取文字
text = page.get_text("text")
# 提取文字块（带位置信息）
blocks = page.get_text("blocks")
# 提取图片
images = []
for img_index, img in enumerate(page.get_images(full=True)):
xref = img[0]
base_image = doc.extract_image(xref)
images.append({
'index': img_index,
'width': base_image['width'],
'height': base_image['height'],
'ext': base_image['ext'],
'data': base_image['image']
})
# 提取链接
links = page.get_links()
# 提取表格
tables = page.find_tables()
content['pages'].append({
'page_num': page_num,
'text': text,
'blocks': blocks,
'images': images,
'links': links,
'table_count': len(tables.tables) if tables else 0
})
doc.close()
return content
def extract_tables_from_pdf(pdf_path):
"""提取PDF中的所有表格"""
doc = fitz.open(pdf_path)
all_tables = []
for page_num in range(len(doc)):
page = doc[page_num]
tables = page.find_tables()
for table_idx, table in enumerate(tables.tables):
# 转换为pandas DataFrame
df = table.to_pandas()
all_tables.append({
'page': page_num,
'table_index': table_idx,
'rows': len(df),
'cols': len(df.columns),
'dataframe': df
})
doc.close()
return all_tables
# 使用示例
content = extract_pdf_content('report.pdf')
print(f"文档共 {content['page_count']} 页")
for page in content['pages']:
print(f"第 {page['page_num']+1} 页: {len(page['text'])} 字符, "
f"{len(page['images'])} 张图片, {page['table_count']} 个表格")
3.3 pdfplumber实战
pdfplumber在表格提取方面表现尤为出色，能够精确提取表格结构。
# 安装: pip install pdfplumber
import pdfplumber
import pandas as pd
def extract_with_pdfplumber(pdf_path):
"""使用pdfplumber提取PDF内容"""
with pdfplumber.open(pdf_path) as pdf:
for page_num, page in enumerate(pdf.pages):
# 提取文字
text = page.extract_text()
# 提取表格
tables = page.extract_tables()
for table_idx, table in enumerate(tables):
# 转为DataFrame
df = pd.DataFrame(table[1:], columns=table[0])
print(f"第{page_num+1}页 表格{table_idx+1}:")
print(df.to_string())
# 提取单词（带坐标）
words = page.extract_words(
x_tolerance=3,
y_tolerance=3,
keep_blank_chars=True,
extra_attrs=["fontname", "size"]
)
# 提取字符级别信息
chars = page.chars
yield {
'page_num': page_num,
'text': text,
'tables': tables,
'words': words,
'char_count': len(chars)
}
def smart_table_extraction(pdf_path):
"""智能表格提取，自动检测和解析"""
with pdfplumber.open(pdf_path) as pdf:
all_dataframes = []
for page in pdf.pages:
# 使用自定义设置提取表格
table_settings = {
"vertical_strategy": "lines",
"horizontal_strategy": "lines",
"min_words_vertical": 2,
"min_words_horizontal": 2,
}
tables = page.extract_tables(table_settings)
for table in tables:
if table and len(table) > 1:
# 清理数据
cleaned = []
for row in table:
cleaned_row = [
cell.strip().replace('\n', ' ') if cell else ''
for cell in row
]
cleaned_row = [c for c in cleaned_row if any(c)]
if cleaned_row:
cleaned.append(cleaned_row)
if cleaned:
df = pd.DataFrame(cleaned[1:], columns=cleaned[0])
all_dataframes.append(df)
return all_dataframes
3.4 Unstructured框架
Unstructured是一个统一的文档解析框架，支持多种文档格式的自动化处理。
# 安装: pip install unstructured
from unstructured.partition.auto import partition
from unstructured.partition.pdf import partition_pdf
from unstructured.cleaners.core import clean_extra_whitespace
def parse_with_unstructured(file_path):
"""使用Unstructured解析文档"""
# 自动检测文件类型并解析
elements = partition(filename=file_path)
result = {
'text_blocks': [],
'tables': [],
'images': [],
'titles': []
}
for element in elements:
element_type = type(element).__name__
text = str(element)
if element_type == 'Title':
result['titles'].append(text)
elif element_type == 'Table':
result['tables'].append(text)
elif element_type == 'Image':
result['images'].append(text)
else:
result['text_blocks'].append({
'type': element_type,
'text': clean_extra_whitespace(text)
})
return result
def parse_pdf_with_strategy(pdf_path, strategy='hi_res'):
"""
使用不同策略解析PDF
strategy: 'fast'(快速), 'hi_res'(高精度), 'ocr_only'(纯OCR)
"""
elements = partition_pdf(
filename=strategy,
strategy=strategy,
languages=['chi_sim', 'eng'],
include_page_breaks=True,
infer_table_structure=True,
extract_images_in_pdf=True,
chunking_strategy="by_title",
max_characters=10000,
new_after_n_chars=6000,
combine_text_under_n_chars=2000,
)
return elements
# 高级用法：结构化输出
def extract_structured_content(pdf_path):
"""提取结构化内容"""
from unstructured.staging.base import convert_to_isd, elements_to_json
elements = partition_pdf(filename=pdf_path, strategy='hi_res')
# 转为JSON格式
json_output = elements_to_json(elements)
# 转为字典格式
isd = convert_to_isd(elements)
return isd
3.5 版面分析技术
版面分析是理解文档空间结构的关键步骤，能够识别标题、段落、图片、表格等区域。
# 使用LayoutParser进行版面分析
# 安装: pip install layoutparser torchvision
import layoutparser as lp
import cv2
def analyze_layout(image_path):
"""分析文档版面布局"""
# 加载预训练模型
model = lp.Detectron2LayoutModel(
config_path='lp://PubLayNet/faster_rcnn_R_50_FPN_3x/config',
extra_config=["MODEL.ROI_HEADS.SCORE_THRESH_TEST", 0.5],
label_map={0: "Text", 1: "Title", 2: "List", 3: "Table", 4: "Figure"}
)
image = cv2.imread(image_path)
image = image[:, :, ::-1] # BGR转RGB
# 检测版面元素
layout = model.detect(image)
# 按位置排序（从上到下，从左到右）
layout.sort(key=lambda x: (x.block.y_1, x.block.x_1))
results = []
for block in layout:
results.append({
'type': block.type,
'confidence': block.score,
'bbox': {
'x1': int(block.block.x_1),
'y1': int(block.block.y_1),
'x2': int(block.block.x_2),
'y2': int(block.block.y_2)
}
})
return results
def extract_region_content(image_path, bbox, ocr_engine):
"""提取特定区域的文字内容"""
image = cv2.imread(image_path)
x1, y1, x2, y2 = bbox['x1'], bbox['y1'], bbox['x2'], bbox['y2']
cropped = image[y1:y2, x1:x2]
# 对裁剪区域进行OCR
result = ocr_engine.ocr(cropped, cls=True)
text = '\n'.join([line[1][0] for line in result[0]])
return text
4. 表格识别与结构化提取
4.1 表格识别的挑战
表格识别面临的主要挑战包括：
- 有线表格：有明确的边界线，相对容易识别
- 无线表格：通过间距和对齐隐式表示，识别难度大
- 跨页表格：表格跨越多个页面，需要合并处理
- 嵌套表格：表格中包含子表格，结构复杂
- 不规则表格：合并单元格、斜线表头等
4.2 Table Transformer实战
# 使用Microsoft的Table Transformer
# 安装: pip install torch transformers timm
from transformers import AutoModelForObjectDetection, AutoProcessor
from PIL import Image
import torch
class TableExtractor:
def __init__(self):
# 加载表格检测模型
self.detection_model = AutoModelForObjectDetection.from_pretrained(
"microsoft/table-transformer-detection"
)
# 加载表格结构识别模型
self.structure_model = AutoModelForObjectDetection.from_pretrained(
"microsoft/table-structure-recognition-v1.1-all"
)
self.processor = AutoProcessor.from_pretrained(
"microsoft/table-transformer-detection"
)
def detect_tables(self, image_path):
"""检测图片中的表格区域"""
image = Image.open(image_path).convert("RGB")
inputs = self.processor(images=image, return_tensors="pt")
with torch.no_grad():
outputs = self.detection_model(**inputs)
results = self.processor.post_process_object_detection(
outputs, threshold=0.7, target_sizes=[image.size[::-1]]
)[0]
tables = []
for score, label, box in zip(results["scores"], results["labels"], results["boxes"]):
tables.append({
'bbox': box.tolist(),
'score': score.item(),
'label': self.detection_model.config.id2label[label.item()]
})
return tables
def recognize_structure(self, table_image):
"""识别表格内部结构（行、列、单元格）"""
inputs = self.processor(images=table_image, return_tensors="pt")
with torch.no_grad():
outputs = self.structure_model(**inputs)
results = self.processor.post_process_object_detection(
outputs, threshold=0.7, target_sizes=[table_image.size[::-1]]
)[0]
elements = []
for score, label, box in zip(results["scores"], results["labels"], results["boxes"]):
elements.append({
'type': self.structure_model.config.id2label[label.item()],
'bbox': box.tolist(),
'confidence': score.item()
})
return elements
# 使用示例
extractor = TableExtractor()
tables = extractor.detect_tables('document_with_table.jpg')
print(f"检测到 {len(tables)} 个表格")
4.3 img2table实战
# 安装: pip install img2table
from img2table.document import Image as Img2TableImage, PDF
from img2table.ocr import TesseractOCR
def extract_tables_from_image(image_path):
"""从图片中提取表格"""
# 创建文档对象
img = Img2TableImage(src=image_path)
# 使用Tesseract作为OCR引擎
ocr = TesseractOCR(n_threads=4, lang="chi_sim+eng")
# 提取表格
extracted_tables = img.extract_tables(
ocr=ocr,
implicit_rows=True,
borderless_tables=True,
min_confidence=50
)
for idx, table in enumerate(extracted_tables):
df = table.df
print(f"表格 {idx + 1}:")
print(df)
print("-" * 50)
return extracted_tables
def extract_tables_from_pdf(pdf_path):
"""从PDF中提取表格"""
pdf = PDF(src=pdf_path)
# 提取所有页面的表格
extracted_tables = pdf.extract_tables()
all_tables = []
for page_num, page_tables in extracted_tables.items():
for table in page_tables:
df = table.df
all_tables.append({
'page': page_num,
'dataframe': df
})
return all_tables
4.4 表格数据清洗与规范化
import pandas as pd
import re
class TableCleaner:
"""表格数据清洗器"""
@staticmethod
def clean_dataframe(df):
"""清洗DataFrame"""
# 去除前后空格
df = df.applymap(lambda x: x.strip() if isinstance(x, str) else x)
# 替换空字符串为NaN
df = df.replace('', pd.NA)
# 去除全空行和全空列
df = df.dropna(how='all', axis=0)
df = df.dropna(how='all', axis=1)
# 重置索引
df = df.reset_index(drop=True)
return df
@staticmethod
def detect_column_types(df):
"""自动检测列的数据类型"""
type_mapping = {}
for col in df.columns:
sample = df[col].dropna().head(20)
# 尝试解析为数字
numeric_count = 0
for val in sample:
try:
float(str(val).replace(',', '').replace('¥', '').replace('%', ''))
numeric_count += 1
except:
pass
if numeric_count > len(sample) * 0.7:
type_mapping[col] = 'numeric'
elif TableCleaner._is_date_column(sample):
type_mapping[col] = 'date'
else:
type_mapping[col] = 'text'
return type_mapping
@staticmethod
def _is_date_column(series):
"""判断是否为日期列"""
date_patterns = [
r'\d{4}[-/]\d{1,2}[-/]\d{1,2}',
r'\d{1,2}[-/]\d{1,2}[-/]\d{4}',
r'\d{4}年\d{1,2}月\d{1,2}日'
]
for val in series:
val_str = str(val)
if any(re.match(p, val_str) for p in date_patterns):
return True
return False
@staticmethod
def convert_types(df, type_mapping):
"""根据类型映射转换数据类型"""
for col, dtype in type_mapping.items():
if dtype == 'numeric':
df[col] = pd.to_numeric(
df[col].astype(str).str.replace(',', '').str.replace('¥', '').str.replace('%', ''),
errors='coerce'
)
elif dtype == 'date':
df[col] = pd.to_datetime(df[col], errors='coerce', format='mixed')
return df
# 使用示例
cleaner = TableCleaner()
df = cleaner.clean_dataframe(raw_df)
types = cleaner.detect_column_types(df)
df = cleaner.convert_types(df, types)
print(df.dtypes)
5. 多模态文档理解
5.1 VLM（视觉语言模型）在文档理解中的应用
随着GPT-4V、Qwen-VL等多模态大模型的出现，文档理解进入了新阶段。这些模型能够直接理解文档图像，无需传统的OCR+解析流程。
import base64
from openai import OpenAI
class DocumentUnderstandingVLM:
"""基于视觉语言模型的文档理解"""
def __init__(self, api_key, model="gpt-4o"):
self.client = OpenAI(api_key=api_key)
self.model = model
def encode_image(self, image_path):
"""将图片编码为base64"""
with open(image_path, "rb") as f:
return base64.b64encode(f.read()).decode("utf-8")
def understand_document(self, image_path, prompt=None):
"""理解文档图片内容"""
if prompt is None:
prompt = """请仔细分析这份文档，完成以下任务：
1. 识别文档类型（发票、合同、报告、表格等）
2. 提取所有关键信息
3. 以结构化的JSON格式返回结果"""
base64_image = self.encode_image(image_path)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}",
"detail": "high"
}
}
]
}
],
max_tokens=4096
)
return response.choices[0].message.content
def extract_table_from_image(self, image_path):
"""从图片中提取表格数据"""
prompt = """请识别图片中的表格，将表格内容提取为JSON格式。
输出格式示例：
{
"headers": ["列名1", "列名2", ...],
"rows": [
["值1", "值2", ...],
...
]
}"""
return self.understand_document(image_path, prompt)
def multi_page_understanding(self, image_paths):
"""理解多页文档"""
content = [
{"type": "text", "text": "这是一份多页文档，请综合分析所有页面的内容，提取关键信息。"}
]
for path in image_paths:
base64_img = self.encode_image(path)
content.append({
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{base64_img}"}
})
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": content}],
max_tokens=4096
)
return response.choices[0].message.content
5.2 本地多模态模型方案
# 使用Qwen-VL进行本地文档理解
# 安装: pip install transformers torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from PIL import Image
class LocalDocumentVLM:
def __init__(self, model_path="Qwen/Qwen-VL-Chat"):
self.tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
self.model = AutoModelForCausalLM.from_pretrained(
model_path, device_map="auto", trust_remote_code=True
).eval()
def understand(self, image_path, query):
"""理解文档"""
image = Image.open(image_path)
response, history = self.model.chat(
self.tokenizer,
query=f"<img>{image_path}</img>\n{query}",
history=None
)
return response
def extract_info(self, image_path):
"""提取文档关键信息"""
query = """请分析这份文档，提取以下信息：
1. 文档类型
2. 主要内容摘要
3. 关键数据和数字
4. 重要日期和时间
以JSON格式返回结果。"""
return self.understand(image_path, query)
6. LLM结构化数据提取
6.1 使用OpenAI Function Calling进行结构化提取
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Optional, List
import json
class InvoiceInfo(BaseModel):
"""发票信息"""
invoice_number: str = Field(description="发票号码")
date: str = Field(description="开票日期")
seller: str = Field(description="销售方名称")
buyer: str = Field(description="购买方名称")
total_amount: float = Field(description="价税合计金额")
tax_amount: float = Field(description="税额")
items: List[dict] = Field(description="发票明细项目列表")
class ContractInfo(BaseModel):
"""合同信息"""
contract_number: str = Field(description="合同编号")
party_a: str = Field(description="甲方名称")
party_b: str = Field(description="乙方名称")
signing_date: str = Field(description="签订日期")
effective_date: str = Field(description="生效日期")
expiry_date: Optional[str] = Field(description="到期日期")
total_value: float = Field(description="合同总金额")
key_terms: List[str] = Field(description="关键条款摘要")
def extract_structured_info(text, schema_class):
"""使用LLM提取结构化信息"""
client = OpenAI()
# 从Pydantic模型生成JSON Schema
schema = schema_class.model_json_schema()
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": "你是一个专业的文档信息提取助手。请从文档文本中提取结构化信息，严格按照给定的JSON Schema格式输出。"
},
{
"role": "user",
"content": f"请从以下文档中提取信息：\n\n{text}"
}
],
response_format={"type": "json_object"},
temperature=0
)
result = json.loads(response.choices[0].message.content)
return schema_class(**result)
6.2 使用Instructor库简化提取
# 安装: pip install instructor
import instructor
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import List, Optional
# 启用instructor的patch
client = instructor.from_openai(OpenAI())
class Person(BaseModel):
"""人员信息"""
name: str = Field(description="姓名")
role: str = Field(description="职务/角色")
organization: Optional[str] = Field(description="所属组织")
contact: Optional[str] = Field(description="联系方式")
class DocumentSummary(BaseModel):
"""文档摘要"""
title: str = Field(description="文档标题")
document_type: str = Field(description="文档类型")
summary: str = Field(description="核心内容摘要，100字以内")
key_points: List[str] = Field(description="关键要点列表")
people_mentioned: List[Person] = Field(description="文中提到的人物")
dates: List[str] = Field(description="重要日期")
numbers: List[dict] = Field(description="关键数字及其含义")
action_items: Optional[List[str]] = Field(description="待办事项")
def extract_document_summary(text):
"""提取文档摘要和关键信息"""
summary = client.chat.completions.create(
model="gpt-4o",
response_model=DocumentSummary,
max_retries=3,
messages=[
{
"role": "system",
"content": "你是一个专业的文档分析助手，请准确提取文档中的关键信息。"
},
{
"role": "user",
"content": f"请分析以下文档内容：\n\n{text}"
}
]
)
return summary
# 使用示例
text = "会议纪要：2024年3月15日，张三（项目经理）和李四（技术总监）讨论了..."
result = extract_document_summary(text)
print(f"标题: {result.title}")
print(f"要点: {result.key_points}")
6.3 使用LangChain进行复杂文档处理
# 安装: pip install langchain langchain-openai
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List
class FinancialReport(BaseModel):
"""财务报告信息"""
company_name: str = Field(description="公司名称")
report_period: str = Field(description="报告期间")
revenue: float = Field(description="营业收入(万元)")
net_profit: float = Field(description="净利润(万元)")
total_assets: float = Field(description="总资产(万元)")
growth_rate: float = Field(description="营收增长率(%)")
risk_factors: List[str] = Field(description="风险因素")
outlook: str = Field(description="未来展望")
def extract_financial_info(text):
"""提取财务报告关键数据"""
parser = PydanticOutputParser(pydantic_object=FinancialReport)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的财务分析助手。请从财务报告中提取关键数据。\n{format_instructions}"),
("human", "请分析以下财务报告：\n\n{text}")
])
llm = ChatOpenAI(model="gpt-4o", temperature=0)
chain = prompt | llm | parser
result = chain.invoke({
"text": text,
"format_instructions": parser.get_format_instructions()
})
return result
7. 发票/合同/报告场景实战
7.1 发票处理全流程
import re
from datetime import datetime
class InvoiceProcessor:
"""发票处理器"""
def __init__(self, ocr_engine='paddleocr'):
self.ocr_engine = ocr_engine
self._init_ocr()
def _init_ocr(self):
if self.ocr_engine == 'paddleocr':
from paddleocr import PaddleOCR
self.ocr = PaddleOCR(use_angle_cls=True, lang='ch')
def process_invoice(self, image_path):
"""处理单张发票"""
# Step 1: OCR识别
result = self.ocr.ocr(image_path, cls=True)
texts = [line[1][0] for line in result[0]]
full_text = '\n'.join(texts)
# Step 2: 提取关键字段
invoice_data = self._extract_fields(full_text, texts)
# Step 3: 验证和纠错
invoice_data = self._validate(invoice_data)
return invoice_data
def _extract_fields(self, full_text, text_lines):
"""提取发票关键字段"""
data = {}
# 发票号码
for line in text_lines:
match = re.search(r'发票号码[：:]\s*(\d+)', line)
if match:
data['invoice_number'] = match.group(1)
break
# 开票日期
for line in text_lines:
match = re.search(r'(\d{4})\s*年\s*(\d{1,2})\s*月\s*(\d{1,2})\s*日', line)
if match:
data['date'] = f"{match.group(1)}-{match.group(2).zfill(2)}-{match.group(3).zfill(2)}"
break
# 金额提取
for line in text_lines:
match = re.search(r'价税合计.*?[¥￥]?\s*([\d,]+\.?\d*)', line)
if match:
data['total_amount'] = float(match.group(1).replace(',', ''))
match = re.search(r'税\s*额.*?[¥￥]?\s*([\d,]+\.?\d*)', line)
if match:
data['tax_amount'] = float(match.group(1).replace(',', ''))
# 购买方和销售方
for i, line in enumerate(text_lines):
if '购买方' in line or '购方' in line:
# 下一行通常是名称
if i + 1 < len(text_lines):
data['buyer'] = text_lines[i + 1]
if '销售方' in line or '销方' in line:
if i + 1 < len(text_lines):
data['seller'] = text_lines[i + 1]
return data
def _validate(self, data):
"""验证发票数据"""
errors = []
if 'total_amount' in data and 'tax_amount' in data:
# 简单的金额逻辑校验
if data['tax_amount'] > data['total_amount']:
errors.append("税额大于总金额")
if 'date' in data:
try:
date = datetime.strptime(data['date'], '%Y-%m-%d')
if date > datetime.now():
errors.append("开票日期在未来")
except ValueError:
errors.append("日期格式错误")
data['validation_errors'] = errors
data['is_valid'] = len(errors) == 0
return data
# 批量处理发票
def batch_process_invoices(image_paths):
processor = InvoiceProcessor()
results = []
for path in image_paths:
try:
data = processor.process_invoice(path)
data['source_file'] = path
results.append(data)
except Exception as e:
results.append({
'source_file': path,
'error': str(e)
})
return results
7.2 合同信息提取
class ContractProcessor:
"""合同处理器"""
def extract_contract_info(self, text):
"""提取合同关键信息"""
info = {}
# 合同编号
match = re.search(r'合同编号[：:]\s*(\S+)', text)
if match:
info['contract_number'] = match.group(1)
# 甲方乙方
match = re.search(r'甲\s*方[：:]\s*(.+?)[\n\r]', text)
if match:
info['party_a'] = match.group(1).strip()
match = re.search(r'乙\s*方[：:]\s*(.+?)[\n\r]', text)
if match:
info['party_b'] = match.group(1).strip()
# 金额
match = re.search(r'[合同总金总额价款]+.*?[¥￥]?\s*([\d,]+\.?\d*)\s*元', text)
if match:
info['amount'] = float(match.group(1).replace(',', ''))
# 日期
date_patterns = [
r'签订日期[：:]\s*(\d{4}[-/年]\d{1,2}[-/月]\d{1,2}日?)',
r'(\d{4}年\d{1,2}月\d{1,2}日)\s*签订',
]
for pattern in date_patterns:
match = re.search(pattern, text)
if match:
info['signing_date'] = match.group(1)
break
# 合同期限
match = re.search(r'期限.*?(\d+)\s*[年月日]', text)
if match:
info['duration'] = match.group(0)
# 关键条款
clauses = []
clause_patterns = [
r'第[一二三四五六七八九十\d]+条\s*(.+)',
r'[（(]\s*[一二三四五六七八九十\d]+\s*[）)]\s*(.+)'
]
for pattern in clause_patterns:
matches = re.findall(pattern, text)
clauses.extend(matches[:10]) # 最多取10条
info['clauses'] = clauses
return info
def identify_risk_clauses(self, text):
"""识别合同风险条款"""
risk_keywords = [
'违约金', '赔偿', '免责', '不可抗力', '解除合同',
'终止', '保密', '竞业限制', '知识产权归属',
'争议解决', '仲裁', '诉讼'
]
risks = []
for keyword in risk_keywords:
if keyword in text:
# 找到包含关键词的句子
sentences = re.split(r'[。；;]', text)
for sent in sentences:
if keyword in sent:
risks.append({
'keyword': keyword,
'clause': sent.strip()
})
break
return risks
7.3 财务报告分析
class ReportAnalyzer:
"""财务报告分析器"""
def analyze_annual_report(self, text):
"""分析年度报告"""
from collections import defaultdict
analysis = {
'financials': self._extract_financials(text),
'growth_indicators': self._analyze_growth(text),
'risk_factors': self._extract_risks(text),
'outlook': self._extract_outlook(text)
}
return analysis
def _extract_financials(self, text):
"""提取财务数据"""
financials = {}
patterns = {
'revenue': r'营[业收]*收入.*?([\d,]+\.?\d*)\s*[万亿]',
'net_profit': r'净利润.*?([\d,]+\.?\d*)\s*[万亿]',
'total_assets': r'总资产.*?([\d,]+\.?\d*)\s*[万亿]',
'eps': r'每股收益.*?([\d.]+)\s*元',
}
for key, pattern in patterns.items():
match = re.search(pattern, text)
if match:
financials[key] = float(match.group(1).replace(',', ''))
return financials
def _analyze_growth(self, text):
"""分析增长指标"""
growth = {}
patterns = {
'revenue_growth': r'收入.*?增长.*?([\d.]+)%',
'profit_growth': r'利润.*?增长.*?([\d.]+)%',
}
for key, pattern in patterns.items():
match = re.search(pattern, text)
if match:
growth[key] = float(match.group(1))
return growth
def _extract_risks(self, text):
"""提取风险因素"""
risk_section = re.search(r'风险因素[：:]?\s*(.*?)(?=##|$)', text, re.DOTALL)
if risk_section:
risks = re.split(r'[。；\n]', risk_section.group(1))
return [r.strip() for r in risks if r.strip()]
return []
def _extract_outlook(self, text):
"""提取未来展望"""
outlook_patterns = [
r'未来展望[：:]?\s*(.*?)(?=##|$)',
r'发展前景[：:]?\s*(.*?)(?=##|$)',
r'经营计划[：:]?\s*(.*?)(?=##|$)'
]
for pattern in outlook_patterns:
match = re.search(pattern, text, re.DOTALL)
if match:
return match.group(1).strip()[:500]
return None
8. 文档分类与信息抽取Pipeline
8.1 文档分类器
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
import pickle
class DocumentClassifier:
"""文档分类器"""
def __init__(self):
self.pipeline = Pipeline([
('tfidf', TfidfVectorizer(
max_features=10000,
ngram_range=(1, 2),
sublinear_tf=True
)),
('classifier', MultinomialNB(alpha=0.1))
])
self.label_map = {
0: '发票',
1: '合同',
2: '报告',
3: '简历',
4: '通知',
5: '其他'
}
def train(self, texts, labels):
"""训练分类器"""
self.pipeline.fit(texts, labels)
def predict(self, text):
"""预测文档类别"""
label_idx = self.pipeline.predict([text])[0]
probabilities = self.pipeline.predict_proba([text])[0]
return {
'category': self.label_map[label_idx],
'confidence': float(probabilities[label_idx]),
'all_probabilities': {
self.label_map[i]: float(p)
for i, p in enumerate(probabilities)
}
}
def save(self, path):
with open(path, 'wb') as f:
pickle.dump(self.pipeline, f)
def load(self, path):
with open(path, 'rb') as f:
self.pipeline = pickle.load(f)
# 使用LLM进行零样本分类
def classify_with_llm(text, categories):
"""使用LLM进行零样本文档分类"""
client = OpenAI()
categories_str = '、'.join(categories)
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": f"你是一个文档分类助手。请将文档分类到以下类别之一：{categories_str}。返回JSON格式：{{\"category\": \"类别\", \"confidence\": 0.95, \"reason\": \"原因\"}}"
},
{
"role": "user",
"content": f"请分类以下文档：\n\n{text[:2000]}"
}
],
response_format={"type": "json_object"},
temperature=0
)
import json
return json.loads(response.choices[0].message.content)
8.2 命名实体识别Pipeline
import spacy
from transformers import pipeline as hf_pipeline
class NERPipeline:
"""命名实体识别Pipeline"""
def __init__(self, method='spacy'):
self.method = method
if method == 'spacy':
# 加载中文模型
self.nlp = spacy.load("zh_core_web_trf")
elif method == 'transformers':
# 使用Hugging Face NER模型
self.ner = hf_pipeline(
"ner",
model="bert-base-chinese",
aggregation_strategy="simple"
)
def extract_entities(self, text):
"""提取命名实体"""
if self.method == 'spacy':
return self._extract_with_spacy(text)
elif self.method == 'transformers':
return self._extract_with_transformers(text)
def _extract_with_spacy(self, text):
doc = self.nlp(text)
entities = []
for ent in doc.ents:
entities.append({
'text': ent.text,
'label': ent.label_,
'start': ent.start_char,
'end': ent.end_char,
'description': spacy.explain(ent.label_)
})
return entities
def _extract_with_transformers(self, text):
results = self.ner(text)
entities = []
for entity in results:
entities.append({
'text': entity['word'],
'label': entity['entity_group'],
'score': entity['score'],
'start': entity['start'],
'end': entity['end']
})
return entities
def extract_custom_entities(self, text, entity_types):
"""使用LLM提取自定义实体类型"""
client = OpenAI()
types_str = '、'.join(entity_types)
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": f"请从文本中提取以下类型的实体：{types_str}。返回JSON数组格式。"
},
{"role": "user", "content": text}
],
response_format={"type": "json_object"},
temperature=0
)
return json.loads(response.choices[0].message.content)
8.3 完整文档处理Pipeline
class DocumentPipeline:
"""完整文档处理Pipeline"""
def __init__(self):
self.classifier = DocumentClassifier()
self.ner = NERPipeline(method='spacy')
self.invoice_processor = InvoiceProcessor()
self.contract_processor = ContractProcessor()
self.report_analyzer = ReportAnalyzer()
def process(self, document_input, input_type='text'):
"""
处理文档的完整Pipeline
Args:
document_input: 文档内容（文本或图片路径）
input_type: 'text' 或 'image'
"""
result = {
'input_type': input_type,
'processing_steps': []
}
# Step 1: 如果是图片，先进行OCR
if input_type == 'image':
ocr_result = self._ocr(document_input)
text = ocr_result['text']
result['ocr'] = ocr_result
result['processing_steps'].append('OCR')
else:
text = document_input
# Step 2: 文档分类
classification = self.classifier.predict(text)
result['classification'] = classification
result['processing_steps'].append('Classification')
# Step 3: 根据分类结果选择处理策略
category = classification['category']
if category == '发票':
result['extracted_info'] = self.invoice_processor.process_invoice(document_input)
elif category == '合同':
result['extracted_info'] = self.contract_processor.extract_contract_info(text)
elif category == '报告':
result['extracted_info'] = self.report_analyzer.analyze_annual_report(text)
else:
result['extracted_info'] = self._general_extraction(text)
result['processing_steps'].append('Extraction')
# Step 4: 实体识别
entities = self.ner.extract_entities(text)
result['entities'] = entities
result['processing_steps'].append('NER')
# Step 5: 生成摘要
result['summary'] = self._generate_summary(text)
result['processing_steps'].append('Summary')
return result
def _ocr(self, image_path):
"""OCR识别"""
from paddleocr import PaddleOCR
ocr = PaddleOCR(use_angle_cls=True, lang='ch')
result = ocr.ocr(image_path, cls=True)
texts = []
for line in result[0]:
texts.append({
'text': line[1][0],
'confidence': line[1][1],
'bbox': line[0]
})
return {
'text': '\n'.join([t['text'] for t in texts]),
'details': texts
}
def _general_extraction(self, text):
"""通用信息提取"""
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": "请从文档中提取关键信息，包括：标题、日期、关键人物、关键数字、主要内容。返回JSON格式。"
},
{"role": "user", "content": text[:3000]}
],
response_format={"type": "json_object"},
temperature=0
)
return json.loads(response.choices[0].message.content)
def _generate_summary(self, text):
"""生成文档摘要"""
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": "请用不超过100字概括文档的核心内容。"
},
{"role": "user", "content": text[:3000]}
],
max_tokens=200,
temperature=0
)
return response.choices[0].message.content
9. 文档Chunking策略
9.1 为什么需要文档Chunking
在构建文档检索和问答系统时，由于LLM的上下文窗口限制，需要将长文档切分为较小的块（chunks）。好的Chunking策略能够：
- 保持语义完整性
- 维护上下文关联
- 提高检索精度
- 优化生成质量
9.2 按段落Chunking
class ParagraphChunker:
"""按段落进行文档切分"""
def __init__(self, max_chunk_size=1000, overlap=100):
self.max_chunk_size = max_chunk_size
self.overlap = overlap
def chunk(self, text):
"""按段落切分文档"""
# 按双换行符分割段落
paragraphs = re.split(r'\n\s*\n', text)
chunks = []
current_chunk = []
current_size = 0
for para in paragraphs:
para = para.strip()
if not para:
continue
para_size = len(para)
if current_size + para_size > self.max_chunk_size and current_chunk:
# 当前块已满，保存并开始新块
chunk_text = '\n\n'.join(current_chunk)
chunks.append(chunk_text)
# 保留overlap
if self.overlap > 0:
overlap_text = chunk_text[-self.overlap:]
current_chunk = [overlap_text]
current_size = len(overlap_text)
else:
current_chunk = []
current_size = 0
current_chunk.append(para)
current_size += para_size
# 处理最后一个块
if current_chunk:
chunks.append('\n\n'.join(current_chunk))
return chunks
# 使用示例
chunker = ParagraphChunker(max_chunk_size=1000, overlap=100)
chunks = chunker.chunk(document_text)
for i, chunk in enumerate(chunks):
print(f"Chunk {i+1} ({len(chunk)} chars): {chunk[:100]}...")
9.3 按语义Chunking
from sentence_transformers import SentenceTransformer
import numpy as np
class SemanticChunker:
"""基于语义相似度的文档切分"""
def __init__(self, model_name='paraphrase-multilingual-MiniLM-L12-v2',
max_chunk_size=1000, similarity_threshold=0.5):
self.model = SentenceTransformer(model_name)
self.max_chunk_size = max_chunk_size
self.similarity_threshold = similarity_threshold
def chunk(self, text):
"""基于语义相似度切分"""
# 分割句子
sentences = self._split_sentences(text)
if len(sentences) <= 1:
return [text]
# 计算句子嵌入
embeddings = self.model.encode(sentences)
# 计算相邻句子的相似度
similarities = []
for i in range(len(embeddings) - 1):
sim = np.dot(embeddings[i], embeddings[i + 1]) / (
np.linalg.norm(embeddings[i]) * np.linalg.norm(embeddings[i + 1])
)
similarities.append(sim)
# 在相似度低于阈值的位置切分
chunks = []
current_chunk = [sentences[0]]
current_size = len(sentences[0])
for i, sim in enumerate(similarities):
next_sent = sentences[i + 1]
should_split = (
sim < self.similarity_threshold or
current_size + len(next_sent) > self.max_chunk_size
)
if should_split and current_chunk:
chunks.append(' '.join(current_chunk))
current_chunk = [next_sent]
current_size = len(next_sent)
else:
current_chunk.append(next_sent)
current_size += len(next_sent)
if current_chunk:
chunks.append(' '.join(current_chunk))
return chunks
def _split_sentences(self, text):
"""分割句子"""
# 中英文混合句子分割
sentences = re.split(r'(?<=[。！？.!?])\s*', text)
return [s.strip() for s in sentences if s.strip()]
9.4 按版面Chunking
class LayoutChunker:
"""基于文档版面结构的切分"""
def __init__(self, max_chunk_size=1500):
self.max_chunk_size = max_chunk_size
def chunk_by_headings(self, text):
"""按标题层级切分"""
# 匹配Markdown风格标题
heading_pattern = r'^(#{1,6})\s+(.+)$'
sections = []
current_section = {'level': 0, 'title': '', 'content': ''}
for line in text.split('\n'):
match = re.match(heading_pattern, line)
if match:
# 保存当前section
if current_section['content'].strip():
sections.append(current_section)
# 开始新section
level = len(match.group(1))
current_section = {
'level': level,
'title': match.group(2),
'content': ''
}
else:
current_section['content'] += line + '\n'
# 保存最后一个section
if current_section['content'].strip():
sections.append(current_section)
# 合并小section，拆分大section
chunks = self._merge_and_split(sections)
return chunks
def _merge_and_split(self, sections):
"""合并小段落，拆分大段落"""
chunks = []
current_chunk = ''
for section in sections:
header = '#' * section['level'] + ' ' + section['title']
content = header + '\n' + section['content']
if len(current_chunk) + len(content) > self.max_chunk_size:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = content
else:
current_chunk += '\n' + content
if current_chunk.strip():
chunks.append(current_chunk.strip())
return chunks
9.5 递归字符切分（LangChain风格）
class RecursiveChunker:
"""递归字符切分器"""
def __init__(self, chunk_size=1000, chunk_overlap=200):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
# 按优先级排序的分隔符
self.separators = ['\n\n', '\n', '。', '！', '？', '；', '.', '!', '?', ';', ' ', '']
def chunk(self, text):
"""递归切分"""
return self._recursive_split(text, self.separators)
def _recursive_split(self, text, separators):
"""递归分割"""
final_chunks = []
# 找到合适的分隔符
separator = separators[-1]
new_separators = []
for i, _sep in enumerate(separators):
if _sep == '':
separator = _sep
break
if _sep in text:
separator = _sep
new_separators = separators[i + 1:]
break
# 使用分隔符分割
splits = text.split(separator) if separator else list(text)
# 合并小块
good_splits = []
current = ''
for s in splits:
if len(current) + len(s) + len(separator) <= self.chunk_size:
current += s + separator
else:
if current:
good_splits.append(current)
current = s + separator
if current:
good_splits.append(current)
# 对仍然过大的块递归分割
for split in good_splits:
if len(split) > self.chunk_size and new_separators:
sub_chunks = self._recursive_split(split, new_separators)
final_chunks.extend(sub_chunks)
else:
final_chunks.append(split)
# 添加重叠
if self.chunk_overlap > 0 and len(final_chunks) > 1:
final_chunks = self._add_overlap(final_chunks)
return final_chunks
def _add_overlap(self, chunks):
"""添加重叠部分"""
overlapped = [chunks[0]]
for i in range(1, len(chunks)):
prev = chunks[i - 1]
overlap = prev[-self.chunk_overlap:]
overlapped.append(overlap + chunks[i])
return overlapped
10. 多语言文档处理
10.1 多语言OCR支持
class MultilingualOCR:
"""多语言OCR处理器"""
# 语言配置映射
LANGUAGE_CONFIGS = {
'chinese': {'paddle': 'ch', 'easyocr': ['ch_sim', 'en'], 'tesseract': 'chi_sim'},
'english': {'paddle': 'en', 'easyocr': ['en'], 'tesseract': 'eng'},
'japanese': {'paddle': 'japan', 'easyocr': ['ja', 'en'], 'tesseract': 'jpn'},
'korean': {'paddle': 'korean', 'easyocr': ['ko', 'en'], 'tesseract': 'kor'},
'french': {'paddle': 'fr', 'easyocr': ['fr', 'en'], 'tesseract': 'fra'},
'german': {'paddle': 'german', 'easyocr': ['de', 'en'], 'tesseract': 'deu'},
'arabic': {'paddle': 'arabic', 'easyocr': ['ar', 'en'], 'tesseract': 'ara'},
'thai': {'paddle': 'thai', 'easyocr': ['th', 'en'], 'tesseract': 'tha'},
}
def __init__(self, engine='paddleocr'):
self.engine = engine
def detect_language(self, text):
"""检测文本语言"""
from langdetect import detect, detect_langs
try:
lang = detect(text)
probs = detect_langs(text)
return {
'primary': lang,
'probabilities': {str(p.lang): p.prob for p in probs}
}
except:
return {'primary': 'unknown', 'probabilities': {}}
def process_document(self, image_path, languages=None):
"""处理多语言文档"""
if languages is None:
# 自动检测语言
languages = ['ch_sim', 'en'] # 默认中英文
if self.engine == 'easyocr':
import easyocr
reader = easyocr.Reader(languages, gpu=True)
results = reader.readtext(image_path)
return [{'text': t, 'confidence': c, 'bbox': b} for b, t, c in results]
elif self.engine == 'paddleocr':
from paddleocr import PaddleOCR
lang = self._get_paddle_lang(languages)
ocr = PaddleOCR(use_angle_cls=True, lang=lang)
result = ocr.ocr(image_path, cls=True)
return [{'text': line[1][0], 'confidence': line[1][1]} for line in result[0]]
def _get_paddle_lang(self, languages):
"""将语言列表转换为PaddleOCR语言代码"""
lang_map = {
'ch_sim': 'ch', 'ch_tra': 'chinese_cht',
'en': 'en', 'ja': 'japan', 'ko': 'korean'
}
for lang in languages:
if lang in lang_map:
return lang_map[lang]
return 'en'
10.2 翻译与跨语言处理
class CrossLingualProcessor:
"""跨语言文档处理器"""
def __init__(self, translator='google'):
self.translator = translator
def translate_document(self, text, source_lang, target_lang):
"""翻译文档"""
if self.translator == 'google':
from googletrans import Translator
translator = Translator()
result = translator.translate(text, src=source_lang, dest=target_lang)
return result.text
elif self.translator == 'deepl':
import deepl
translator = deepl.Translator("YOUR_API_KEY")
result = translator.translate_text(text, target_lang=target_lang.upper())
return str(result)
def bilingual_extraction(self, text, source_lang, target_lang):
"""双语信息提取"""
# 原文提取
original_info = self._extract_info(text)
# 翻译后提取
translated = self.translate_document(text, source_lang, target_lang)
translated_info = self._extract_info(translated)
return {
'original': original_info,
'translated': translated_info,
'source_lang': source_lang,
'target_lang': target_lang
}
def _extract_info(self, text):
"""通用信息提取"""
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "提取文档关键信息，返回JSON格式"},
{"role": "user", "content": text[:2000]}
],
response_format={"type": "json_object"},
temperature=0
)
return json.loads(response.choices[0].message.content)
11. 文档质量评估与纠错
11.1 OCR质量评估
class OCRQualityAssessor:
"""OCR质量评估器"""
def assess(self, ocr_result, reference=None):
"""评估OCR结果质量"""
metrics = {
'confidence_score': self._avg_confidence(ocr_result),
'text_density': self._text_density(ocr_result),
'consistency_score': self._consistency_check(ocr_result),
}
if reference:
metrics['accuracy'] = self._calculate_accuracy(ocr_result, reference)
metrics['overall_quality'] = self._overall_score(metrics)
return metrics
def _avg_confidence(self, ocr_result):
"""平均置信度"""
confidences = [item['confidence'] for item in ocr_result]
return sum(confidences) / len(confidences) if confidences else 0
def _text_density(self, ocr_result):
"""文本密度（有效文字比例）"""
total_chars = sum(len(item['text']) for item in ocr_result)
valid_chars = sum(
len(re.sub(r'[^\w\s]', '', item['text']))
for item in ocr_result
)
return valid_chars / total_chars if total_chars > 0 else 0
def _consistency_check(self, ocr_result):
"""一致性检查（检测重复、乱码等）"""
texts = [item['text'] for item in ocr_result]
# 检查重复
unique_ratio = len(set(texts)) / len(texts) if texts else 1
# 检查乱码（非文字字符比例）
garble_count = 0
for text in texts:
non_text = re.sub(r'[\w\s\u4e00-\u9fff]', '', text)
garble_count += len(non_text)
total_len = sum(len(t) for t in texts)
garble_ratio = 1 - (garble_count / total_len) if total_len > 0 else 1
return (unique_ratio + garble_ratio) / 2
def _overall_score(self, metrics):
"""综合质量分数"""
weights = {
'confidence_score': 0.4,
'text_density': 0.3,
'consistency_score': 0.3
}
score = sum(
metrics.get(key, 0) * weight
for key, weight in weights.items()
)
return round(score, 4)
11.2 文本纠错
class TextCorrector:
"""文本纠错器"""
def __init__(self):
self.common_ocr_errors = {
'0': 'O', 'O': '0', # 数字0和字母O
'1': 'l', 'l': '1', # 数字1和字母l
'5': 'S', 'S': '5', # 数字5和字母S
'8': 'B', 'B': '8', # 数字8和字母B
'—': '-', '–': '-',
'"': '"', '"': '"',
''': "'", ''': "'",
'。': '.',
}
def correct_ocr_text(self, text, context='general'):
"""纠正OCR文本错误"""
# 1. 基础清理
text = self._basic_cleanup(text)
# 2. 常见OCR错误纠正
text = self._fix_common_errors(text, context)
# 3. 数字/金额纠正
if context in ['invoice', 'financial']:
text = self._fix_numbers(text)
# 4. 日期格式纠正
text = self._fix_dates(text)
return text
def _basic_cleanup(self, text):
"""基础清理"""
# 去除多余空格
text = re.sub(r'\s+', ' ', text)
# 去除特殊字符
text = re.sub(r'[^\w\s\u4e00-\u9fff，。！？、；：""''（）\-\+\*\/\.\,\!\?\;\:\(\)]', '', text)
return text.strip()
def _fix_common_errors(self, text, context):
"""纠正常见OCR错误"""
# 中文标点纠正
replacements = {
',': '，',
'.': '。',
':': '：',
';': '；',
'!': '！',
'?': '？',
}
# 只在中文上下文中替换
if context in ['chinese', 'general']:
for eng, chn in replacements.items():
# 避免替换数字中的点
text = re.sub(r'(?<=[\u4e00-\u9fff])' + re.escape(eng) + r'(?=[\u4e00-\u9fff\s])', chn, text)
return text
def _fix_numbers(self, text):
"""纠正数字错误"""
# 修复常见数字OCR错误
text = re.sub(r'(?<=\d)[Oo](?=\d)', '0', text)
text = re.sub(r'(?<=\d)[lI](?=\d)', '1', text)
text = re.sub(r'(?<=\d)[S](?=\d)', '5', text)
# 修复金额格式
text = re.sub(r'(\d+)[,.](\d{3})', r'\1,\2', text)
return text
def _fix_dates(self, text):
"""纠正日期格式"""
# 统一日期格式
text = re.sub(r'(\d{4})\s*年\s*(\d{1,2})\s*月\s*(\d{1,2})\s*日', r'\1-\2-\3', text)
text = re.sub(r'(\d{4})[/\.](\d{1,2})[/\.](\d{1,2})', r'\1-\2-\3', text)
return text
def llm_correct(self, text):
"""使用LLM进行高级纠错"""
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": "你是一个文本纠错专家。请纠正以下OCR识别文本中的错误，保持原文格式，只修正明显的识别错误。"
},
{"role": "user", "content": text}
],
temperature=0
)
return response.choices[0].message.content
12. 批量文档处理工作流
12.1 文件监控与自动处理
import os
import time
import hashlib
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
class DocumentWorkflow:
"""批量文档处理工作流"""
def __init__(self, input_dir, output_dir, max_workers=4):
self.input_dir = Path(input_dir)
self.output_dir = Path(output_dir)
self.max_workers = max_workers
self.pipeline = DocumentPipeline()
# 支持的文件格式
self.supported_formats = {'.pdf', '.jpg', '.jpeg', '.png', '.tiff', '.bmp'}
# 创建输出目录
self.output_dir.mkdir(parents=True, exist_ok=True)
# 状态文件
self.state_file = self.output_dir / '.processing_state.json'
self.state = self._load_state()
def _load_state(self):
"""加载处理状态"""
if self.state_file.exists():
with open(self.state_file) as f:
return json.load(f)
return {'processed': {}, 'failed': {}}
def _save_state(self):
"""保存处理状态"""
with open(self.state_file, 'w') as f:
json.dump(self.state, f, ensure_ascii=False, indent=2)
def _get_file_hash(self, file_path):
"""计算文件哈希"""
hasher = hashlib.md5()
with open(file_path, 'rb') as f:
for chunk in iter(lambda: f.read(8192), b''):
hasher.update(chunk)
return hasher.hexdigest()
def scan_documents(self):
"""扫描待处理文档"""
documents = []
for file_path in self.input_dir.rglob('*'):
if file_path.suffix.lower() in self.supported_formats:
file_hash = self._get_file_hash(file_path)
# 跳过已处理的文件
if file_hash in self.state['processed']:
continue
documents.append({
'path': str(file_path),
'hash': file_hash,
'size': file_path.stat().st_size,
'format': file_path.suffix.lower()
})
return documents
def process_single(self, doc_info):
"""处理单个文档"""
file_path = doc_info['path']
try:
# 判断输入类型
if doc_info['format'] == '.pdf':
input_type = 'pdf'
else:
input_type = 'image'
# 执行处理Pipeline
result = self.pipeline.process(file_path, input_type=input_type)
# 保存结果
output_path = self.output_dir / f"{Path(file_path).stem}.json"
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(result, f, ensure_ascii=False, indent=2, default=str)
# 更新状态
self.state['processed'][doc_info['hash']] = {
'path': file_path,
'output': str(output_path),
'timestamp': time.time()
}
return {'status': 'success', 'file': file_path, 'output': str(output_path)}
except Exception as e:
self.state['failed'][doc_info['hash']] = {
'path': file_path,
'error': str(e),
'timestamp': time.time()
}
return {'status': 'failed', 'file': file_path, 'error': str(e)}
def run(self):
"""运行批量处理"""
documents = self.scan_documents()
print(f"发现 {len(documents)} 个待处理文档")
results = []
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = {
executor.submit(self.process_single, doc): doc
for doc in documents
}
for future in as_completed(futures):
result = future.result()
results.append(result)
status = '✓' if result['status'] == 'success' else '✗'
print(f"{status} {result['file']}")
# 定期保存状态
self._save_state()
self._save_state()
# 生成报告
success = sum(1 for r in results if r['status'] == 'success')
failed = sum(1 for r in results if r['status'] == 'failed')
print(f"\n处理完成: 成功 {success}, 失败 {failed}")
return results
def watch(self, interval=10):
"""监控文件夹，自动处理新文件"""
print(f"开始监控 {self.input_dir}，每 {interval} 秒检查一次...")
try:
while True:
documents = self.scan_documents()
if documents:
print(f"发现 {len(documents)} 个新文档")
self.run()
time.sleep(interval)
except KeyboardInterrupt:
print("\n停止监控")
# 使用示例
workflow = DocumentWorkflow(
input_dir='./input_documents',
output_dir='./output_results',
max_workers=4
)
# 一次性处理
workflow.run()
# 或持续监控
# workflow.watch(interval=30)
12.2 进度跟踪与错误恢复
import logging
from datetime import datetime
class ProcessingTracker:
"""处理进度跟踪器"""
def __init__(self, log_file='processing.log'):
self.logger = logging.getLogger('DocumentProcessing')
self.logger.setLevel(logging.INFO)
handler = logging.FileHandler(log_file, encoding='utf-8')
handler.setFormatter(logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s'
))
self.logger.addHandler(handler)
self.stats = {
'total': 0,
'processed': 0,
'failed': 0,
'skipped': 0,
'start_time': None,
'errors': []
}
def start(self, total):
"""开始处理"""
self.stats['total'] = total
self.stats['start_time'] = datetime.now()
self.logger.info(f"开始处理 {total} 个文档")
def record_success(self, file_path, duration):
"""记录成功"""
self.stats['processed'] += 1
self.logger.info(f"成功: {file_path} ({duration:.2f}s)")
def record_failure(self, file_path, error):
"""记录失败"""
self.stats['failed'] += 1
self.stats['errors'].append({
'file': file_path,
'error': str(error),
'time': datetime.now().isoformat()
})
self.logger.error(f"失败: {file_path} - {error}")
def get_progress(self):
"""获取进度"""
processed = self.stats['processed'] + self.stats['failed']
elapsed = (datetime.now() - self.stats['start_time']).total_seconds()
return {
'progress': f"{processed}/{self.stats['total']}",
'percentage': f"{processed/self.stats['total']*100:.1f}%",
'success_rate': f"{self.stats['processed']/max(processed,1)*100:.1f}%",
'elapsed': f"{elapsed:.0f}s",
'eta': f"{elapsed/max(processed,1)*(self.stats['total']-processed):.0f}s"
}
def generate_report(self):
"""生成处理报告"""
return {
'summary': self.stats,
'progress': self.get_progress(),
'error_details': self.stats['errors']
}
13. 文档智能API服务搭建
13.1 使用FastAPI构建文档处理服务
# 安装: pip install fastapi uvicorn python-multipart
from fastapi import FastAPI, UploadFile, File, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import List, Optional
import uuid
import asyncio
app = FastAPI(title="文档智能API", version="1.0.0")
# 全局Pipeline实例
pipeline = DocumentPipeline()
# 任务存储
tasks = {}
class TaskStatus(BaseModel):
task_id: str
status: str # pending, processing, completed, failed
progress: float = 0.0
result: Optional[dict] = None
error: Optional[str] = None
class ExtractRequest(BaseModel):
text: str
schema: Optional[dict] = None
document_type: Optional[str] = None
@app.post("/api/v1/ocr", summary="OCR识别")
async def ocr_recognize(file: UploadFile = File(...)):
"""上传图片进行OCR识别"""
# 保存临时文件
temp_path = f"/tmp/{uuid.uuid4()}.{file.filename.split('.')[-1]}"
with open(temp_path, "wb") as f:
content = await file.read()
f.write(content)
try:
result = pipeline._ocr(temp_path)
return {"status": "success", "data": result}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
finally:
os.remove(temp_path)
@app.post("/api/v1/extract", summary="结构化提取")
async def extract_info(request: ExtractRequest):
"""从文本中提取结构化信息"""
try:
result = extract_structured_info(request.text, DocumentSummary)
return {"status": "success", "data": result.model_dump()}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/v1/analyze", summary="文档分析")
async def analyze_document(
file: UploadFile = File(...),
background_tasks: BackgroundTasks = None
):
"""完整文档分析（异步）"""
task_id = str(uuid.uuid4())
tasks[task_id] = TaskStatus(task_id=task_id, status="pending")
# 保存临时文件
temp_path = f"/tmp/{task_id}.{file.filename.split('.')[-1]}"
with open(temp_path, "wb") as f:
content = await file.read()
f.write(content)
# 异步处理
background_tasks.add_task(process_document_async, task_id, temp_path)
return {"task_id": task_id, "status": "processing"}
async def process_document_async(task_id, file_path):
"""异步处理文档"""
tasks[task_id].status = "processing"
try:
result = pipeline.process(file_path, input_type='image')
tasks[task_id].status = "completed"
tasks[task_id].result = result
except Exception as e:
tasks[task_id].status = "failed"
tasks[task_id].error = str(e)
finally:
os.remove(file_path)
@app.get("/api/v1/task/{task_id}", summary="查询任务状态")
async def get_task_status(task_id: str):
"""查询异步任务状态"""
if task_id not in tasks:
raise HTTPException(status_code=404, detail="Task not found")
return tasks[task_id]
@app.post("/api/v1/batch", summary="批量处理")
async def batch_process(files: List[UploadFile] = File(...)):
"""批量处理多个文档"""
results = []
for file in files:
temp_path = f"/tmp/{uuid.uuid4()}.{file.filename.split('.')[-1]}"
with open(temp_path, "wb") as f:
content = await file.read()
f.write(content)
try:
result = pipeline.process(temp_path, input_type='image')
results.append({
"filename": file.filename,
"status": "success",
"data": result
})
except Exception as e:
results.append({
"filename": file.filename,
"status": "failed",
"error": str(e)
})
finally:
os.remove(temp_path)
return {"results": results}
@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)
13.2 Docker部署配置
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
libgl1-mesa-glx \
libglib2.0-0 \
tesseract-ocr \
tesseract-ocr-chi-sim \
&& rm -rf /var/lib/apt/lists/*
# 安装Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制代码
COPY . .
# 暴露端口
EXPOSE 8000
# 启动服务
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
# docker-compose.yml
version: '3.8'
services:
document-api:
build: .
ports:
- "8000:8000"
volumes:
- ./data:/app/data
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- MAX_WORKERS=4
deploy:
resources:
limits:
memory: 4G
reservations:
devices:
- driver: nvidia
count: 1
capabilities: [gpu]
restart: unless-stopped
13.3 客户端SDK
import requests
from pathlib import Path
class DocumentAIClient:
"""文档智能API客户端"""
def __init__(self, base_url="http://localhost:8000"):
self.base_url = base_url.rstrip('/')
def ocr(self, file_path):
"""OCR识别"""
with open(file_path, 'rb') as f:
response = requests.post(
f"{self.base_url}/api/v1/ocr",
files={"file": f}
)
response.raise_for_status()
return response.json()['data']
def extract(self, text, schema=None):
"""结构化提取"""
response = requests.post(
f"{self.base_url}/api/v1/extract",
json={"text": text, "schema": schema}
)
response.raise_for_status()
return response.json()['data']
def analyze(self, file_path):
"""文档分析（异步）"""
with open(file_path, 'rb') as f:
response = requests.post(
f"{self.base_url}/api/v1/analyze",
files={"file": f}
)
response.raise_for_status()
task_id = response.json()['task_id']
# 轮询等待结果
import time
while True:
status = self._get_task_status(task_id)
if status['status'] == 'completed':
return status['result']
elif status['status'] == 'failed':
raise Exception(status['error'])
time.sleep(1)
def batch(self, file_paths):
"""批量处理"""
files = [("files", open(f, 'rb')) for f in file_paths]
response = requests.post(
f"{self.base_url}/api/v1/batch",
files=files
)
response.raise_for_status()
return response.json()['results']
def _get_task_status(self, task_id):
response = requests.get(f"{self.base_url}/api/v1/task/{task_id}")
response.raise_for_status()
return response.json()
# 使用示例
client = DocumentAIClient("http://localhost:8000")
# 单张OCR
result = client.ocr("invoice.jpg")
print(result)
# 批量处理
results = client.batch(["doc1.pdf", "doc2.jpg", "doc3.png"])
for r in results:
print(f"{r['filename']}: {r['status']}")
14. 最佳实践与总结
14.1 技术选型建议
def recommend_tech_stack(requirements):
"""
根据需求推荐技术栈
Args:
requirements: dict，包含以下字段：
- document_types: list, 文档类型（PDF/图片/扫描件）
- languages: list, 需要支持的语言
- volume: str, 处理量级（小/中/大）
- accuracy_priority: str, 精度要求（一般/高/极高）
- budget: str, 预算（低/中/高）
"""
recommendations = {
'ocr_engine': 'PaddleOCR', # 默认推荐
'pdf_parser': 'PyMuPDF',
'layout_analyzer': 'LayoutParser',
'table_extractor': 'img2table',
'llm': 'gpt-4o',
'framework': 'FastAPI'
}
# 根据具体需求调整
if '中文' in requirements.get('languages', []):
recommendations['ocr_engine'] = 'PaddleOCR'
elif len(requirements.get('languages', [])) > 3:
recommendations['ocr_engine'] = 'EasyOCR'
if requirements.get('accuracy_priority') == '极高':
recommendations['ocr_engine'] = 'Surya'
recommendations['llm'] = 'gpt-4o'
if requirements.get('volume') == '大':
recommendations['framework'] = 'FastAPI + Celery'
return recommendations
14.2 性能优化建议
| 优化方向 | 具体措施 | 预期效果 |
|---|---|---|
| OCR加速 | 使用GPU推理、批量处理 | 5-10倍提速 |
| 并行处理 | 多线程/多进程处理 | 线性提升 |
| 缓存机制 | 缓存OCR结果和解析结果 | 避免重复计算 |
| 模型优化 | 量化、蒸馏模型 | 减少内存和计算 |
| 异步IO | 使用asyncio处理IO密集操作 | 提高吞吐量 |
14.3 常见问题与解决方案
- OCR识别率低：尝试图像预处理（去噪、矫正、增强对比度）
- 表格提取不准确：结合多种提取方法，取交集或投票
- 中文识别乱码：确保使用中文OCR模型，检查编码设置
- PDF解析不完整：区分文本型和扫描型PDF，使用不同策略
- 批量处理内存溢出：使用流式处理，限制并发数量
14.4 总结
本教程全面介绍了AI文档智能与数据提取的核心技术栈：
- OCR技术：从PaddleOCR到Surya，覆盖不同场景需求
- PDF解析：PyMuPDF、pdfplumber、Unstructured三大方案
- 表格处理：Table Transformer和img2table的实战应用
- 多模态理解：VLM模型在文档理解中的前沿应用
- 结构化提取：Instructor和LangChain的LLM提取方案
- 端到端Pipeline：从文档输入到结构化输出的完整流程
- 生产部署：FastAPI服务、Docker容器化、批量工作流
文档智能是一个快速发展的领域，随着多模态大模型的进步，未来将实现更精准、更智能的文档理解与处理能力。开发者应根据实际业务需求，选择合适的技术组合，构建高效的文档处理系统。
📌 推荐阅读
- PaddleOCR官方文档：https://github.com/PaddlePaddle/PaddleOCR
- Unstructured官方文档：https://unstructured.io/
- Instructor库文档：https://jxnl.github.io/instructor/
- Microsoft Table Transformer：https://github.com/microsoft/table-transformer
