多模态实战：VLM 文档理解与图片问答

def select_vlm(task: str) -> str:
    if "extract_structured_data" in task or "ocr" in task:
        return "claude-sonnet-4-5"  # Claude 文档能力最强
    elif "video_understanding" in task:
        return "gemini-2.5-pro"  # Gemini 视频最强
    elif "real_time" in task or "voice" in task:
        return "gpt-4o"  # GPT-4o 实时能力最强
    elif "self_hosted" in task or "cost" in task:
        return "qwen-2.5-vl-72b"  # 开源 SOTA
    else:
        return "claude-sonnet-4-5"  # 默认

import anthropic
import base64

def extract_invoice(image_path: str) -> dict:
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode()
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/jpeg",
                        "data": image_data
                    }
                },
                {
                    "type": "text",
                    "text": """从这张发票图片中提取信息，严格按 JSON 格式输出：

{
    "invoice_number": "发票号码",
    "date": "YYYY-MM-DD",
    "seller": "销售方名称",
    "buyer": "购买方名称",
    "items": [
        {"name": "商品名称", "quantity": 数量, "unit_price": 单价, "amount": 金额}
    ],
    "total_amount": "总金额（含税）",
    "tax_id": "税号"
}

如果某字段无法识别，用 null。
"""
                }
            ]
        }]
    )
    
    return json.loads(response.content[0].text)

# 方案 1：整页送进 VLM（适合 ≤20 页）
def process_pdf_short(pdf_path):
    # 把每页转成图片
    images = pdf_to_images(pdf_path)
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        messages=[{
            "role": "user",
            "content": [
                *[{"type": "image", "source": {"type": "base64", "data": img}} for img in images],
                {"type": "text", "text": "总结这份 PDF 的核心内容..."}
            ]
        }]
    )

# 方案 2：分块 + 摘要 + 综合（适合 >20 页）
def process_pdf_long(pdf_path):
    images = pdf_to_images(pdf_path)
    
    # 每 5 页一组，分别总结
    summaries = []
    for i in range(0, len(images), 5):
        chunk = images[i:i+5]
        summary = summarize_chunk(chunk)
        summaries.append(summary)
    
    # 综合所有 summary
    final = synthesize(summaries)
    return final

# Claude 在表格抽取上特别强
TABLE_EXTRACTION_PROMPT = """提取图片中的所有表格。

要求：
1. 保留原始的行列结构
2. 数字保持原样，不要格式化
3. 空单元格用 null
4. 多个表格用 markdown 格式分别输出

输出格式：
## 表格 1
| 列1 | 列2 | 列3 |
|-----|-----|-----|
| ... | ... | ... |

## 表格 2
...
"""

def analyze_chart(image_path: str, question: str) -> str:
    """让 VLM 回答关于图表的问题"""
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        messages=[{
            "role": "user",
            "content": [
                {"type": "image", "source": {"type": "base64", "data": image_b64}},
                {"type": "text", "text": question}
            ]
        }]
    )
    
    return response.content[0].text

# 示例
answer = analyze_chart(
    "quarterly_revenue.png",
    "Q3 的收入是多少？同比增长多少？哪个产品线贡献最大？"
)

def analyze_ui_screenshot(screenshot_path: str) -> dict:
    """分析 UI 截图，提取设计信息"""
    
    prompt = """分析这张 UI 截图，按 JSON 格式输出：

{
    "layout": "整体布局描述（顶部导航 / 左侧菜单 / 主内容区 / 右侧边栏等）",
    "components": [
        {"type": "按钮/输入框/卡片/菜单", "text": "文本", "position": "位置"}
    ],
    "color_scheme": {
        "primary": "主色调 hex",
        "background": "背景色 hex",
        "text": "文字色 hex"
    },
    "issues": ["发现的可用性问题"],
    "suggestions": ["改进建议"]
}
"""
    # 调用 VLM...

# VLM 经常"自信地"识别根本不存在的物体
# 例如：图中是白色背景的简单形状，VLM 可能说"看到了一只猫"

# ❌ VLM 很难准确说"按钮在距离左边 234 像素、上边 156 像素"
# ✅ 但能说"按钮在屏幕中上部偏左"

# VLM 在数 1-5 个物体时基本准确
# 数 20+ 个物体时准确率显著下降

# 1080p 截图里 12px 的文字，VLM 经常认错

class HybridDocExtractor:
    def __init__(self):
        self.ocr = PaddleOCR(use_angle_cls=True, lang="ch")
        self.vlm = anthropic.Anthropic()
    
    def extract(self, image_path):
        # 1. OCR 提取所有文字 + 位置
        ocr_result = self.ocr.ocr(image_path, cls=True)
        
        # 2. VLM 提取结构化字段
        vlm_extraction = self.vlm_extract(image_path)
        
        # 3. 融合
        # - OCR 提供精确的文字内容
        # - VLM 提供语义归类（哪个字段是什么）
        result = self.merge(ocr_result, vlm_extraction)
        
        return result
    
    def merge(self, ocr_data, vlm_data):
        """把 OCR 的精确文字 + VLM 的语义分类合并"""
        # 示例：VLM 说"发票号"对应 OCR 的某段文字
        # 直接用 OCR 的精确值替换 VLM 的猜测
        for field, position in vlm_data["field_positions"].items():
            text_at_position = self.get_text_at(ocr_data, position)
            vlm_data["fields"][field] = text_at_position
        
        return vlm_data["fields"]

def optimize_for_vlm(image_path: str, max_size: int = 1024) -> bytes:
    """缩放到合适大小，节省 token"""
    img = Image.open(image_path)
    
    # 限制最大尺寸
    img.thumbnail((max_size, max_size))
    
    # 转 JPEG 压缩
    buf = io.BytesIO()
    img.save(buf, format="JPEG", quality=85)
    return buf.getvalue()

# low: 85 tokens（图像被压缩成 512×512）
# high: 170 tokens（图像被压缩成 1024×1024）——默认
# auto: 模型自己决定

response = openai.ChatCompletion.create(
    model="gpt-4o",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "描述这张图"},
            {
                "type": "image_url",
                "image_url": {
                    "url": "...",
                    "detail": "low"  # 节省 80% token
                }
            }
        ]
    }]
)

async def batch_process_images(images: list[str]) -> list[dict]:
    """并发处理多张图片"""
    semaphore = asyncio.Semaphore(10)  # 控制并发
    
    async def process_one(img):
        async with semaphore:
            return await extract_async(img)
    
    return await asyncio.gather(*[process_one(img) for img in images])