多模态输入处理

摄取流水线

从上传到模型可用内容，按阶段拆分责任：先挡明显无效输入，再做安全与元数据治理，最后才进入识别与结构化。

[ 上传 / URL 拉取 ]
              │
              ▼
        [ MIME · 魔数 · 大小上限 ]
              │
         ┌────┴────────────┐
         ▼                 ▼
   [ 安全扫描 ]      [ 元数据清理 ]
   (隐写/宏/沙箱)     (EXIF/敏感字段)
         │                 │
         └────────┬────────┘
                  ▼
         [ 转码 / 裁剪 / 归一化 ]
                  │
         ┌────────┴────────┐
         ▼                 ▼
   [ OCR / 视觉块 ]   [ ASR / 分段 ]
         │                 │
         └────────┬────────┘
                  ▼
    [ 统一消息 schema · token 估算 ]
                  │
                  ▼
            [ 模型 / Agent ]

图像输入：base64 vs URL 两种方式

import base64, httpx
from openai import OpenAI
from pathlib import Path

client = OpenAI()

# === 方式 1：base64 编码 ===
# 适用场景：本地文件、内网图片、需要确保图片可用性、图片较小（<4MB）
# 优点：无需公网 URL，图片生命周期可控
# 缺点：增大请求体体积（约 1.37x），大图会超出请求大小限制

def image_to_base64(image_path: str) -> str:
    """将本地图片转为 base64 data URL。"""
    path = Path(image_path)
    # 校验魔数而非仅依赖扩展名
    with open(path, "rb") as f:
        header = f.read(12)
    mime = detect_mime_from_magic(header)  # 见下方实现
    if mime not in {"image/jpeg", "image/png", "image/gif", "image/webp"}:
        raise ValueError(f"不支持的图片格式: {mime}")

    data = path.read_bytes()
    b64 = base64.standard_b64encode(data).decode()
    return f"data:{mime};base64,{b64}"

def call_vision_base64(image_path: str, question: str) -> str:
    b64_url = image_to_base64(image_path)
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": question},
                {"type": "image_url", "image_url": {"url": b64_url, "detail": "high"}},
            ]
        }],
        max_tokens=1000,
    )
    return resp.choices[0].message.content


# === 方式 2：URL 直接传入 ===
# 适用场景：公网可访问的图片、图片较大（>4MB）、CDN 上的媒体资源
# 优点：请求体小，模型直接拉取
# 缺点：需要图片公网可访问；URL 失效会导致调用失败；模型 IP 需能访问源站

def call_vision_url(image_url: str, question: str) -> str:
    # 先 HEAD 检查 URL 可访问性（避免模型调用时才发现 404）
    r = httpx.head(image_url, timeout=5, follow_redirects=True)
    if r.status_code != 200:
        raise ValueError(f"图片 URL 不可访问: {r.status_code} {image_url}")

    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": question},
                {"type": "image_url", "image_url": {"url": image_url, "detail": "auto"}},
            ]
        }],
        max_tokens=1000,
    )
    return resp.choices[0].message.content


# === 多图顺序编号（跨模态引用）===
def call_vision_multi_image(images: list[dict], question: str) -> str:
    """
    images: [{"url": "...", "caption": "图1：系统架构图"}, ...]
    在 question 中可以直接引用"图1左上角"、"图2中的红框区域"等。
    """
    content = [{"type": "text", "text": question + "\n\n图片说明：\n" +
                "\n".join(f"- {img['caption']}" for img in images)}]
    for img in images:
        content.append({
            "type": "image_url",
            "image_url": {"url": img["url"], "detail": "high"}
        })
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": content}],
        max_tokens=2000,
    )
    return resp.choices[0].message.content

# 使用示例：跨模态引用
result = call_vision_multi_image(
    images=[
        {"url": "https://cdn.example.com/arch.png", "caption": "图1：微服务架构图"},
        {"url": "https://cdn.example.com/error.png", "caption": "图2：错误日志截图"},
    ],
    question="图1左上角的 API Gateway 与图2中的 504 错误有什么关联？"
)


def detect_mime_from_magic(header: bytes) -> str:
    """根据文件魔数检测 MIME 类型（比扩展名更可靠）。"""
    if header[:2] == b"\xff\xd8":
        return "image/jpeg"
    if header[:8] == b"\x89PNG\r\n\x1a\n":
        return "image/png"
    if header[:6] in (b"GIF87a", b"GIF89a"):
        return "image/gif"
    if header[:4] == b"RIFF" and header[8:12] == b"WEBP":
        return "image/webp"
    return "application/octet-stream"

音频输入：格式转换 + API 调用

import subprocess, tempfile, os
from pathlib import Path
from openai import OpenAI

client = OpenAI()

# 步骤 1：格式转换前置处理（使用 ffmpeg）
# Whisper API 接受：flac, mp3, mp4, mpeg, mpga, m4a, ogg, wav, webm
# 最大文件大小：25MB
# 建议格式：mp3（压缩好）或 wav（无损，精度高）

def convert_audio_for_whisper(
    input_path: str,
    target_format: str = "mp3",
    sample_rate: int = 16000,
    channels: int = 1,  # 单声道，减小文件体积
) -> str:
    """将任意音频转换为 Whisper 兼容格式。需要系统安装 ffmpeg。"""
    output_path = tempfile.mktemp(suffix=f".{target_format}")
    cmd = [
        "ffmpeg", "-i", input_path,
        "-ar", str(sample_rate),    # 采样率 16kHz（语音识别标准）
        "-ac", str(channels),        # 声道数
        "-ab", "64k",                # 比特率（mp3）
        "-y",                        # 覆盖输出文件
        output_path
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
    if result.returncode != 0:
        raise RuntimeError(f"ffmpeg 转换失败: {result.stderr}")
    return output_path

# 步骤 2：检查文件大小（超 25MB 需分片）
MAX_WHISPER_SIZE = 25 * 1024 * 1024  # 25MB

def chunk_audio_if_needed(audio_path: str, chunk_duration_sec: int = 600) -> list[str]:
    """超大音频按固定时长分片（默认 10 分钟/片）。"""
    file_size = os.path.getsize(audio_path)
    if file_size <= MAX_WHISPER_SIZE:
        return [audio_path]

    chunks = []
    output_dir = tempfile.mkdtemp()
    cmd = [
        "ffmpeg", "-i", audio_path,
        "-f", "segment",
        "-segment_time", str(chunk_duration_sec),
        "-c", "copy",
        f"{output_dir}/chunk_%04d.mp3"
    ]
    subprocess.run(cmd, check=True, capture_output=True, timeout=300)
    chunks = sorted(Path(output_dir).glob("chunk_*.mp3"))
    return [str(c) for c in chunks]

# 步骤 3：调用 Whisper API
def transcribe_audio(audio_path: str, language: str = "zh") -> dict:
    """
    转录音频，返回带时间戳的文本（便于后续跨模态引用）。
    response_format="verbose_json" 包含每段的 start/end 时间戳。
    """
    # 先转换格式
    converted = convert_audio_for_whisper(audio_path)
    chunks = chunk_audio_if_needed(converted)

    segments = []
    offset = 0.0

    for chunk_path in chunks:
        with open(chunk_path, "rb") as f:
            resp = client.audio.transcriptions.create(
                model="whisper-1",
                file=f,
                language=language,
                response_format="verbose_json",  # 包含时间戳
                timestamp_granularities=["segment"],
            )
        # 调整分片内的时间戳偏移
        for seg in resp.segments:
            segments.append({
                "start": seg.start + offset,
                "end": seg.end + offset,
                "text": seg.text.strip(),
            })
        # 估算当前分片时长作为下一片偏移
        if chunks.index(chunk_path) < len(chunks) - 1:
            offset += 600.0  # 每片 600 秒

    return {
        "full_text": " ".join(s["text"] for s in segments),
        "segments": segments,  # 带时间戳，便于引用"第 2 分 15 秒处的内容"
        "language": language,
    }

# 步骤 4：跨模态引用——在 prompt 中引用音频片段
def ask_about_audio_segment(transcript: dict, question: str) -> str:
    """基于转录结果和时间戳构建 prompt，支持具体时间引用。"""
    # 格式化带时间戳的转录文本
    formatted = "\n".join(
        f"[{s['start']:.1f}s - {s['end']:.1f}s] {s['text']}"
        for s in transcript["segments"]
    )
    prompt = f"""以下是音频转录文本（带时间戳）：

{formatted}

问题：{question}

回答时如需引用音频内容，使用"[X.Xs处]"格式标注时间点。"""
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=1000,
    )
    return resp.choices[0].message.content

结构化载荷：PDF/CSV 转文本处理流程

import csv, io
from pathlib import Path

# === PDF 转文本 ===
# pip install pymupdf（fitz）— 速度快，支持文本、图片和表格提取

def pdf_to_text(pdf_path: str, max_pages: int = 50) -> dict:
    """
    PDF 转文本，保留页码信息便于引用（"第 3 页"）。
    超过 max_pages 时截断并标注。
    """
    import fitz  # pymupdf

    doc = fitz.open(pdf_path)
    pages = []
    total_chars = 0
    MAX_CHARS = 100_000  # 约 50k token，超出截断

    for page_num in range(min(len(doc), max_pages)):
        page = doc[page_num]
        text = page.get_text("text").strip()
        if not text:
            continue  # 跳过空白页（可能是扫描图片页）
        pages.append({
            "page": page_num + 1,
            "text": text,
            "char_count": len(text),
        })
        total_chars += len(text)
        if total_chars > MAX_CHARS:
            pages[-1]["text"] = pages[-1]["text"][:MAX_CHARS - (total_chars - len(text))]
            pages[-1]["truncated"] = True
            break

    return {
        "pages": pages,
        "total_pages": len(doc),
        "extracted_pages": len(pages),
        "truncated": total_chars > MAX_CHARS,
        "full_text": "\n\n---\n\n".join(
            f"[第 {p['page']} 页]\n{p['text']}" for p in pages
        ),
    }

# === CSV 转文本（结构化摘要）===
def csv_to_text(
    csv_content: str,
    max_rows: int = 100,
    include_stats: bool = True,
) -> dict:
    """
    CSV 转文本，包含统计摘要和前 N 行数据。
    大 CSV 不直接全量传入，先生成摘要 + 样本行。
    """
    reader = csv.DictReader(io.StringIO(csv_content))
    rows = list(reader)
    headers = reader.fieldnames or []

    sample_rows = rows[:max_rows]
    truncated = len(rows) > max_rows

    # 构建数值列统计
    stats = {}
    if include_stats:
        for col in headers:
            values = []
            for row in rows:
                try:
                    values.append(float(row[col]))
                except (ValueError, TypeError):
                    pass
            if values:
                stats[col] = {
                    "min": min(values),
                    "max": max(values),
                    "avg": round(sum(values) / len(values), 2),
                    "count": len(values),
                }

    # 格式化为 markdown 表格（前 max_rows 行）
    header_row = " | ".join(headers)
    separator = " | ".join(["---"] * len(headers))
    data_rows = "\n".join(
        " | ".join(str(row.get(h, "")) for h in headers)
        for row in sample_rows
    )
    table_text = f"| {header_row} |\n| {separator} |\n" + "\n".join(
        f"| {' | '.join(str(row.get(h, '')) for h in headers)} |"
        for row in sample_rows
    )

    summary = f"CSV 文件：{len(rows)} 行 × {len(headers)} 列"
    if truncated:
        summary += f"（仅显示前 {max_rows} 行）"
    if stats:
        summary += "\n数值列统计：\n" + "\n".join(
            f"  {col}: min={s['min']}, max={s['max']}, avg={s['avg']}, count={s['count']}"
            for col, s in stats.items()
        )

    return {
        "summary": summary,
        "table": table_text,
        "headers": headers,
        "total_rows": len(rows),
        "stats": stats,
        "full_text": summary + "\n\n" + table_text,
    }

# 跨模态引用：prompt 中引用"图2左上角"的写法
CROSS_MODAL_PROMPT_TEMPLATE = """
图片信息：
{image_captions}

文档内容（摘要）：
{doc_summary}

问题：{question}

回答规范：
- 引用图片区域时用「图N的[位置]」（如"图1左上角的流程图"）
- 引用文档内容时用「文档第N页」或「表格第N行」
- 如果图片和文档内容相互印证，明确指出关联
"""

MIME 与体积估算

浏览器报告的 type 可能为空（依赖扩展名猜测不可靠）；服务端应以魔数为准。下方在本地选一个文件，查看 MIME 提示、原始字节与 Base64 展开后的近似长度（用于粗算 JSON / data URL 体积）。

本地试算

• MIME（File.type）：—
• 原始大小：—
• Base64 字符数（约）：—