模态路由、时间对齐与压缩

New-Item -ItemType Directory -Force l9-multimodal
cd l9-multimodal

mkdir -p l9-multimodal
cd l9-multimodal

TOLERANCE_MS = 120
FRAMES = [(0, "封面"), (1000, "订单表"), (2000, "物流图")]
UTTERANCE = (1080, "查这张订单")

def align(frames, utterance, tolerance):
    if tolerance < 0:
        raise ValueError("时间容差不能为负数")
    if not frames:
        return None
    # 时间戳单位统一为毫秒；用绝对距离选择最近画面。
    stamp, text = utterance
    nearest = min(frames, key=lambda frame: abs(frame[0] - stamp))
    distance = abs(nearest[0] - stamp)
    if distance > tolerance:
        return None
    # 压缩保留任务相关的一帧、原时间戳和语句，便于追溯。
    return {"route": "图文", "frame_ms": nearest[0],
            "frame": nearest[1], "text": text, "gap_ms": distance}


def route(task, *, image=None, audio=None, text=None):
    # 路由是应用规则：材料送往哪个处理器，并把缺少的模态及时返回。
    choices = {
        "转写录音": ("ASR", {"audio": audio}),
        "读图金额": ("视觉", {"image": image}),
        "解释这张图": ("图文", {"image": image, "text": text}),
    }
    if task not in choices:
        raise ValueError("未知任务")
    processor, payload = choices[task]
    if any(value is None or value == "" for value in payload.values()):
        raise ValueError("任务所需模态缺失")
    return {"processor": processor, "payload": payload}

if __name__ == "__main__":
    result = align(FRAMES, UTTERANCE, TOLERANCE_MS)
    if result is None:
        print("路由：等待更接近的画面")
    else:
        bundle = route("解释这张图", image=result["frame"], text=result["text"])
        print("路由：" + bundle["processor"])
        print("保留画面：" + result["frame"] + "，时间=" + str(result["frame_ms"]) + "ms")
        print("对齐误差：" + str(result["gap_ms"]) + "ms")
        print("压缩：3帧 → 1帧")
    audio_route = route("转写录音", audio="录音片段A")
    image_route = route("读图金额", image="订单截图A")
    print("录音任务 → " + audio_route["processor"] + "，载荷键=audio")
    print("读图任务 → " + image_route["processor"] + "，载荷键=image")

import unittest
import align_lab as lab


class MechanismTests(unittest.TestCase):
    # 正常、边界和失败输入都由这里提供，直接运行即可。

    def test_normal(self):
        self.assertEqual(lab.align(lab.FRAMES, lab.UTTERANCE, 120)["frame"], "订单表")
        self.assertEqual(lab.route("转写录音", audio="片段")["processor"], "ASR")
        self.assertEqual(lab.route("读图金额", image="图")["payload"], {"image": "图"})
        self.assertEqual(lab.route("解释这张图", image="图", text="解释")["processor"], "图文")

    def test_boundary_tolerance(self):
        self.assertEqual(lab.align(lab.FRAMES, lab.UTTERANCE, 80)["gap_ms"], 80)
        self.assertIsNone(lab.align([], lab.UTTERANCE, 80))

    def test_failure(self):
        self.assertIsNone(lab.align(lab.FRAMES, lab.UTTERANCE, 79))
        with self.assertRaises(ValueError):
            lab.align(lab.FRAMES, lab.UTTERANCE, -1)
        with self.assertRaises(ValueError):
            lab.route("解释这张图", image="图")
        with self.assertRaises(ValueError):
            lab.route("未知任务", text="请求")


if __name__ == "__main__":
    unittest.main()

# 在刚才创建的练习目录检查文件，再运行程序。
ls
python -X utf8 align_lab.py
python -X utf8 -m unittest -v test_align_lab.py

路由：图文
保留画面：订单表，时间=1000ms
对齐误差：80ms
压缩：3帧 → 1帧
录音任务 → ASR，载荷键=audio
读图任务 → 视觉，载荷键=image

python -X utf8 align_lab.py