多模态大模型
从对齐到协作

参考结论：去掉任一模态后是否真的影响任务；输出包含来源/区域/时间；不确定和权限不足都有明确动作。

完整实现路径：先定义媒体 ID、时间/空间范围、状态和证据 schema，再定义模型输入。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 1：先复制运行，再替换为自己的数据
from dataclasses import dataclass

@dataclass(frozen=True)
class MediaRequest:
    request_id: str
    media_id: str | None
    question: str
    permission: bool

def validate(r):
    if not r.permission: return "permission_denied"
    if not r.media_id: return "missing_media"
    if not r.question.strip(): return "invalid_question"
    return "ready"

assert validate(MediaRequest("r1", None, "图里是什么？", True)) == "missing_media"

验收标准：合法、缺失、冲突和撤权四类请求都有确定状态。

若结果不对：先查媒体是否存在和可访问，再查对齐，最后才查模型输出。

参考结论：时间和坐标能映射回原始媒体；缺失不是空字符串；回答会说明证据范围缩小。

完整实现路径：先定义媒体 ID、时间/空间范围、状态和证据 schema，再定义模型输入。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 2：先复制运行，再替换为自己的数据
from dataclasses import dataclass

@dataclass(frozen=True)
class MediaRequest:
    request_id: str
    media_id: str | None
    question: str
    permission: bool

def validate(r):
    if not r.permission: return "permission_denied"
    if not r.media_id: return "missing_media"
    if not r.question.strip(): return "invalid_question"
    return "ready"

assert validate(MediaRequest("r1", None, "图里是什么？", True)) == "missing_media"

验收标准：合法、缺失、冲突和撤权四类请求都有确定状态。

若结果不对：先查媒体是否存在和可访问，再查对齐，最后才查模型输出。

参考结论：合格标准：校验器能拒绝非法状态、陌生媒体、缺失证据或无动作的拒答；历史结果能根据版本重放。

完整实现路径：先定义媒体 ID、时间/空间范围、状态和证据 schema，再定义模型输入。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 3：先复制运行，再替换为自己的数据
from dataclasses import dataclass

@dataclass(frozen=True)
class MediaRequest:
    request_id: str
    media_id: str | None
    question: str
    permission: bool

def validate(r):
    if not r.permission: return "permission_denied"
    if not r.media_id: return "missing_media"
    if not r.question.strip(): return "invalid_question"
    return "ready"

assert validate(MediaRequest("r1", None, "图里是什么？", True)) == "missing_media"

验收标准：合法、缺失、冲突和撤权四类请求都有确定状态。

若结果不对：先查媒体是否存在和可访问，再查对齐，最后才查模型输出。

参考结论：Token 序列能回到原字符范围；未知词和截断可见；OCR/用户文本来源不同。

完整实现路径：分别记录文本 token、图像 patch、音频帧的生成规则，不用一个‘长度’掩盖三种尺度。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 1：先复制运行，再替换为自己的数据
from math import ceil

def patch_count(width, height, patch, add_cls=True):
    tokens = ceil(width / patch) * ceil(height / patch)
    return tokens + int(add_cls)

def frame_count(duration_ms, window_ms=25, hop_ms=10):
    if duration_ms <= window_ms: return 1
    return 1 + ceil((duration_ms - window_ms) / hop_ms)

assert patch_count(128, 96, 16) == 49
assert frame_count(83) == 7

验收标准：边界输入的计数可以手算复核，预处理参数写入 manifest。

若结果不对：先核对原始媒体属性，再逐步执行规范化、切片和编码。

参考结论：记录原图与输入图尺寸；无效 Patch 有 mask；答案区域能映射回原图。

完整实现路径：分别记录文本 token、图像 patch、音频帧的生成规则，不用一个‘长度’掩盖三种尺度。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 2：先复制运行，再替换为自己的数据
from math import ceil

def patch_count(width, height, patch, add_cls=True):
    tokens = ceil(width / patch) * ceil(height / patch)
    return tokens + int(add_cls)

def frame_count(duration_ms, window_ms=25, hop_ms=10):
    if duration_ms <= window_ms: return 1
    return 1 + ceil((duration_ms - window_ms) / hop_ms)

assert patch_count(128, 96, 16) == 49
assert frame_count(83) == 7

验收标准：边界输入的计数可以手算复核，预处理参数写入 manifest。

若结果不对：先核对原始媒体属性，再逐步执行规范化、切片和编码。

参考结论：时间从原始音频开始计算；补零不等于真实声音；输出引用带容差和状态。

完整实现路径：分别记录文本 token、图像 patch、音频帧的生成规则，不用一个‘长度’掩盖三种尺度。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 3：先复制运行，再替换为自己的数据
from math import ceil

def patch_count(width, height, patch, add_cls=True):
    tokens = ceil(width / patch) * ceil(height / patch)
    return tokens + int(add_cls)

def frame_count(duration_ms, window_ms=25, hop_ms=10):
    if duration_ms <= window_ms: return 1
    return 1 + ceil((duration_ms - window_ms) / hop_ms)

assert patch_count(128, 96, 16) == 49
assert frame_count(83) == 7

验收标准：边界输入的计数可以手算复核，预处理参数写入 manifest。

若结果不对：先核对原始媒体属性，再逐步执行规范化、切片和编码。

参考结论：合格标准：每种表示的粒度、位置/时间和版本清楚；缺失或遮蔽不会被写成正常内容。

完整实现路径：分别记录文本 token、图像 patch、音频帧的生成规则，不用一个‘长度’掩盖三种尺度。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 4：先复制运行，再替换为自己的数据
from math import ceil

def patch_count(width, height, patch, add_cls=True):
    tokens = ceil(width / patch) * ceil(height / patch)
    return tokens + int(add_cls)

def frame_count(duration_ms, window_ms=25, hop_ms=10):
    if duration_ms <= window_ms: return 1
    return 1 + ceil((duration_ms - window_ms) / hop_ms)

assert patch_count(128, 96, 16) == 49
assert frame_count(83) == 7

验收标准：边界输入的计数可以手算复核，预处理参数写入 manifest。

若结果不对：先核对原始媒体属性，再逐步执行规范化、切片和编码。

参考结论：正例关系有依据；分别讨论图到文和文到图；相似度不被写成概率或事实。

完整实现路径：对嵌入做归一化，显式计算相似度矩阵、温度缩放和双向检索；困难负例必须人工核验。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 1：先复制运行，再替换为自己的数据
import torch
import torch.nn.functional as F

image = F.normalize(torch.tensor([[1., 0.], [0., 1.], [1., 1.]]), dim=1)
text = F.normalize(torch.tensor([[.9, .1], [.2, .8], [.7, .7]]), dim=1)
temperature = 0.07
logits = image @ text.T / temperature
image_to_text = logits.argmax(dim=1)
text_to_image = logits.argmax(dim=0)
assert image_to_text.tolist() == [0, 1, 2]
assert text_to_image.tolist() == [0, 1, 2]

验收标准：正配对排名、双向召回和反事实样例都有记录。

若结果不对：先查配对 ID，再查归一化与温度，最后分析负例质量。

参考结论：难负例不是随便找的相似文本；假负例有明确依据；拆分不会让同一实体跨集合。

完整实现路径：对嵌入做归一化，显式计算相似度矩阵、温度缩放和双向检索；困难负例必须人工核验。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 2：先复制运行，再替换为自己的数据
import torch
import torch.nn.functional as F

image = F.normalize(torch.tensor([[1., 0.], [0., 1.], [1., 1.]]), dim=1)
text = F.normalize(torch.tensor([[.9, .1], [.2, .8], [.7, .7]]), dim=1)
temperature = 0.07
logits = image @ text.T / temperature
image_to_text = logits.argmax(dim=1)
text_to_image = logits.argmax(dim=0)
assert image_to_text.tolist() == [0, 1, 2]
assert text_to_image.tolist() == [0, 1, 2]

验收标准：正配对排名、双向召回和反事实样例都有记录。

若结果不对：先查配对 ID，再查归一化与温度，最后分析负例质量。

参考结论：合格标准：结论有样例和指标；能区分假负例、捷径、域偏移和索引问题；修复不会绕过权限。

完整实现路径：对嵌入做归一化，显式计算相似度矩阵、温度缩放和双向检索；困难负例必须人工核验。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 3：先复制运行，再替换为自己的数据
import torch
import torch.nn.functional as F

image = F.normalize(torch.tensor([[1., 0.], [0., 1.], [1., 1.]]), dim=1)
text = F.normalize(torch.tensor([[.9, .1], [.2, .8], [.7, .7]]), dim=1)
temperature = 0.07
logits = image @ text.T / temperature
image_to_text = logits.argmax(dim=1)
text_to_image = logits.argmax(dim=0)
assert image_to_text.tolist() == [0, 1, 2]
assert text_to_image.tolist() == [0, 1, 2]

验收标准：正配对排名、双向召回和反事实样例都有记录。

若结果不对：先查配对 ID，再查归一化与温度，最后分析负例质量。

参考结论：合格标准：每个结论有样例/指标；修复能回滚；不把相似度写成概率或事实。

完整实现路径：对嵌入做归一化，显式计算相似度矩阵、温度缩放和双向检索；困难负例必须人工核验。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 4：先复制运行，再替换为自己的数据
import torch
import torch.nn.functional as F

image = F.normalize(torch.tensor([[1., 0.], [0., 1.], [1., 1.]]), dim=1)
text = F.normalize(torch.tensor([[.9, .1], [.2, .8], [.7, .7]]), dim=1)
temperature = 0.07
logits = image @ text.T / temperature
image_to_text = logits.argmax(dim=1)
text_to_image = logits.argmax(dim=0)
assert image_to_text.tolist() == [0, 1, 2]
assert text_to_image.tolist() == [0, 1, 2]

验收标准：正配对排名、双向召回和反事实样例都有记录。

若结果不对：先查配对 ID，再查归一化与温度，最后分析负例质量。

参考结论：架构与证据关系对应；比较时固定数据和编码器；降级状态可被用户看见。

完整实现路径：把视觉编码器、投影器、查询 token、语言模型的形状写成接口合同，再比较早期、拼接和交叉注意力。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 1：先复制运行，再替换为自己的数据
def fusion_contract(visual, query, hidden=1024):
    if visual.ndim != 3: raise ValueError("visual must be (B,N,D)")
    if query.ndim != 3: raise ValueError("query must be (B,Q,H)")
    if visual.shape[0] != query.shape[0]: raise ValueError("batch mismatch")
    if query.shape[-1] != hidden: raise ValueError("hidden mismatch")
    return {"batch": visual.shape[0], "visual_tokens": visual.shape[1],
            "query_tokens": query.shape[1], "hidden": hidden}

visual = torch.randn(2, 196, 768)
query = torch.randn(2, 32, 1024)
assert fusion_contract(visual, query)["query_tokens"] == 32

验收标准：缺失模态、伪模态和真实模态三种输入行为可区分。

若结果不对：逐层打印 B、token 数和 hidden size，优先定位第一处不一致。

参考结论：每个维度含义明确；空间来源可展开；失败不会静默变成全局向量。

完整实现路径：把视觉编码器、投影器、查询 token、语言模型的形状写成接口合同，再比较早期、拼接和交叉注意力。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 2：先复制运行，再替换为自己的数据
def fusion_contract(visual, query, hidden=1024):
    if visual.ndim != 3: raise ValueError("visual must be (B,N,D)")
    if query.ndim != 3: raise ValueError("query must be (B,Q,H)")
    if visual.shape[0] != query.shape[0]: raise ValueError("batch mismatch")
    if query.shape[-1] != hidden: raise ValueError("hidden mismatch")
    return {"batch": visual.shape[0], "visual_tokens": visual.shape[1],
            "query_tokens": query.shape[1], "hidden": hidden}

visual = torch.randn(2, 196, 768)
query = torch.randn(2, 32, 1024)
assert fusion_contract(visual, query)["query_tokens"] == 32

验收标准：缺失模态、伪模态和真实模态三种输入行为可区分。

若结果不对：逐层打印 B、token 数和 hidden size，优先定位第一处不一致。

参考结论：合格标准：质量损失和资源收益同时记录；缓存键含版本与权限；超预算有用户可理解的状态。

完整实现路径：把视觉编码器、投影器、查询 token、语言模型的形状写成接口合同，再比较早期、拼接和交叉注意力。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 3：先复制运行，再替换为自己的数据
def fusion_contract(visual, query, hidden=1024):
    if visual.ndim != 3: raise ValueError("visual must be (B,N,D)")
    if query.ndim != 3: raise ValueError("query must be (B,Q,H)")
    if visual.shape[0] != query.shape[0]: raise ValueError("batch mismatch")
    if query.shape[-1] != hidden: raise ValueError("hidden mismatch")
    return {"batch": visual.shape[0], "visual_tokens": visual.shape[1],
            "query_tokens": query.shape[1], "hidden": hidden}

visual = torch.randn(2, 196, 768)
query = torch.randn(2, 32, 1024)
assert fusion_contract(visual, query)["query_tokens"] == 32

验收标准：缺失模态、伪模态和真实模态三种输入行为可区分。

若结果不对：逐层打印 B、token 数和 hidden size，优先定位第一处不一致。

参考结论：控制其他变量；引用与内容同步变化；失败时进入不确定或阻断。

完整实现路径：把视觉编码器、投影器、查询 token、语言模型的形状写成接口合同，再比较早期、拼接和交叉注意力。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 4：先复制运行，再替换为自己的数据
def fusion_contract(visual, query, hidden=1024):
    if visual.ndim != 3: raise ValueError("visual must be (B,N,D)")
    if query.ndim != 3: raise ValueError("query must be (B,Q,H)")
    if visual.shape[0] != query.shape[0]: raise ValueError("batch mismatch")
    if query.shape[-1] != hidden: raise ValueError("hidden mismatch")
    return {"batch": visual.shape[0], "visual_tokens": visual.shape[1],
            "query_tokens": query.shape[1], "hidden": hidden}

visual = torch.randn(2, 196, 768)
query = torch.randn(2, 32, 1024)
assert fusion_contract(visual, query)["query_tokens"] == 32

验收标准：缺失模态、伪模态和真实模态三种输入行为可区分。

若结果不对：逐层打印 B、token 数和 hidden size，优先定位第一处不一致。

参考结论：目标与可观测指标对应；写出假负例/伪标签风险；不把 loss 下降当作任务完成。

完整实现路径：数据卡先于训练；主目标和辅助目标分别写出输入、损失、捷径风险和停止条件。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 1：先复制运行，再替换为自己的数据
from collections import Counter

def audit_pairs(rows):
    ids = [r["id"] for r in rows]
    sources = Counter(r["source"] for r in rows)
    return {
        "rows": len(rows),
        "duplicate_ids": len(ids) - len(set(ids)),
        "missing_license": sum(not r.get("license") for r in rows),
        "source_distribution": dict(sources),
    }

report = audit_pairs(rows)
if report["duplicate_ids"] or report["missing_license"]:
    raise ValueError(report)

验收标准：去重、许可、隐私、切分与删除链路都能审计。

若结果不对：分数异常升高时先排查近重复、模板和标签泄漏。

参考结论：原始和派生数据分开；过滤规则有版本；unknown 不被悄悄当作 accepted 或 rejected。

完整实现路径：数据卡先于训练；主目标和辅助目标分别写出输入、损失、捷径风险和停止条件。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 2：先复制运行，再替换为自己的数据
from collections import Counter

def audit_pairs(rows):
    ids = [r["id"] for r in rows]
    sources = Counter(r["source"] for r in rows)
    return {
        "rows": len(rows),
        "duplicate_ids": len(ids) - len(set(ids)),
        "missing_license": sum(not r.get("license") for r in rows),
        "source_distribution": dict(sources),
    }

report = audit_pairs(rows)
if report["duplicate_ids"] or report["missing_license"]:
    raise ValueError(report)

验收标准：去重、许可、隐私、切分与删除链路都能审计。

若结果不对：分数异常升高时先排查近重复、模板和标签泄漏。

参考结论：合格标准：阶段切换有证据；保存完整 manifest；能解释指标变化来自数据、目标还是参数策略。

完整实现路径：数据卡先于训练；主目标和辅助目标分别写出输入、损失、捷径风险和停止条件。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 3：先复制运行，再替换为自己的数据
from collections import Counter

def audit_pairs(rows):
    ids = [r["id"] for r in rows]
    sources = Counter(r["source"] for r in rows)
    return {
        "rows": len(rows),
        "duplicate_ids": len(ids) - len(set(ids)),
        "missing_license": sum(not r.get("license") for r in rows),
        "source_distribution": dict(sources),
    }

report = audit_pairs(rows)
if report["duplicate_ids"] or report["missing_license"]:
    raise ValueError(report)

验收标准：去重、许可、隐私、切分与删除链路都能审计。

若结果不对：分数异常升高时先排查近重复、模板和标签泄漏。

参考结论：合格标准：目标与下游指标对应；数据许可可追溯；退化不会被平均分掩盖。

完整实现路径：数据卡先于训练；主目标和辅助目标分别写出输入、损失、捷径风险和停止条件。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 4：先复制运行，再替换为自己的数据
from collections import Counter

def audit_pairs(rows):
    ids = [r["id"] for r in rows]
    sources = Counter(r["source"] for r in rows)
    return {
        "rows": len(rows),
        "duplicate_ids": len(ids) - len(set(ids)),
        "missing_license": sum(not r.get("license") for r in rows),
        "source_distribution": dict(sources),
    }

report = audit_pairs(rows)
if report["duplicate_ids"] or report["missing_license"]:
    raise ValueError(report)

验收标准：去重、许可、隐私、切分与删除链路都能审计。

若结果不对：分数异常升高时先排查近重复、模板和标签泄漏。

参考结论：问题类型对应不同证据；看不清时拒答；引用区域能回到原图。

完整实现路径：回答对象必须同时返回 answer、evidence、media span 和 uncertainty；暂定转写不能冒充确认文本。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 1：先复制运行，再替换为自己的数据
def grounded_answer(answer, evidence, uncertainty):
    required = {"media_id", "span", "quote"}
    if not evidence or any(required - item.keys() for item in evidence):
        return {"status": "insufficient_evidence", "answer": None}
    return {"status": "answered", "answer": answer,
            "evidence": evidence, "uncertainty": uncertainty}

result = grounded_answer("右上角有 3 个图标", [{
    "media_id": "img-07", "span": [0.62, 0.05, 0.96, 0.30], "quote": "3"
}], "medium")
assert result["status"] == "answered"

验收标准：每个结论能定位到页、区域或时间段，低证据请求会拒答。

若结果不对：先重放同一媒体和处理器版本，再查证据定位，最后查生成参数。

参考结论：时间区间可回放；修订保留差异；未确认文本不能触发外部动作。

完整实现路径：回答对象必须同时返回 answer、evidence、media span 和 uncertainty；暂定转写不能冒充确认文本。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 2：先复制运行，再替换为自己的数据
def grounded_answer(answer, evidence, uncertainty):
    required = {"media_id", "span", "quote"}
    if not evidence or any(required - item.keys() for item in evidence):
        return {"status": "insufficient_evidence", "answer": None}
    return {"status": "answered", "answer": answer,
            "evidence": evidence, "uncertainty": uncertainty}

result = grounded_answer("右上角有 3 个图标", [{
    "media_id": "img-07", "span": [0.62, 0.05, 0.96, 0.30], "quote": "3"
}], "medium")
assert result["status"] == "answered"

验收标准：每个结论能定位到页、区域或时间段，低证据请求会拒答。

若结果不对：先重放同一媒体和处理器版本，再查证据定位，最后查生成参数。

参考结论：合格标准：旧媒体不会被同名替换；观察/推断/执行分层；撤权后缓存与引用都受影响。

完整实现路径：回答对象必须同时返回 answer、evidence、media span 和 uncertainty；暂定转写不能冒充确认文本。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 3：先复制运行，再替换为自己的数据
def grounded_answer(answer, evidence, uncertainty):
    required = {"media_id", "span", "quote"}
    if not evidence or any(required - item.keys() for item in evidence):
        return {"status": "insufficient_evidence", "answer": None}
    return {"status": "answered", "answer": answer,
            "evidence": evidence, "uncertainty": uncertainty}

result = grounded_answer("右上角有 3 个图标", [{
    "media_id": "img-07", "span": [0.62, 0.05, 0.96, 0.30], "quote": "3"
}], "medium")
assert result["status"] == "answered"

验收标准：每个结论能定位到页、区域或时间段，低证据请求会拒答。

若结果不对：先重放同一媒体和处理器版本，再查证据定位，最后查生成参数。

参考结论：合格标准：媒体 ID 稳定；摘要不替代原证据；确认前不执行外部动作。

完整实现路径：回答对象必须同时返回 answer、evidence、media span 和 uncertainty；暂定转写不能冒充确认文本。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 4：先复制运行，再替换为自己的数据
def grounded_answer(answer, evidence, uncertainty):
    required = {"media_id", "span", "quote"}
    if not evidence or any(required - item.keys() for item in evidence):
        return {"status": "insufficient_evidence", "answer": None}
    return {"status": "answered", "answer": answer,
            "evidence": evidence, "uncertainty": uncertainty}

result = grounded_answer("右上角有 3 个图标", [{
    "media_id": "img-07", "span": [0.62, 0.05, 0.96, 0.30], "quote": "3"
}], "medium")
assert result["status"] == "answered"

验收标准：每个结论能定位到页、区域或时间段，低证据请求会拒答。

若结果不对：先重放同一媒体和处理器版本，再查证据定位，最后查生成参数。

参考结论：观察与解释分开；引用有坐标/版本；媒体文字不会改变系统指令。

完整实现路径：工具调用采用输入/输出 schema、幂等键、超时和状态机；模型不能把工具计划当成工具结果。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 1：先复制运行，再替换为自己的数据
ALLOWED = {
    "queued": {"running", "cancelled"},
    "running": {"succeeded", "failed", "timed_out", "cancelled"},
}

def transition(job, target):
    if target not in ALLOWED.get(job["state"], set()):
        raise ValueError(f"illegal transition: {job['state']} -> {target}")
    job["state"] = target
    return job

job = {"id": "ocr-17", "state": "queued", "idempotency_key": "abc"}
assert transition(job, "running")["state"] == "running"

验收标准：超时、重复提交、权限失败和用户撤权均能回退。

若结果不对：按 request_id 串起模型、工具和用户确认日志，定位第一次状态跳变。

参考结论：工具结果带来源与范围；参数先校验；失败不会被模型语言掩盖。

完整实现路径：工具调用采用输入/输出 schema、幂等键、超时和状态机；模型不能把工具计划当成工具结果。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 2：先复制运行，再替换为自己的数据
ALLOWED = {
    "queued": {"running", "cancelled"},
    "running": {"succeeded", "failed", "timed_out", "cancelled"},
}

def transition(job, target):
    if target not in ALLOWED.get(job["state"], set()):
        raise ValueError(f"illegal transition: {job['state']} -> {target}")
    job["state"] = target
    return job

job = {"id": "ocr-17", "state": "queued", "idempotency_key": "abc"}
assert transition(job, "running")["state"] == "running"

验收标准：超时、重复提交、权限失败和用户撤权均能回退。

若结果不对：按 request_id 串起模型、工具和用户确认日志，定位第一次状态跳变。

参考结论：合格标准：模型不能直接执行；每个状态有停止动作；外部写入有幂等和回执。

完整实现路径：工具调用采用输入/输出 schema、幂等键、超时和状态机；模型不能把工具计划当成工具结果。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 3：先复制运行，再替换为自己的数据
ALLOWED = {
    "queued": {"running", "cancelled"},
    "running": {"succeeded", "failed", "timed_out", "cancelled"},
}

def transition(job, target):
    if target not in ALLOWED.get(job["state"], set()):
        raise ValueError(f"illegal transition: {job['state']} -> {target}")
    job["state"] = target
    return job

job = {"id": "ocr-17", "state": "queued", "idempotency_key": "abc"}
assert transition(job, "running")["state"] == "running"

验收标准：超时、重复提交、权限失败和用户撤权均能回退。

若结果不对：按 request_id 串起模型、工具和用户确认日志，定位第一次状态跳变。

参考结论：合格标准：冲突不被平均；参数先校验；副作用有确认、幂等和回执。

完整实现路径：工具调用采用输入/输出 schema、幂等键、超时和状态机；模型不能把工具计划当成工具结果。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 4：先复制运行，再替换为自己的数据
ALLOWED = {
    "queued": {"running", "cancelled"},
    "running": {"succeeded", "failed", "timed_out", "cancelled"},
}

def transition(job, target):
    if target not in ALLOWED.get(job["state"], set()):
        raise ValueError(f"illegal transition: {job['state']} -> {target}")
    job["state"] = target
    return job

job = {"id": "ocr-17", "state": "queued", "idempotency_key": "abc"}
assert transition(job, "running")["state"] == "running"

验收标准：超时、重复提交、权限失败和用户撤权均能回退。

若结果不对：按 request_id 串起模型、工具和用户确认日志，定位第一次状态跳变。

参考结论：指标来自任务合同；切片有样本量；能区分答案正确与证据正确。

完整实现路径：按任务质量、证据正确、切片鲁棒、安全和系统成本五层评估，并设计缺失/冲突模态回放。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 1：先复制运行，再替换为自己的数据
def slice_report(rows):
    groups = {}
    for row in rows:
        key = (row["modality_state"], row["language"])
        groups.setdefault(key, []).append(row)
    return {str(k): {
        "n": len(v),
        "answer_accuracy": sum(x["answer_ok"] for x in v) / len(v),
        "evidence_accuracy": sum(x["evidence_ok"] for x in v) / len(v),
    } for k, v in groups.items()}

report = slice_report(rows)
assert all(v["evidence_accuracy"] >= .90 for v in report.values())

验收标准：总体分数与关键切片同时过门，P0 事故可按 runbook 回滚。

若结果不对：先区分输入问题、工具问题、融合问题和生成问题，再归因。

参考结论：缺失原因不同状态不同；冲突不会被平均；隐私/权限失败直接阻断。

完整实现路径：按任务质量、证据正确、切片鲁棒、安全和系统成本五层评估，并设计缺失/冲突模态回放。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 2：先复制运行，再替换为自己的数据
def slice_report(rows):
    groups = {}
    for row in rows:
        key = (row["modality_state"], row["language"])
        groups.setdefault(key, []).append(row)
    return {str(k): {
        "n": len(v),
        "answer_accuracy": sum(x["answer_ok"] for x in v) / len(v),
        "evidence_accuracy": sum(x["evidence_ok"] for x in v) / len(v),
    } for k, v in groups.items()}

report = slice_report(rows)
assert all(v["evidence_accuracy"] >= .90 for v in report.values())

验收标准：总体分数与关键切片同时过门，P0 事故可按 runbook 回滚。

若结果不对：先区分输入问题、工具问题、融合问题和生成问题，再归因。

参考结论：合格标准：恢复组件版本完整；硬失败立即阻断；回滚后重新验证证据、状态、权限和成本。

完整实现路径：按任务质量、证据正确、切片鲁棒、安全和系统成本五层评估，并设计缺失/冲突模态回放。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 3：先复制运行，再替换为自己的数据
def slice_report(rows):
    groups = {}
    for row in rows:
        key = (row["modality_state"], row["language"])
        groups.setdefault(key, []).append(row)
    return {str(k): {
        "n": len(v),
        "answer_accuracy": sum(x["answer_ok"] for x in v) / len(v),
        "evidence_accuracy": sum(x["evidence_ok"] for x in v) / len(v),
    } for k, v in groups.items()}

report = slice_report(rows)
assert all(v["evidence_accuracy"] >= .90 for v in report.values())

验收标准：总体分数与关键切片同时过门，P0 事故可按 runbook 回滚。

若结果不对：先区分输入问题、工具问题、融合问题和生成问题，再归因。

参考结论：合格标准：硬失败不能被平均分抵消；门槛有人负责；回滚后有重新验证。

完整实现路径：按任务质量、证据正确、切片鲁棒、安全和系统成本五层评估，并设计缺失/冲突模态回放。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 4：先复制运行，再替换为自己的数据
def slice_report(rows):
    groups = {}
    for row in rows:
        key = (row["modality_state"], row["language"])
        groups.setdefault(key, []).append(row)
    return {str(k): {
        "n": len(v),
        "answer_accuracy": sum(x["answer_ok"] for x in v) / len(v),
        "evidence_accuracy": sum(x["evidence_ok"] for x in v) / len(v),
    } for k, v in groups.items()}

report = slice_report(rows)
assert all(v["evidence_accuracy"] >= .90 for v in report.values())

验收标准：总体分数与关键切片同时过门，P0 事故可按 runbook 回滚。

若结果不对：先区分输入问题、工具问题、融合问题和生成问题，再归因。

参考结论：任务低风险；状态有动作；证据可定位；模型不能直接改变外部系统。

完整实现路径：从一次失败回答反推数据、编码、对齐、融合、工具、评估和回滚接口，交付可复放项目。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 1：先复制运行，再替换为自己的数据
from pathlib import Path
import json

required = {"contract", "media_manifest", "model_revision",
            "gold_requests", "evaluation", "rollback"}
release = json.loads(Path("release.json").read_text(encoding="utf-8"))
missing = required - release.keys()
if missing: raise ValueError(f"release missing {sorted(missing)}")
for request in release["gold_requests"]:
    assert {"request", "expected_status", "evidence"} <= request.keys()
print("multimodal release is replayable")

验收标准：第二位同学只看 README 能重现一次成功、拒答和删除回放。

若结果不对：保存最小失败包：请求、媒体哈希、版本、日志和预期状态。

参考结论：合格标准：根因是可检验假设；修复带版本和负责人；没有用“模型幻觉”结束分析。

完整实现路径：从一次失败回答反推数据、编码、对齐、融合、工具、评估和回滚接口，交付可复放项目。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 2：先复制运行，再替换为自己的数据
from pathlib import Path
import json

required = {"contract", "media_manifest", "model_revision",
            "gold_requests", "evaluation", "rollback"}
release = json.loads(Path("release.json").read_text(encoding="utf-8"))
missing = required - release.keys()
if missing: raise ValueError(f"release missing {sorted(missing)}")
for request in release["gold_requests"]:
    assert {"request", "expected_status", "evidence"} <= request.keys()
print("multimodal release is replayable")

验收标准：第二位同学只看 README 能重现一次成功、拒答和删除回放。

若结果不对：保存最小失败包：请求、媒体哈希、版本、日志和预期状态。

参考结论：合格标准：可复现、可审查、可拒答、可删除、可回滚；能解释每个函数职责和每个结论的证据。

完整实现路径：从一次失败回答反推数据、编码、对齐、融合、工具、评估和回滚接口，交付可复放项目。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 3：先复制运行，再替换为自己的数据
from pathlib import Path
import json

required = {"contract", "media_manifest", "model_revision",
            "gold_requests", "evaluation", "rollback"}
release = json.loads(Path("release.json").read_text(encoding="utf-8"))
missing = required - release.keys()
if missing: raise ValueError(f"release missing {sorted(missing)}")
for request in release["gold_requests"]:
    assert {"request", "expected_status", "evidence"} <= request.keys()
print("multimodal release is replayable")

验收标准：第二位同学只看 README 能重现一次成功、拒答和删除回放。

若结果不对：保存最小失败包：请求、媒体哈希、版本、日志和预期状态。

参考结论：合格标准：干净目录可运行；证据、权限、状态和回滚均可追溯；没有口头隐藏步骤。

完整实现路径：从一次失败回答反推数据、编码、对齐、融合、工具、评估和回滚接口，交付可复放项目。 本题不要只交一段说明：应同时提交输入样例、可运行实现、输出或断言，以及失败时的明确状态。

# 对应本节练习 4：先复制运行，再替换为自己的数据
from pathlib import Path
import json

required = {"contract", "media_manifest", "model_revision",
            "gold_requests", "evaluation", "rollback"}
release = json.loads(Path("release.json").read_text(encoding="utf-8"))
missing = required - release.keys()
if missing: raise ValueError(f"release missing {sorted(missing)}")
for request in release["gold_requests"]:
    assert {"request", "expected_status", "evidence"} <= request.keys()
print("multimodal release is replayable")

验收标准：第二位同学只看 README 能重现一次成功、拒答和删除回放。

若结果不对：保存最小失败包：请求、媒体哈希、版本、日志和预期状态。