图像处理与多模态消息组装#

1
def transcribe_image(image_path: str) -> dict | None:

1
def transcribe_image(image_path: str) -> dict | None:
2
    try:
3
        with Image.open(image_path) as img:
4
            # RGBA（含透明通道）和 P（调色板）无法存为 JPEG
5
            # 统一转 RGB 保证格式兼容
6
            if img.mode in ('RGBA', 'P'):
7
                img = img.convert('RGB')
8

9
            img_format = img.format or 'JPEG'  # 读取不到时默认 JPEG
10

11
            # 用内存流代替临时文件，效率更高
12
            buffered = io.BytesIO()
13
            img.save(buffered, format=img_format)
14
            image_data = base64.b64encode(buffered.getvalue()).decode('utf-8')
15

16
            return {
17
                "type": "image_url",
18
                "image_url": {
19
                    "url": f"data:image/{img_format.lower()};base64,{image_data}",
20
                    # detail 控制 token 消耗：
21
                    # "low"  → 固定 85 token，速度快但粗糙
22
                    # "high" → 按图片尺寸计算，精准但贵
23
                    "detail": "high",
24
                },
25
            }
26
    except Exception as e:
27
        print(f"[图像处理失败] {e}")
28
        return None

1
def get_last_user_messages(history: list) -> list:

1
history 示意：
2
  [user₁, assistant₁, user₂, assistant₂, user₃, user₄]
3
                                            ↑         ↑
4
                               这两条是本轮新增，需要提取

1
def get_last_user_messages(history: list) -> list:
2
    if not history:
3
        return []
4
    if history[-1]["role"] == "assistant":
5
        return []  # 最后是 AI 回复，说明本轮已处理完
6

7
    # next() + 生成器：从后往前找第一个 assistant，比 for 循环更简洁
8
    last_assistant_idx = next(
9
        (i for i in range(len(history) - 1, -1, -1)
10
         if history[i]["role"] == "assistant"),
11
        -1  # 找不到时返回 -1（第一轮对话，全是 user 消息）
12
    )
13
    return history[last_assistant_idx + 1:]

1
content = [
2
    {"type": "text",      "text": "描述一下图片"},
3
    {"type": "image_url", "image_url": {"url": "data:image/...", "detail": "high"}},
4
    {"type": "text",      "text": "🎤 [语音输入]: 这是语音转换的文字"},
5
]

1
for x in user_messages:
2
    msg_content = x['content']
3

4
    # 情况 A：纯字符串
5
    # 来源：文字输入 或 音频转换后的文字
6
    if isinstance(msg_content, str):
7
        content.append({'type': 'text', 'text': msg_content})
8

9
    # 情况 B：列表格式（新版 Gradio type='messages' 的文本消息）
10
    # 结构：[{'type': 'text', 'text': '...'}, ...]
11
    elif isinstance(msg_content, list):
12
        for item in msg_content:
13
            if item.get('type') == 'text':
14
                content.append({'type': 'text', 'text': item['text']})
15

16
    # 情况 C：字典格式（图片文件路径）
17
    # 结构：{'path': '绝对路径', ...}
18
    elif isinstance(msg_content, dict):
19
        fp = msg_content.get('path') or msg_content.get('url')
20
        if fp:
21
            _append_file(content, fp)  # 读取并编码为 base64