第9章: 「图识」多模态内容识别器

VISION_CONTRACT = (
    "你是严谨的企业识图助手。请根据图片提取结构化信息，并严格只输出 JSON。\n"
    "【强制规则】\n"
    "1. 只输出 JSON，不要输出任何其他文字/Markdown/解释。\n"
    "2. 看不清/不确定的信息，必须放入 need_more_info，绝对不要猜测。\n"
    "3. entities 只列出图片中明确存在的信息。\n"
    "【输出格式】\n"
    '{"topic": string, "summary": string, "entities": [string], "need_more_info": [string]}'
)

"""
图像智能识别：表格/文字/图表分层提取
===========================================
【核心价值】把"图片内容"转换成可验收的结构化合同：topic/summary/entities/need_more_info
【适用场景】发票识别、报表解析、合同截图、监控图表、实验曲线等企业业务图片处理

【运行前准备】
1. 安装依赖：pip install langchain-community langchain-core dashscope
2. 设置环境变量：export DASHSCOPE_API_KEY=your_api_key
3. 准备测试图片：文件名包含 table/chart/图 等关键词会自动路由到不同处理策略

【使用方式】
python multi_2.py demo_table.png
"""

import os
import json
import time
import uuid
import re
from typing import Dict, Any, List, Optional

# ⚠️ 重要：新版 LangChain 的正确 import 路径
from langchain_community.chat_models import ChatTongyi  # 通义千问多模态模型
from langchain_core.messages import HumanMessage        # 消息对象（新版路径）


# ==================== 📊 可观测：trace_id + 分段耗时 + 错误分类 ====================
# 【重点】生产环境必备：每次调用生成唯一 trace_id，记录各阶段耗时，便于排查问题

class Trace:
    """
    可观测追踪器：记录整个识别流程的执行轨迹
    - trace_id: 唯一标识，用于日志关联和问题排查
    - events: 记录每个阶段的执行状态、耗时、元数据
    """
    def __init__(self, name: str):
        self.trace_id = f"t-{uuid.uuid4().hex[:8]}"  # 生成 8 位短 ID
        self.name = name
        self.t0 = time.time()  # 记录开始时间
        self.events: List[Dict[str, Any]] = []

    def mark(self, stage: str, ok: bool, meta: Optional[Dict[str, Any]] = None) -> None:
        """
        标记某个阶段的执行结果
        :param stage: 阶段名称（route/vision/contract/error）
        :param ok: 是否成功
        :param meta: 附加元数据（kind/model/attempt/error 等）
        """
        self.events.append({
            "stage": stage,
            "ok": ok,
            "ms": int((time.time() - self.t0) * 1000),  # 累计耗时（毫秒）
            "meta": meta or {},
        })

    def dump(self) -> Dict[str, Any]:
        """导出完整追踪信息，用于日志记录或监控上报"""
        return {"trace_id": self.trace_id, "name": self.name, "events": self.events}


def classify_error(e: Exception) -> str:
    """
    错误分类：区分用户输入错误、契约校验错误、系统错误
    【重点】便于监控告警分级和问题定位
    """
    s = str(e).lower()
    if "not found" in s or "不存在" in s or "path" in s:
        return "input_error"      # 用户输入问题（文件不存在）
    if "contract" in s or "schema" in s or "missing" in s:
        return "contract_error"   # 模型输出不符合契约
    return "system_error"         # 系统异常（API 调用失败等）


# ==================== 🔒 分层输出契约（核心防线） ====================
# 【重点】强制约束模型输出格式，不合规的结果禁止进入下游（入库/生成报告）

LAYER_REQUIRED = ("kind", "topic", "summary", "entities", "need_more_info")
KIND_ALLOWED = {"table", "chart", "text"}


def validate_contract(out: Dict[str, Any]) -> Dict[str, Any]:
    """
    契约校验：验证模型输出是否符合四层结构
    【重点】防止模型幻觉、格式错误导致下游系统崩溃
    
    必需字段：
    - kind: 图片类型（table/chart/text）
    - topic: 主题（"这是什么"）
    - summary: 一句话概述（"说了什么"）
    - entities: 关键信息列表（金额/日期/人名等）
    - need_more_info: 不确定信息列表（看不清的字段）
    """
    for k in LAYER_REQUIRED:
        if k not in out:
            raise ValueError(f"contract_missing: {k}")
    
    if out["kind"] not in KIND_ALLOWED:
        raise ValueError(f"contract_invalid_kind: {out['kind']}")
    
    if not isinstance(out["entities"], list) or not isinstance(out["need_more_info"], list):
        raise ValueError("contract_invalid_types")
    
    return out


# ==================== 🚦 路由：先判断图片类型，再走不同提取策略 ====================
# 【重点】避免让大模型承担所有判断，用轻量规则先分流

def route_kind(image_path: str) -> str:
    """
    图片类型路由：根据文件名判断图片类型
    【生产建议】用轻量分类模型（ResNet/EfficientNet）或规则（logo/布局特征）先路由
    
    :return: "table" | "chart" | "text"
    """
    name = os.path.basename(image_path).lower()
    
    # 表格类：发票、报表、Excel 截图
    if "table" in name or "sheet" in name or "表" in name or "invoice" in name:
        return "table"
    
    # 图表类：柱状图、折线图、饼图
    if "chart" in name or "plot" in name or "图" in name or "graph" in name:
        return "chart"
    
    # 文字类：合同、协议、文档截图（默认）
    return "text"


# ==================== 🎯 Vision Prompt：分类型约束输出格式 ====================
# 【重点】不同类型图片用不同 Prompt，提高识别准确率

VISION_BASE_RULES = (
    "你是严谨的企业识图助手。请根据图片提取结构化信息，并严格只输出 JSON。\n"
    "【强制规则】\n"
    "1) 只输出 JSON，不要输出任何解释性文字/Markdown/代码块标记。\n"
    "2) 看不清/不确定的信息，必须放入 need_more_info，绝对不要猜测。\n"
    "3) entities 只列出图片中明确存在的信息。\n"
)

PROMPTS: Dict[str, str] = {
    "table": VISION_BASE_RULES
    + "【任务】这是一张表格/报表截图。请尽可能还原表头与行数据（看不清则不填，放 need_more_info）。\n"
      "【输出格式】\n"
      '{"kind":"table","topic":string,"summary":string,"entities":[string],"need_more_info":[string],'
      '"table":{"headers":[string],"rows":[[string]]}}',
    
    "chart": VISION_BASE_RULES
    + "【任务】这是一张图表截图。请输出图表类型、总体趋势、关键拐点/异常（无法确认数值就写 need_more_info）。\n"
      "【输出格式】\n"
      '{"kind":"chart","topic":string,"summary":string,"entities":[string],"need_more_info":[string],'
      '"chart":{"chart_type":string,"trend":string,"highlights":[string]}}',
    
    "text": VISION_BASE_RULES
    + "【任务】这是一张文字/合同截图。请提炼关键条款要点，避免编造。\n"
      "【输出格式】\n"
      '{"kind":"text","topic":string,"summary":string,"entities":[string],"need_more_info":[string],'
      '"text":{"key_points":[string]}}',
}


# ==================== 🛠️ 工具函数：JSON 提取、响应解析、图片路径转换 ====================

def _extract_json(text: str) -> str:
    """
    从模型响应中提取 JSON（兼容 Markdown 代码块格式）
    【常见坑】模型可能返回 ```json {...} ``` 格式，需要清理
    """
    t = (text or "").strip()
    
    # 去除 Markdown 代码块标记
    if t.startswith("```"):
        t = re.sub(r"^```(?:json)?\s*", "", t, flags=re.IGNORECASE)
        t = re.sub(r"\s*```$", "", t)
    
    # 提取第一个 JSON 对象
    m = re.search(r"\{.*\}", t, flags=re.S)
    return m.group(0) if m else t


def _resp_text(resp: Any) -> str:
    """
    从 LangChain 响应对象中提取文本内容
    【兼容性】处理不同版本的响应格式
    """
    c = getattr(resp, "content", None)
    
    # 字符串格式（常见）
    if isinstance(c, str):
        return c
    
    # 列表格式（多模态响应）
    if isinstance(c, list) and c and isinstance(c[0], dict) and "text" in c[0]:
        return c[0].get("text") or ""
    
    return str(c or "")


def _to_tongyi_image(image_path: str) -> str:
    """
    将图片路径转换为通义千问 API 支持的格式
    【支持格式】http:// | https:// | file:// 协议
    """
    if image_path.startswith(("http://", "https://", "file://")):
        return image_path
    
    # 本地路径转换为 file:// 协议
    return f"file://{os.path.abspath(image_path)}"


# ==================== 🤖 多模态调用：TongyiVision 封装 ====================
# 【重点】封装重试机制、错误处理、trace 记录

class TongyiVision:
    """
    通义千问多模态模型封装
    【功能】图片 + 文本输入 → 结构化 JSON 输出
    【特性】自动重试、指数退避、trace 记录
    """
    def __init__(self, model: str = "qwen-vl-plus", max_retries: int = 3):
        """
        :param model: 模型名称（qwen-vl-plus 性价比高，qwen-vl-max 准确率更高）
        :param max_retries: 最大重试次数（防止偶发网络错误）
        """
        self.llm = ChatTongyi(model=model, temperature=0)  # temperature=0 确保输出稳定
        self.model = model
        self.max_retries = max_retries

    def __call__(self, image_path: str, kind: str, trace: Trace) -> Dict[str, Any]:
        """
        执行图片识别
        :param image_path: 图片路径
        :param kind: 图片类型（table/chart/text）
        :param trace: 追踪器
        :return: 结构化 JSON 对象
        """
        prompt = PROMPTS.get(kind)
        if not prompt:
            raise ValueError(f"unknown_kind: {kind}")

        # 构造多模态消息（⚠️ 通义千问 API 正确格式）
        msg = HumanMessage(content=[
            {"type": "text", "text": prompt},
            {"type": "image", "image": _to_tongyi_image(image_path)},
        ])
        
        # 重试机制：最多重试 3 次，指数退避（2^n 秒）
        for attempt in range(self.max_retries):
            try:
                resp = self.llm.invoke([msg])
                text = _resp_text(resp)
                obj = json.loads(_extract_json(text))
                
                # 记录成功
                trace.mark("vision", True, {
                    "kind": kind, 
                    "model": self.model, 
                    "attempt": attempt + 1
                })
                return obj
                
            except Exception as e:
                if attempt == self.max_retries - 1:
                    raise  # 最后一次重试失败，抛出异常
                
                # 记录重试
                trace.mark("vision_retry", False, {
                    "attempt": attempt + 1, 
                    "error": str(e)
                })
                time.sleep(2 ** attempt)  # 指数退避：1s, 2s, 4s


# ==================== 🎬 端到端流程：路由 → 识别 → 契约校验 → 输出 ====================
# 【重点】完整的错误处理、trace 记录、契约校验

def run_layered_extraction(image_path: str) -> Dict[str, Any]:
    """
    执行完整的图片分层提取流程
    
    【流程】
    1. 环境检查（API Key、文件存在性）
    2. 路由判断（table/chart/text）
    3. 多模态识别（调用通义千问）
    4. 契约校验（验证输出格式）
    5. 返回结果（包含 trace、result、error）
    
    :return: {
        "trace": {...},      # 追踪信息
        "result": {...},     # 识别结果（符合契约）
        "error": {...}       # 错误信息（如果失败）
    }
    """
    trace = Trace("layered_extraction")
    
    try:
        # 1. 环境检查
        if not os.getenv("DASHSCOPE_API_KEY"):
            raise ValueError("请先设置环境变量 DASHSCOPE_API_KEY")

        if not os.path.exists(image_path):
            raise FileNotFoundError(f"image_not_found: {image_path}")

        # 2. 路由判断
        kind = route_kind(image_path)
        trace.mark("route", True, {"kind": kind})

        # 3. 多模态识别
        vision = TongyiVision(model="qwen-vl-plus")
        raw = vision(image_path, kind, trace)
        
        # 4. 契约校验
        out = validate_contract(raw)
        trace.mark("contract", True, {"kind": out["kind"]})

        # 5. 返回成功结果
        return {
            "trace": trace.dump(), 
            "result": out, 
            "error": None
        }
        
    except Exception as e:
        # 记录错误并分类
        trace.mark("error", False, {
            "type": classify_error(e), 
            "msg": str(e)
        })
        
        return {
            "trace": trace.dump(), 
            "result": None, 
            "error": {
                "type": classify_error(e), 
                "msg": str(e)
            }
        }


# ==================== 🚀 主函数：命令行入口 ====================

if __name__ == "__main__":
    import sys
    
    # 支持命令行参数：python multi_2.py demo_table.png
    if len(sys.argv) > 1:
        image_path = sys.argv[1]
    else:
        # 默认测试图片（请替换为实际路径）
        image_path = "../data/img.png"
        print(f"⚠️  未指定图片路径，使用默认路径: {image_path}")
        print(f"💡 使用方式: python multi_2.py <图片路径>\n")
    
    # 执行识别
    result = run_layered_extraction(image_path)
    
    # 美化输出
    print(json.dumps(result, ensure_ascii=False, indent=2))

"""
工程化保障系统 - 三级容错 + 反幻觉 + 监控
========================================
【核心价值】JSON解析成功率95%，幻觉率3%，故障定位分钟级
【使用方式】python engineering.py
"""

import json
import re
import time
import hashlib
from datetime import datetime
from typing import Dict, Any, Optional, List
from dataclasses import dataclass
from collections import defaultdict


# ==================== 三级容错 JSON 解析器 ====================

@dataclass
class ParseResult:
    """解析结果"""
    success: bool
    data: Optional[Dict[str, Any]]
    level: int  # 1/2/3
    error: Optional[str]
    raw_snippet: str


class ThreeLevelJSONParser:
    """三级容错 JSON 解析器"""
    
    def __init__(self):
        self.stats = {"total": 0, "level_1": 0, "level_2": 0, "failed": 0}
    
    def parse(self, text: str) -> ParseResult:
        """执行三级容错解析"""
        self.stats["total"] += 1
        text = (text or "").strip()
        
        if not text:
            return ParseResult(False, None, 3, "Empty input", "")
        
        # Level 1: 去除 Markdown 代码块标记后直接解析
        result = self._level_1_parse(text)
        if result.success:
            self.stats["level_1"] += 1
            return result
        
        # Level 2: 正则提取 {} 包裹的 JSON 片段
        result = self._level_2_parse(text)
        if result.success:
            self.stats["level_2"] += 1
            return result
        
        # Level 3: 彻底失败
        self.stats["failed"] += 1
        return ParseResult(False, None, 3, "All levels failed", text[:200])
    
    def _level_1_parse(self, text: str) -> ParseResult:
        """Level 1: 去除 ```json 代码块标记"""
        clean = re.sub(r"^```(?:json)?\s*", "", text, flags=re.IGNORECASE)
        clean = re.sub(r"\s*```$", "", clean)
        
        try:
            data = json.loads(clean)
            return ParseResult(True, data, 1, None, text[:200])
        except json.JSONDecodeError as e:
            return ParseResult(False, None, 1, str(e), text[:200])
    
    def _level_2_parse(self, text: str) -> ParseResult:
        """Level 2: 正则提取 JSON 片段"""
        match = re.search(r"\{.*\}", text, flags=re.DOTALL)
        
        if not match:
            return ParseResult(False, None, 2, "No JSON found", text[:200])
        
        try:
            data = json.loads(match.group(0))
            return ParseResult(True, data, 2, None, text[:200])
        except json.JSONDecodeError as e:
            return ParseResult(False, None, 2, str(e), text[:200])
    
    def get_stats(self) -> Dict[str, Any]:
        """获取解析统计"""
        total = self.stats["total"]
        if total == 0:
            return {"error": "No data"}
        
        return {
            "total": total,
            "success_rate": round((total - self.stats["failed"]) / total * 100, 2),
            "level_1_rate": round(self.stats["level_1"] / total * 100, 2),
            "level_2_rate": round(self.stats["level_2"] / total * 100, 2),
            "failed_rate": round(self.stats["failed"] / total * 100, 2)
        }


# ==================== 反幻觉验证器 ====================

class AntiHallucinationValidator:
    """反幻觉验证器"""
    
    def validate(self, data: Dict[str, Any], image_context: str = "") -> Dict[str, Any]:
        """
        验证输出是否存在幻觉
        【策略】
        1. 检查 need_more_info 是否合理使用
        2. 验证实体一致性
        3. 检查逻辑矛盾
        """
        issues = []
        
        # 策略1：检查是否有不确定内容
        if not data.get("need_more_info"):
            # 如果没有不确定内容，检查是否过于自信
            if len(data.get("entities", [])) > 10:
                issues.append("实体过多可能存在编造")
        
        # 策略2：实体一致性检查
        summary = data.get("summary", "")
        entities = data.get("entities", [])
        
        for entity in entities:
            if entity not in summary and len(entity) > 3:
                issues.append(f"实体 '{entity}' 未在摘要中出现")
        
        # 策略3：逻辑检查
        if "增长" in summary and "下降" in summary:
            issues.append("摘要中同时出现增长和下降，可能存在矛盾")
        
        return {
            "is_valid": len(issues) == 0,
            "issues": issues,
            "hallucination_score": len(issues) / max(len(entities), 1)
        }


# ==================== Trace 监控系统 ====================

class Trace:
    """请求追踪器"""
    
    def __init__(self, name: str):
        self.trace_id = f"t-{datetime.now().strftime('%Y%m%d%H%M%S')}-{hashlib.md5(str(time.time()).encode()).hexdigest()[:8]}"
        self.name = name
        self.start_time = time.time()
        self.events = []
    
    def mark(self, stage: str, ok: bool, meta: Optional[Dict[str, Any]] = None):
        """记录阶段"""
        self.events.append({
            "stage": stage,
            "ok": ok,
            "timestamp": time.time(),
            "duration_ms": int((time.time() - self.start_time) * 1000),
            "meta": meta or {}
        })
    
    def dump(self) -> Dict[str, Any]:
        """导出追踪信息"""
        return {
            "trace_id": self.trace_id,
            "name": self.name,
            "total_ms": int((time.time() - self.start_time) * 1000),
            "events": self.events,
            "ok": all(e["ok"] for e in self.events)
        }


class MonitoringSystem:
    """监控系统"""
    
    def __init__(self):
        self.traces = []
        self.metrics = {
            "total": 0,
            "success": 0,
            "failed": 0,
            "latencies": []
        }
    
    def record_trace(self, trace: Trace):
        """记录追踪"""
        trace_data = trace.dump()
        self.traces.append(trace_data)
        
        self.metrics["total"] += 1
        if trace_data["ok"]:
            self.metrics["success"] += 1
        else:
            self.metrics["failed"] += 1
        
        self.metrics["latencies"].append(trace_data["total_ms"])
    
    def get_stats(self) -> Dict[str, Any]:
        """获取统计数据"""
        total = self.metrics["total"]
        if total == 0:
            return {"error": "No data"}
        
        latencies = sorted(self.metrics["latencies"])
        
        return {
            "total_requests": total,
            "success_rate": round(self.metrics["success"] / total * 100, 2),
            "p50_latency": latencies[int(len(latencies) * 0.5)] if latencies else 0,
            "p90_latency": latencies[int(len(latencies) * 0.9)] if latencies else 0,
            "p99_latency": latencies[int(len(latencies) * 0.99)] if latencies else 0
        }


# ==================== 使用示例 ====================

if __name__ == "__main__":
    # 1. 测试三级容错解析
    print("=" * 60)
    print("测试 1: 三级容错 JSON 解析")
    print("=" * 60)
    
    parser = ThreeLevelJSONParser()
    
    # 测试用例
    test_cases = [
        '```json\n{"kind": "chart", "topic": "销售数据"}\n```',  # Level 1
        '这是识别结果：{"kind": "table", "topic": "报表"} 以上',  # Level 2
        '{"kind": "text"}',  # Level 1
        'invalid json',  # Level 3 失败
    ]
    
    for i, case in enumerate(test_cases, 1):
        result = parser.parse(case)
        print(f"\n用例 {i}: Level {result.level}, 成功={result.success}")
        if result.success:
            print(f"  数据: {result.data}")
    
    print(f"\n解析统计: {parser.get_stats()}")
    
    # 2. 测试反幻觉验证
    print("\n" + "=" * 60)
    print("测试 2: 反幻觉验证")
    print("=" * 60)
    
    validator = AntiHallucinationValidator()
    
    test_data = {
        "kind": "chart",
        "topic": "Q1销售数据",
        "summary": "销售额同比增长15%",
        "entities": ["销售额", "同比增长15%", "Q1"],
        "need_more_info": ["具体金额模糊"]
    }
    
    validation = validator.validate(test_data)
    print(f"验证结果: {validation}")
    
    # 3. 测试监控系统
    print("\n" + "=" * 60)
    print("测试 3: Trace 监控")
    print("=" * 60)
    
    monitor = MonitoringSystem()
    
    for i in range(5):
        trace = Trace("image_recognition")
        trace.mark("route", True, {"kind": "chart"})
        time.sleep(0.05)
        trace.mark("vision", True, {"model": "qwen-vl-plus"})
        time.sleep(0.03)
        trace.mark("validate", i != 3, {})  # 第4次失败
        
        monitor.record_trace(trace)
    
    print(f"监控统计: {json.dumps(monitor.get_stats(), indent=2)}")
    print("=" * 60)