LLM应用开发——文档智能处理实战

LLM 应用开发系列

• Embedding 嵌入向量
• 向量数据库
• RAG 检索增强生成
• Prompt Engineering
• LangChain 框架
• Agent 智能体
• Function Calling
• 部署与优化
• 安全防护
• 对话记忆管理
• Streaming 流式输出
• LlamaIndex 框架
• 结构化输出
• 多模态 LLM 应用
• LLM 评估与测试
• 多模型路由
• 缓存策略
• 成本优化
• 监控与可观测性
• 高级 RAG 技术
• GraphRAG 知识图谱
• 长上下文处理
• LLM 微调实战
• 本地模型部署
• MCP 协议开发
• AI 搜索引擎
• 代码生成与执行
• 文档智能处理
• 语音交互系统
• 多租户架构
• Prompt 版本管理
• 混合检索策略
• 对话系统设计
• 内容审核与过滤
• 个性化推荐
• 知识库构建
• AI 写作助手
• 实时协作
• 边缘部署

from docling.datamodel.base_models import InputFormat
from docling.document_converter import DocumentConverter

def convert_with_docling(file_path: str):
    converter = DocumentConverter()
    result = converter.convert(file_path)
    
    # 导出为 Markdown
    markdown_content = result.document.export_to_markdown()
    
    # 导出为 JSON (包含完整的布局信息)
    json_content = result.document.export_to_dict()
    
    return markdown_content, json_content

# 示例
md, _ = convert_with_docling("complex_report.pdf")

from langchain_text_splitters import MarkdownHeaderTextSplitter

def hierarchical_chunking(markdown_text: str):
    headers_to_split_on = [
        ("#", "Header 1"),
        ("##", "Header 2"),
        ("###", "Header 3"),
    ]
    
    splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
    chunks = splitter.split_text(markdown_text)
    
    # 每个 chunk 都会自动带上其所属的层级元数据
    # 例如: metadata={'Header 1': '第一章', 'Header 2': '1.1 节'}
    return chunks

import base64

def parse_table_with_vlm(image_path: str):
    with open(image_path, "rb") as f:
        base64_image = base64.b64encode(f.read()).decode('utf-8')

    prompt = """
    请分析图片中的表格，并将其转换为结构化的 JSON 格式。
    要求：
    1. 识别合并单元格。
    2. 保持数值的精确性。
    3. 输出格式为: {"tables": [{"headers": [], "rows": []}]}
    """
    # 调用 GPT-4o API...
    pass

---

#### 行业革命：ColPali (OCR-Free 检索)

传统的文档 RAG 流程是：`PDF -> OCR -> 文本 -> 向量化 -> 检索`。
**ColPali** 彻底颠覆了这一流程：`PDF -> 图像块 -> 多模态向量化 -> 检索`。

#### 为什么 ColPali 更好？
1.  **零丢失**：不再受 OCR 识别错误的影响。
2.  **视觉感知**：模型能“看到”加粗的字体、红色的警告、复杂的流程图和表格排版。
3.  **极速索引**：直接对 PDF 页面截图进行编码，省去了复杂的解析步骤。

#### 架构
它基于 **PaliGemma** 多模态模型，将图像块映射到与文本查询相同的向量空间。检索时，它能直接定位到 PDF 页面中包含答案的那个“视觉区域”。

---

#### 布局分析 (Layout Analysis)

在处理长文档时，识别出哪些是“正文”，哪些是“图表说明”，哪些是“页码”至关重要。

- **工具**：使用 **LayoutParser** 或 **Unstructured.io**。
- **逻辑**：
    - 移除页眉页脚（防止检索到重复的干扰信息）。
    - 将图片和表格与其最近的标题关联起来。
    - 识别出多栏排版的阅读顺序。

---

#### 公式与学术文档处理 (Nougat)

对于包含大量数学公式的论文或技术手册，普通的解析器会把公式变成乱码。
- **方案**：使用 Meta 开源的 **Nougat** 模型。
- **特点**：它是一个“图像到序列”的模型，能直接将 PDF 页面图像翻译成高质量的 LaTeX 源码。

---

#### 总结：文档处理的“金字塔”策略

1.  **底层：纯文本 PDF** -> 使用 `PyMuPDF` 或 `pypdf` 快速提取。
2.  **中层：复杂排版/表格** -> 使用 **Docling** 或 **Marker** 转换为 Markdown。
3.  **高层：扫描件/学术文档** -> 使用 **Nougat** 或 **GPT-4o (VLM)**。
4.  **顶层：极致体验** -> 引入 **ColPali** 实现视觉增强检索。

文档处理不再是简单的“读字”，而是一场结合了计算机视觉（CV）和自然语言处理（NLP）的综合战役。
    """
    
    # 调用多模态 LLM
    # response = llm.invoke([HumanMessage(content=[{"type": "text", "text": prompt}, {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}}])])
    # return response.content

def extract_table_with_vision(image_path: str):
    """使用多模态模型提取表格"""
    prompt = "请将图片中的表格转换为 Markdown 格式。确保保留所有行和列，不要遗漏数据。"
    # 调用 GPT-4o-mini 或 Claude 3.5 Sonnet
    response = vision_model.invoke(prompt, image_path)
    return response.content

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

def analyze_complex_document(image_paths: list[str]):
    """多模态文档分析"""
    llm = ChatOpenAI(model="gpt-4o")
    
    content = [
        {"type": "text", "text": "请分析这些文档页面，总结核心财务指标，并以表格形式输出。"}
    ]
    
    for img in image_paths:
        content.append({
            "type": "image_url",
            "image_url": {"url": f"data:image/jpeg;base64,{img}"}
        })
    
    message = HumanMessage(content=content)
    return llm.invoke([message]).content

from unstructured.partition.pdf import partition_pdf

def smart_partitioning(pdf_path: str):
    """使用 Unstructured 进行布局感知分区"""
    elements = partition_pdf(
        filename=pdf_path,
        strategy="hi_res", # 使用模型进行高精度解析
        infer_table_structure=True,
        chunking_strategy="by_title", # 按标题切片
        max_characters=1000,
        combine_text_under_n_chars=200
    )
    
    return elements

# 遍历元素
for el in elements:
    if el.category == "Title":
        print(f"Heading: {el.text}")
    elif el.category == "Table":
        print(f"Table Data: {el.metadata.text_as_html}")

[上传 PDF] 
    │
    ▼
[预处理] (去噪、纠偏、转图片)
    │
    ▼
[布局分析] (识别标题、正文、表格区域)
    │       │
    │       └─→ [表格处理] (OCR + 结构化提取)
    │
    ▼
[文本提取] (Markdown 转换)
    │
    ▼
[语义切片] (基于布局信息的智能切片)
    │
    ▼
[向量化与存储] (写入向量数据库)