多模态大模型实践：图像理解与视觉分析

from openai import OpenAI

client = OpenAI()

def analyze_image_from_url(image_url: str, question: str) -> str:
    response = client.chat.completions.create(
        model="gpt-5.4",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": image_url,
                            "detail": "high"  # low/high/auto，high最精细但token更多
                        }
                    },
                    {
                        "type": "text",
                        "text": question
                    }
                ]
            }
        ],
        max_tokens=1024
    )
    return response.choices[0].message.content

# 使用
result = analyze_image_from_url(
    "https://example.com/architecture-diagram.png",
    "描述这张架构图中的组件和它们之间的数据流"
)

import base64
from pathlib import Path

def encode_image(image_path: str) -> str:
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode("utf-8")

def analyze_local_image(image_path: str, question: str, model: str = "gpt-5.4") -> str:
    # 检测文件格式
    suffix = Path(image_path).suffix.lower()
    media_type_map = {
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".png": "image/png",
        ".gif": "image/gif",
        ".webp": "image/webp"
    }
    media_type = media_type_map.get(suffix, "image/png")

    b64_image = encode_image(image_path)

    response = client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:{media_type};base64,{b64_image}"
                        }
                    },
                    {"type": "text", "text": question}
                ]
            }
        ],
        max_tokens=2048
    )
    return response.choices[0].message.content

def compare_images(image_paths: list[str], comparison_question: str) -> str:
    """对比多张图，例如对比两个版本的UI截图差异"""
    content = []

    for i, path in enumerate(image_paths):
        content.append({
            "type": "text",
            "text": f"图片 {i+1}："
        })
        content.append({
            "type": "image_url",
            "image_url": {
                "url": f"data:image/png;base64,{encode_image(path)}"
            }
        })

    content.append({"type": "text", "text": comparison_question})

    response = client.chat.completions.create(
        model="gpt-5.4",
        messages=[{"role": "user", "content": content}],
        max_tokens=2048
    )
    return response.choices[0].message.content

def extract_chart_data(chart_image_path: str) -> dict:
    prompt = """分析这张图表，提取以下信息（用JSON格式返回）：
    1. 图表类型（折线图/柱状图/饼图等）
    2. X轴和Y轴的含义及单位
    3. 数据系列名称
    4. 关键数值（最大值、最小值、趋势）
    5. 时间范围（如果有）

    只返回JSON，不要其他说明。格式：
    {
        "chart_type": "",
        "x_axis": {"label": "", "unit": ""},
        "y_axis": {"label": "", "unit": ""},
        "series": [],
        "key_values": {},
        "time_range": ""
    }
    """

    result = analyze_local_image(chart_image_path, prompt)

    # 提取 JSON
    import json, re
    json_match = re.search(r'\{.*\}', result, re.DOTALL)
    if json_match:
        return json.loads(json_match.group())
    return {"raw": result}

def extract_document_info(doc_image_path: str, extraction_template: dict) -> dict:
    """
    从证件/表单截图提取结构化信息
    extraction_template 定义要提取的字段
    """
    fields_desc = "\n".join([
        f"- {field}: {desc}"
        for field, desc in extraction_template.items()
    ])

    prompt = f"""从这张图片中提取以下信息，以JSON格式返回：

{fields_desc}

如果某个字段在图片中找不到，值设为null。只返回JSON。"""

    result = analyze_local_image(doc_image_path, prompt)

    import json, re
    json_match = re.search(r'\{.*\}', result, re.DOTALL)
    if json_match:
        try:
            return json.loads(json_match.group())
        except json.JSONDecodeError:
            pass
    return {"raw": result}

# 使用示例
template = {
    "order_id": "订单编号",
    "amount": "金额（数字）",
    "date": "日期（YYYY-MM-DD格式）",
    "status": "状态",
    "items": "商品列表（数组）"
}

data = extract_document_info("order_screenshot.png", template)

def detect_ui_regression(baseline_path: str, current_path: str) -> dict:
    """检测UI视觉回归，对比基准截图和当前截图"""
    prompt = """对比这两张UI截图（图1是基准版本，图2是当前版本）。

请列出：
1. 视觉差异（布局、颜色、字体、间距等变化）
2. 内容差异（文字、图片、组件变化）
3. 是否存在明显的UI错误（元素重叠、截断、错位等）
4. 总体评估：变化是正常的设计更新还是潜在的regression

以JSON格式返回：
{
    "visual_diffs": [],
    "content_diffs": [],
    "ui_errors": [],
    "assessment": "normal|regression|needs_review",
    "summary": ""
}"""

    return compare_images([baseline_path, current_path], prompt)

import httpx
import os
from datetime import datetime, timedelta

def capture_grafana_panel(
    grafana_url: str,
    dashboard_uid: str,
    panel_id: int,
    api_key: str,
    from_time: str = "now-1h",
    to_time: str = "now",
    width: int = 1000,
    height: int = 500
) -> bytes:
    """
    使用 Grafana Render API 截图
    需要 Grafana 安装 rendering 插件
    """
    render_url = (
        f"{grafana_url}/render/d-solo/{dashboard_uid}"
        f"?panelId={panel_id}"
        f"&from={from_time}&to={to_time}"
        f"&width={width}&height={height}"
        f"&theme=light"  # 白底更适合 LLM 分析
    )

    response = httpx.get(
        render_url,
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=30
    )
    response.raise_for_status()
    return response.content

def save_panel_screenshot(panel_bytes: bytes, output_path: str):
    with open(output_path, "wb") as f:
        f.write(panel_bytes)

import anthropic

# Claude 在图表分析上特别准确
anthropic_client = anthropic.Anthropic()

def analyze_grafana_alert(
    screenshot_path: str,
    metric_name: str,
    alert_threshold: float,
    service_name: str
) -> dict:
    """
    分析 Grafana 面板截图，生成告警摘要
    """
    with open(screenshot_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode()

    prompt = f"""你是一位有经验的SRE工程师，正在分析一个告警。

服务名称：{service_name}
监控指标：{metric_name}
告警阈值：{alert_threshold}

请分析这张Grafana监控截图，回答以下问题：

1. **当前状态**：指标当前值是多少？是否超过阈值？
2. **趋势分析**：过去1小时的趋势如何？（急剧上升/缓慢增长/平稳/下降）
3. **异常时间点**：如果有异常，大约在什么时间开始？
4. **严重程度**：评估为 critical/warning/info
5. **可能原因**：基于指标形态，列出2-3个可能的原因
6. **建议行动**：列出立即需要做的排查步骤

以JSON格式返回：
{{
    "current_value": null,
    "is_breaching": false,
    "trend": "",
    "anomaly_start_time": null,
    "severity": "info",
    "possible_causes": [],
    "recommended_actions": [],
    "summary": "一句话告警摘要"
}}"""

    response = anthropic_client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/png",
                            "data": image_data
                        }
                    },
                    {"type": "text", "text": prompt}
                ]
            }
        ]
    )

    import json, re
    result_text = response.content[0].text
    json_match = re.search(r'\{.*\}', result_text, re.DOTALL)
    if json_match:
        try:
            return json.loads(json_match.group())
        except json.JSONDecodeError:
            pass
    return {"raw": result_text}

# 完整告警处理流程
def handle_grafana_alert(alert_webhook: dict):
    """
    接收 Grafana webhook，截图分析，发送到钉钉/Slack
    """
    panel_bytes = capture_grafana_panel(
        grafana_url=os.environ["GRAFANA_URL"],
        dashboard_uid=alert_webhook["dashboardUID"],
        panel_id=alert_webhook["panelId"],
        api_key=os.environ["GRAFANA_API_KEY"]
    )

    screenshot_path = f"/tmp/alert_{alert_webhook['alertId']}.png"
    save_panel_screenshot(panel_bytes, screenshot_path)

    analysis = analyze_grafana_alert(
        screenshot_path=screenshot_path,
        metric_name=alert_webhook["ruleName"],
        alert_threshold=alert_webhook.get("threshold", 0),
        service_name=alert_webhook.get("labels", {}).get("service", "unknown")
    )

    # 构建通知消息
    severity_emoji = {"critical": "🔴", "warning": "🟡", "info": "🔵"}
    message = f"""
{severity_emoji.get(analysis.get('severity', 'info'), '⚪')} **告警分析**

**摘要**：{analysis.get('summary', '无')}
**严重程度**：{analysis.get('severity', 'unknown')}
**趋势**：{analysis.get('trend', '未知')}

**可能原因**：
{chr(10).join(f"- {c}" for c in analysis.get('possible_causes', []))}

**建议行动**：
{chr(10).join(f"- {a}" for a in analysis.get('recommended_actions', []))}
"""
    return message

import cv2
import numpy as np

def extract_key_frames(video_path: str, num_frames: int = 10) -> list[str]:
    """均匀抽取关键帧"""
    cap = cv2.VideoCapture(video_path)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    frame_indices = np.linspace(0, total_frames - 1, num_frames, dtype=int)

    frame_paths = []
    for i, idx in enumerate(frame_indices):
        cap.set(cv2.CAP_PROP_POS_FRAMES, idx)
        ret, frame = cap.read()
        if ret:
            path = f"/tmp/frame_{i:03d}.jpg"
            cv2.imwrite(path, frame)
            frame_paths.append(path)

    cap.release()
    return frame_paths

def analyze_video(video_path: str, question: str) -> str:
    """通过关键帧分析视频内容"""
    frame_paths = extract_key_frames(video_path, num_frames=8)

    content = [{"type": "text", "text": f"以下是视频的{len(frame_paths)}个关键帧（按时间顺序）："}]

    for path in frame_paths:
        content.append({
            "type": "image_url",
            "image_url": {"url": f"data:image/jpeg;base64,{encode_image(path)}"}
        })

    content.append({"type": "text", "text": question})

    response = client.chat.completions.create(
        model="gpt-5.4",
        messages=[{"role": "user", "content": content}],
        max_tokens=2048
    )
    return response.choices[0].message.content

# Gemini 2.5 Pro 原生视频上传示例（适合长视频）
def analyze_video_gemini(video_path: str, question: str) -> str:
    """使用 Gemini 2.5 Pro 直接分析视频文件"""
    import google.generativeai as genai

    genai.configure(api_key=os.environ["GEMINI_API_KEY"])
    model = genai.GenerativeModel("gemini-2.5-pro")

    video_file = genai.upload_file(path=video_path)
    response = model.generate_content([video_file, question])
    return response.text

def resize_for_analysis(image_path: str, max_dimension: int = 1024) -> str:
    """缩小图片以节省 token"""
    from PIL import Image
    img = Image.open(image_path)
    img.thumbnail((max_dimension, max_dimension), Image.LANCZOS)
    output_path = image_path.replace(".", "_resized.")
    img.save(output_path, quality=85)
    return output_path