Files

T

iomgaa 2be2569ed8 docs(tree): add tree module vertical slice design and implementation plan

2026-07-07 01:27:36 -04:00

41 KiB

Raw Blame History

建树模块竖切实现计划

For agentic workers: REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 在 TRM5 中实现完整的建树模块竖切——数据结构、VideoTreeBuilder、字幕、校验、运行时环境、修复模式、适配器、迁移。

Architecture: 扩展 reference 的 TreeIndex 为统一数据结构（Card 体系 + embedding），VideoTreeBuilder 通过 VLMProvider/LLMProvider Protocol 调用已治理的 LLM 客户端。Clean Architecture 四层分层：core/protocols.py → app/tree/ → adapters/。

Tech Stack: Python 3.11, asyncio, loguru, numpy, sentence-transformers, httpx, ffmpeg, pytest

设计文档: research-wiki/designs/2026-07-07-tree-module-design.md

核心算法保真: 本计划涉及算法 #1（L2 轴心建树）、#2（VLM 批量帧描述 + JSON fallback）、#3（断点续跑）、#12（树环境语义搜索，分块→单节点 embedding 变更）。每个涉及保真的 Task 标注了 [保真] 标记和校验检查点。

文件结构总览

操作	文件路径	职责
Create	`app/tree/index.py`	TreeIndex + L1/L2/L3 Node/Card + 序列化
Create	`app/tree/subtitle.py`	SRT 解析 + 完整性检查 + Voronoi 分配
Create	`app/tree/verify.py`	质量校验
Create	`app/tree/video_builder.py`	VideoTreeBuilder（asyncio, VLM）
Create	`app/tree/environment.py`	TreeEnvironment 运行时
Create	`app/tree/repair/__init__.py`	修复模式包
Create	`app/tree/repair/detector.py`	缺失/低质量节点检测
Create	`app/tree/repair/regenerator.py`	VLM 重生成 + 向上级联
Create	`app/tree/repair/supplement.py`	Q&A 反向补全
Modify	`app/ports.py`	新增 EmbeddingProvider Protocol
Create	`adapters/embedding.py`	EmbeddingProvider 实现
Create	`adapters/vlm.py`	VLMProvider 最小可用实现
Create	`tools/migrate_from_trm4.sh`	迁移主脚本
Create	`tools/convert_flat_to_treeindex.py`	格式转换（迁移后归档）
Create	`tests/unit/test_tree_index.py`	TreeIndex 单元测试
Create	`tests/unit/test_subtitle.py`	字幕模块单元测试
Create	`tests/unit/test_verify.py`	校验模块单元测试
Create	`tests/unit/test_video_builder.py`	VideoTreeBuilder 单元测试
Create	`tests/unit/test_tree_environment.py`	TreeEnvironment 单元测试
Create	`tests/unit/test_embedding_adapter.py`	Embedding 适配器测试
Create	`tests/unit/test_vlm_adapter.py`	VLM 适配器测试
Create	`tests/unit/test_repair_detector.py`	修复检测器测试
Create	`tests/unit/test_repair_regenerator.py`	修复重生成器测试
Create	`tests/unit/test_repair_supplement.py`	Q&A 补全测试
Create	`tests/integration/test_tree_build_e2e.py`	建树端到端集成测试

Task 1: TreeIndex 数据结构

Files:

Create: app/tree/index.py
Test: tests/unit/test_tree_index.py

说明: 三级 Card frozen dataclass + 三级 Node dataclass + TreeIndex 容器 + JSON 序列化/反序列化 + embedding 矩阵提取。这是整个竖切的基础，后续所有 Task 依赖此文件。

Step 1: 编写 Card + Node + TreeIndex 的失败测试

# tests/unit/test_tree_index.py
"""TreeIndex 数据结构单元测试。"""

from __future__ import annotations

import json
import tempfile
from pathlib import Path

import numpy as np
import pytest

from app.tree.index import (
    IndexMeta,
    L1Card,
    L1Node,
    L2Card,
    L2Node,
    L3Card,
    L3Node,
    TreeIndex,
)


# ── fixtures ──

def _make_l3(idx: int = 0) -> L3Node:
    return L3Node(
        id=f"l1_0_l2_0_l3_{idx}",
        card=L3Card(
            frame_summary=f"帧{idx}描述",
            visible_entities=["实体A"],
            ongoing_actions=["动作A"],
            visible_text=["文字A"],
            spatial_layout="居中构图",
            visual_attributes={"lighting": "明亮"},
        ),
        timestamp=idx * 2.0,
        frame_path=f"frames/l1_0_l2_0_l3_{idx}.jpg",
    )


def _make_l2(n_l3: int = 2) -> L2Node:
    return L2Node(
        id="l1_0_l2_0",
        card=L2Card(
            event_description="事件描述",
            entities=["实体B"],
            actions=["动作B"],
            action_subjects=["主体B"],
            visible_text=["文字B"],
            spatial_relations="左右排列",
            state_changes=None,
        ),
        time_range=(0.0, 60.0),
        children=[_make_l3(i) for i in range(n_l3)],
    )


def _make_l1(n_l2: int = 1, n_l3: int = 2) -> L1Node:
    return L1Node(
        id="l1_0",
        card=L1Card(
            scene_summary="场景摘要",
            main_setting="室内",
            key_entities=["实体C"],
            main_actions=["动作C"],
            topic_keywords=["关键词"],
            visible_text=["文字C"],
            temporal_flow="从左到右",
        ),
        time_range=(0.0, 600.0),
        children=[_make_l2(n_l3) for _ in range(n_l2)],
    )


def _make_index(n_l1: int = 1) -> TreeIndex:
    meta = IndexMeta(source_path="/test/video.mp4", modality="video")
    return TreeIndex(metadata=meta, roots=[_make_l1() for _ in range(n_l1)])


# ── Card 测试 ──

class TestCards:
    def test_l3_card_frozen(self):
        card = L3Card(
            frame_summary="desc", visible_entities=[], ongoing_actions=[],
            visible_text=[], spatial_layout="", visual_attributes={},
        )
        with pytest.raises(AttributeError):
            card.frame_summary = "changed"

    def test_l2_card_fields(self):
        card = L2Card(
            event_description="evt", entities=[], actions=[],
            action_subjects=[], visible_text=[], spatial_relations="",
            state_changes=None,
        )
        assert card.event_description == "evt"
        assert card.state_changes is None

    def test_l1_card_fields(self):
        card = L1Card(
            scene_summary="scene", main_setting="outdoor",
            key_entities=["e"], main_actions=["a"],
            topic_keywords=["k"], visible_text=["t"],
            temporal_flow="flow",
        )
        assert card.scene_summary == "scene"


# ── Node 测试 ──

class TestNodes:
    def test_l3_description_property(self):
        node = _make_l3()
        assert node.description == node.card.frame_summary

    def test_l2_description_property(self):
        node = _make_l2()
        assert node.description == node.card.event_description

    def test_l1_summary_property(self):
        node = _make_l1()
        assert node.summary == node.card.scene_summary

    def test_l3_default_embedding_none(self):
        node = _make_l3()
        assert node.embedding is None

    def test_l3_subtitle_default_none(self):
        node = _make_l3()
        assert node.subtitle is None


# ── TreeIndex 测试 ──

class TestTreeIndex:
    def test_is_embedded_false_by_default(self):
        index = _make_index()
        assert not index.is_embedded

    def test_embed_all(self):
        index = _make_index()
        def fake_embed(texts):
            if isinstance(texts, str):
                texts = [texts]
            return np.random.randn(len(texts), 4).astype(np.float32)

        index.embed_all(fake_embed, "test-model", 4)
        assert index.is_embedded
        assert index.metadata.embed_model == "test-model"
        assert index.metadata.embed_dim == 4

    def test_l1_embeddings_shape(self):
        index = _make_index(n_l1=2)
        def fake_embed(texts):
            if isinstance(texts, str):
                texts = [texts]
            return np.random.randn(len(texts), 4).astype(np.float32)

        index.embed_all(fake_embed, "test-model", 4)
        m = index.l1_embeddings()
        assert m.shape == (2, 4)

    def test_get_node(self):
        index = _make_index()
        node = index.get_node(0, 0, 1)
        assert node.id == "l1_0_l2_0_l3_1"

    def test_get_node_out_of_bounds(self):
        index = _make_index()
        with pytest.raises(IndexError):
            index.get_node(99, 0, 0)


# ── 序列化测试 ──

class TestSerialization:
    def test_json_roundtrip(self, tmp_path):
        index = _make_index()
        path = tmp_path / "tree.json"
        index.save_json(str(path))

        loaded = TreeIndex.load_json(str(path))
        assert len(loaded.roots) == 1
        assert loaded.roots[0].id == "l1_0"
        assert loaded.roots[0].card.scene_summary == "场景摘要"
        assert loaded.roots[0].children[0].children[0].card.frame_summary == "帧0描述"

    def test_json_roundtrip_with_embedding(self, tmp_path):
        index = _make_index()
        def fake_embed(texts):
            if isinstance(texts, str):
                texts = [texts]
            return np.random.randn(len(texts), 4).astype(np.float32)

        index.embed_all(fake_embed, "test-model", 4)
        path = tmp_path / "tree_emb.json"
        index.save_json(str(path), include_embedding=True)

        loaded = TreeIndex.load_json(str(path))
        assert loaded.is_embedded
        np.testing.assert_array_almost_equal(
            loaded.roots[0].embedding, index.roots[0].embedding, decimal=5
        )

    def test_l1_json_roundtrip(self, tmp_path):
        from app.tree.index import save_l1_json, load_l1_json
        l1 = _make_l1()
        path = tmp_path / "l1_0.json"
        save_l1_json(str(path), l1)
        loaded = load_l1_json(str(path))
        assert loaded.id == "l1_0"
        assert len(loaded.children) == 1
        assert len(loaded.children[0].children) == 2

    def test_id_uniqueness_validation(self, tmp_path):
        """重复 ID 在反序列化时应报错。"""
        index = _make_index()
        d = index.to_dict()
        # 人为制造重复 ID
        d["roots"].append(d["roots"][0])
        path = tmp_path / "dup.json"
        with open(path, "w") as f:
            json.dump(d, f)
        with pytest.raises(ValueError, match="重复"):
            TreeIndex.load_json(str(path))

Step 2: 运行测试，确认全部 FAIL

conda activate Video-Tree-TRM & pytest tests/unit/test_tree_index.py -v 2>&1 | tail -20

预期：所有测试 FAIL（ModuleNotFoundError: No module named 'app.tree.index'）

Step 3: 实现 app/tree/index.py

从 reference/video_tree_trm/tree_index.py 迁移，关键改造：

新增 L3Card、L2Card、L1Card frozen dataclass
L3Node.description / L2Node.description / L1Node.summary 改为 property（从 card 派生）
节点增加 card 字段（替代原来的 description 直接字段）
L3Node 新增 subtitle: str | None 字段
to_dict() / from_dict() 适配 card dict 序列化
load_json() 反序列化时校验 ID 唯一性
删除 pickle 序列化（不需要）
日志用 loguru 替代 utils/logger_system
保留 embed_all(), l1_embeddings(), l2_embeddings_of(), l3_embeddings_of(), get_node(), save_l1_json(), load_l1_json() 的全部逻辑

逐行参考 reference/video_tree_trm/tree_index.py 确保不遗漏：

_embed_to_str() / _embed_from_str(): 保持不变
IndexMeta: 保持不变
TreeIndex.is_embedded: 保持不变
TreeIndex.embed_all(): 保持不变（L3 按 L2 分组批量 embed）
TreeIndex.l1_embeddings() / l2_embeddings_of() / l3_embeddings_of(): 保持不变
Step 4: 运行测试，确认全部 PASS

conda activate Video-Tree-TRM & pytest tests/unit/test_tree_index.py -v

预期：全部 PASS

Step 5: lint 检查

conda activate Video-Tree-TRM & ruff check app/tree/index.py --fix && ruff format app/tree/index.py

Step 6: 提交

git add app/tree/index.py tests/unit/test_tree_index.py
git commit -m "feat(tree): TreeIndex 数据结构 — Card 体系 + 节点 + 序列化"

Task 1.5: TreeConfig 数据类

Files:

Create: app/tree/config.py

说明: 定义 TreeConfig frozen dataclass，字段对齐 config/default.yaml 的 tree: 段。提供 from_dict() 工厂方法。

Step 1: 创建 app/tree/config.py

"""建树模块配置。"""

from __future__ import annotations

from dataclasses import dataclass


@dataclass(frozen=True)
class TreeConfig:
    """建树配置参数，字段对齐 config/default.yaml 的 tree: 段。"""

    l1_segment_duration: float = 600.0
    l2_clip_duration: float = 60.0
    l3_fps: float = 0.5
    l2_representative_frames: int = 6
    cache_dir: str = "cache/trees"
    concurrency: int = 16
    subtitle_inject: bool = True
    srt_window_sec: float = 5.0

    @classmethod
    def from_dict(cls, d: dict) -> TreeConfig:
        """从 YAML 解析后的 dict 构造。"""
        return cls(**{k: v for k, v in d.items() if k in cls.__dataclass_fields__})

Step 2: 提交

git add app/tree/config.py
git commit -m "feat(tree): TreeConfig 配置 dataclass"

Task 2: EmbeddingProvider Protocol + 适配器

Files:

Modify: app/ports.py
Create: adapters/embedding.py
Test: tests/unit/test_embedding_adapter.py

说明: 定义 EmbeddingProvider Protocol，实现 local/remote 双后端适配器。从 reference/video_tree_trm/embeddings.py 迁移，拆分为 Protocol + 实现。

Step 1: 编写失败测试

# tests/unit/test_embedding_adapter.py
"""EmbeddingProvider 适配器单元测试。"""

from __future__ import annotations

import numpy as np
import pytest

from app.ports import EmbeddingProvider


class TestEmbeddingProviderProtocol:
    def test_protocol_shape(self):
        """确认 Protocol 定义了 embed() 和 dim 属性。"""
        assert hasattr(EmbeddingProvider, "embed")
        assert hasattr(EmbeddingProvider, "dim")


class TestMockEmbeddingProvider:
    """用 mock 测试 Protocol 契约。"""

    def test_embed_single_text(self):
        from adapters.embedding import LocalEmbeddingProvider

        # 仅测试接口——实际初始化需要模型，这里先跳过
        # 真实测试需在 integration 中做
        pass

    def test_embed_returns_correct_shape(self):
        """用手工 mock 验证契约。"""

        class FakeEmbed:
            @property
            def dim(self) -> int:
                return 4

            def embed(self, texts):
                if isinstance(texts, str):
                    texts = [texts]
                return np.random.randn(len(texts), 4).astype(np.float32)

        provider = FakeEmbed()
        result = provider.embed(["你好", "世界"])
        assert result.shape == (2, 4)
        assert isinstance(provider, EmbeddingProvider)

Step 2: 运行测试确认 FAIL

conda activate Video-Tree-TRM & pytest tests/unit/test_embedding_adapter.py -v 2>&1 | tail -10

Step 3: 实现 app/ports.py 新增 EmbeddingProvider

# app/ports.py — 完整重写（原文件仅一行 docstring）
"""应用层 Protocol 端口定义。"""

from __future__ import annotations

from typing import Protocol, runtime_checkable

import numpy as np


@runtime_checkable
class EmbeddingProvider(Protocol):
    """文本嵌入端口。"""

    @property
    def dim(self) -> int: ...

    def embed(self, texts: str | list[str]) -> np.ndarray: ...

Step 4: 实现 adapters/embedding.py

从 reference/video_tree_trm/embeddings.py 迁移全部逻辑（191 行），改造：

类名改为 LocalEmbeddingProvider / RemoteEmbeddingProvider
日志用 loguru
配置参数通过构造函数传入（不读 config 文件）
保留 embed() 和 embed_tensor() 接口
保留 L2 归一化逻辑
Step 5: 运行测试确认 PASS

conda activate Video-Tree-TRM & pytest tests/unit/test_embedding_adapter.py -v

Step 6: 提交

git add app/ports.py adapters/embedding.py tests/unit/test_embedding_adapter.py
git commit -m "feat(adapters): EmbeddingProvider Protocol + local/remote 双后端实现"

Task 3: VLMProvider 最小可用适配器

Files:

Create: adapters/vlm.py
Test: tests/unit/test_vlm_adapter.py

说明: 基于 GovernedLLMClient 的 VLM 包装器，将图片编码为 base64 嵌入 messages 中，通过已有的 GovernedLLMClient.chat() 发送。最小可用实现，满足 VLMProvider Protocol。

Step 1: 编写失败测试

测试 VLMProvider 的 Protocol 契约和 base64 图片编码逻辑。

Step 2: 运行测试确认 FAIL
Step 3: 实现 adapters/vlm.py

关键实现：

GovernedVLMClient.__init__(governed_llm: GovernedLLMClient) — 复用已有治理栈
chat_with_images(messages, images) — 将图片路径编码为 base64，构造 OpenAI vision API 格式的 messages，委托给 governed_llm.chat()
实现 VLMProvider Protocol
Step 4: 运行测试确认 PASS
Step 5: 提交

git add adapters/vlm.py tests/unit/test_vlm_adapter.py
git commit -m "feat(adapters): GovernedVLMClient — VLMProvider 最小可用实现"

Task 4: 字幕模块

Files:

Create: app/tree/subtitle.py
Test: tests/unit/test_subtitle.py

说明: SRT 解析 + 完整性检查 + 时间范围提取 + Voronoi 分配。从 TRM4 enhance/merge.py 的 parse_srt() + TRM3 tools/generate_subtitles.py 的 Voronoi 逻辑迁移。

Step 1: 编写失败测试

# tests/unit/test_subtitle.py
"""字幕模块单元测试。"""

from __future__ import annotations

import pytest

from app.tree.subtitle import (
    SRTEntry,
    SubtitleReport,
    parse_srt,
    check_subtitle_completeness,
    extract_subtitle_for_range,
    assign_subtitles_voronoi,
)


_SAMPLE_SRT = """\
1
00:00:01,000 --> 00:00:03,500
Hello world.

2
00:00:05,000 --> 00:00:08,000
<i>This is italic</i> text.

3
00:00:10,000 --> 00:00:12,000
Final line.
"""


class TestParseSrt:
    def test_basic_parse(self, tmp_path):
        srt_file = tmp_path / "test.srt"
        srt_file.write_text(_SAMPLE_SRT, encoding="utf-8")

        entries = parse_srt(str(srt_file))
        assert len(entries) == 3
        assert entries[0] == SRTEntry(start=1.0, end=3.5, text="Hello world.")
        assert entries[1].text == "This is italic text."  # HTML 标签已剥离

    def test_empty_srt(self, tmp_path):
        srt_file = tmp_path / "empty.srt"
        srt_file.write_text("", encoding="utf-8")
        entries = parse_srt(str(srt_file))
        assert entries == []

    def test_malformed_srt_skips_bad_blocks(self, tmp_path):
        """格式损坏的 block 被跳过，不影响正常 block。"""
        bad_srt = "garbage\n\n1\n00:00:01,000 --> 00:00:02,000\nGood line.\n"
        srt_file = tmp_path / "bad.srt"
        srt_file.write_text(bad_srt, encoding="utf-8")
        entries = parse_srt(str(srt_file))
        assert len(entries) == 1
        assert entries[0].text == "Good line."


class TestCompletenessCheck:
    def test_good_coverage(self):
        entries = [
            SRTEntry(0.0, 5.0, "a"),
            SRTEntry(5.0, 10.0, "b"),
        ]
        report = check_subtitle_completeness(entries, duration=10.0, min_coverage=0.5)
        assert report.usable is True
        assert report.coverage_ratio >= 0.5

    def test_poor_coverage(self):
        entries = [SRTEntry(0.0, 1.0, "short")]
        report = check_subtitle_completeness(entries, duration=100.0, min_coverage=0.3)
        assert report.usable is False

    def test_max_gap(self):
        entries = [
            SRTEntry(0.0, 1.0, "a"),
            SRTEntry(50.0, 51.0, "b"),
        ]
        report = check_subtitle_completeness(entries, duration=60.0)
        assert report.max_gap_sec >= 49.0


class TestExtractForRange:
    def test_overlap(self):
        entries = [
            SRTEntry(0.0, 5.0, "first"),
            SRTEntry(4.0, 8.0, "second"),
            SRTEntry(10.0, 12.0, "third"),
        ]
        text = extract_subtitle_for_range(entries, (3.0, 9.0))
        assert "first" in text
        assert "second" in text
        assert "third" not in text


class TestVoronoiAssign:
    def test_assigns_to_l3_nodes(self):
        from app.tree.index import (
            IndexMeta, TreeIndex, L1Node, L1Card,
            L2Node, L2Card, L3Node, L3Card,
        )

        l3_0 = L3Node(
            id="l1_0_l2_0_l3_0",
            card=L3Card("desc0", [], [], [], "", {}),
            timestamp=2.0,
        )
        l3_1 = L3Node(
            id="l1_0_l2_0_l3_1",
            card=L3Card("desc1", [], [], [], "", {}),
            timestamp=6.0,
        )
        l2 = L2Node(
            id="l1_0_l2_0",
            card=L2Card("evt", [], [], [], [], "", None),
            time_range=(0.0, 10.0),
            children=[l3_0, l3_1],
        )
        l1 = L1Node(
            id="l1_0",
            card=L1Card("scene", "", [], [], [], [], ""),
            time_range=(0.0, 10.0),
            children=[l2],
        )
        index = TreeIndex(
            metadata=IndexMeta("/test.mp4", "video"),
            roots=[l1],
        )

        entries = [
            SRTEntry(1.0, 3.0, "hello"),
            SRTEntry(5.0, 7.0, "world"),
        ]
        assign_subtitles_voronoi(index, entries)

        assert l3_0.subtitle is not None
        assert "hello" in l3_0.subtitle
        assert l3_1.subtitle is not None
        assert "world" in l3_1.subtitle

Step 2: 运行测试确认 FAIL

conda activate Video-Tree-TRM & pytest tests/unit/test_subtitle.py -v 2>&1 | tail -10

Step 3: 实现 app/tree/subtitle.py

从 TRM4 core/tree/enhance/merge.py:31-84（parse_srt, extract_subtitle_window）和 TRM3 tools/generate_subtitles.py:439-547（compute_effective_ranges, assign_subtitles）迁移。改造：

parse_srt() → 返回 list[SRTEntry]（frozen dataclass）
新增 check_subtitle_completeness() → 返回 SubtitleReport
新增 extract_subtitle_for_range() — 按时间重叠提取
assign_subtitles_voronoi() — 适配 TreeIndex 嵌套结构（遍历 L1→L2→L3），使用 Voronoi 中点策略
Step 4: 运行测试确认 PASS
Step 5: 提交

git add app/tree/subtitle.py tests/unit/test_subtitle.py
git commit -m "feat(tree): 字幕模块 — SRT 解析 + 完整性检查 + Voronoi 分配"

Task 5: 质量校验

Files:

Create: app/tree/verify.py
Test: tests/unit/test_verify.py

说明: 从 TRM4 core/tree/enhance/verify.py 迁移交叉校验逻辑，适配 TreeIndex + Card 体系。

Step 1: 编写失败测试

覆盖：

_normalize() 归一化
fuzzy_match() 模糊子串匹配
verify_tree() L2 entities 校验（有出处保留、无出处删除）
verify_tree() L2 visible_text 校验（每条须在 L3 visible_text 中有出处）
verify_tree() L1 visible_text 校验
verify_tree() L1 key_entities 校验（交叉验证 L2/L3 文本语料）
verify_tree() frozen Card 替换（创建新 Card 实例）
VerifyStats 统计（各字段保留/删除数量）
Step 2: 运行测试确认 FAIL
Step 3: 实现 app/tree/verify.py

从 TRM4 core/tree/enhance/verify.py 迁移：

_normalize(), fuzzy_match() — 保持不变
_collect_l3_text() — 改为从 L2Node.children 遍历 L3Node
verify_tree(index: TreeIndex) -> VerifyStats — 遍历 TreeIndex，校验 L2.card.entities、L2.card.visible_text、L1.card.visible_text、L1.card.key_entities
校验时创建新 Card 实例替换（因为 Card 是 frozen）
返回 VerifyStats dataclass

注意：TRM4 verify 还处理 named_entities、quantitative_facts、causal_links——这些字段在 6 字段 Card 中不存在，跳过。

Step 4: 运行测试确认 PASS
Step 5: 提交

git add app/tree/verify.py tests/unit/test_verify.py
git commit -m "feat(tree): 质量校验 — 交叉验证 entities/visible_text"

Task 6: VideoTreeBuilder [保真 #1 #2 #3]

Files:

Create: app/tree/video_builder.py
Test: tests/unit/test_video_builder.py

说明: 从 reference/video_tree_trm/video_tree_builder.py（994 行）迁移。核心算法保真：L2 轴心、VLM 批量 + JSON fallback、断点续跑。

保真校验检查点：

比对 _build_async() 的 L2→L3 链式并发结构（算法 #1）
比对 _call_vlm_batch_async() 的批量调用 + fallback 逻辑（算法 #2）
比对 _save_progress() / _load_progress() / _cleanup_intermediate_and_progress() 的断点机制（算法 #3）
Step 1: 编写失败测试

用 mock VLMProvider/LLMProvider 测试：

_segment_video() 时间切分
_get_l2_clips() L2 clip 划分
_parse_json_descriptions() JSON 解析 + fallback
build() 完整流程（mock VLM 返回固定 JSON）
断点续跑（模拟中断 + 恢复）
Step 2: 运行测试确认 FAIL
Step 3: 实现 app/tree/video_builder.py

逐行参考 reference/video_tree_trm/video_tree_builder.py 迁移，关键改造：

依赖注入：__init__(vlm: VLMProvider, llm: LLMProvider, config: TreeConfig) — 不再直接使用 LLMClient
VLM 调用：await self.vlm.chat_with_images(messages, images) → 提取 .content
LLM 调用：await self.llm.chat(messages) → 提取 .content
输出结构化 Card：VLM prompt 返回 JSON 对象数组，解析为 L3Card；解析失败走逐帧 fallback
L2 代表帧复用：先提取所有 L3 帧，L2 从中采样
字幕注入：build(video_path, srt_entries=None) 可选参数
断点续跑：保持 reference 的 progress.json + L1 中间 JSON 机制
清理：_cleanup_intermediate_and_progress() 在最终 JSON 成功后调用

VLM Prompt 增量修改（不简化原有内容）：

L3 批量 prompt 在 reference 原文基础上追加结构化输出格式：

_L3_VIDEO_PROMPT = (
    '该片段的整体内容: "{l2_description}"\n'
    "以下是该片段中连续的 {n} 帧画面。\n"
    "对每帧用一到两句话描述其具体画面内容。\n"
    "重点关注: 动作、物体变化、文字信息、人物表情。\n"
    "不要重复片段整体描述，聚焦每帧的区分性信息。\n"
    "{subtitle_block}"
    "对每帧返回一个 JSON 对象，包含以下字段：\n"
    "- frame_summary: 1-2句画面描述\n"
    "- visible_entities: 可见实体列表\n"
    "- ongoing_actions: 正在进行的动作列表\n"
    "- visible_text: 画面中可见文字列表\n"
    "- spatial_layout: 画面空间布局\n"
    '- visual_attributes: {{"lighting": "...", "dominant_colors": [...], "camera_angle": "..."}}\n'
    '只返回 JSON 数组，格式: [{{...}}, {{...}}, ...]，不要其他内容。'
)

类似地修改 L2、L1 prompt，追加结构化输出要求。

Step 4: 运行测试确认 PASS

conda activate Video-Tree-TRM & pytest tests/unit/test_video_builder.py -v

Step 5: 保真校验 — 逐一比对参考代码

对照 reference 检查三项算法核心逻辑未被简化：

_build_async() 中 asyncio.gather(*[_chain(j, clip) for j, clip in enumerate(clips)]) 的链式并发
_call_vlm_batch_async() 中 _L3_BATCH_SIZE=5 分批 + _parse_json_descriptions() 校验 + 逐帧 fallback
_save_progress() / _load_progress() / _has_l1_intermediate() / _cleanup_intermediate_and_progress() 的完整断点机制

Step 6: 提交

git add app/tree/video_builder.py tests/unit/test_video_builder.py
git commit -m "feat(tree): VideoTreeBuilder — L2轴心建树(算法#1) + VLM批量+fallback(算法#2) + 断点续跑(算法#3)"

Task 7: TreeEnvironment [保真 #12 变更]

Files:

Create: app/tree/environment.py
Test: tests/unit/test_tree_environment.py

说明: 从 TRM4 core/tree/environment.py（451 行）迁移，改为基于 TreeIndex。算法 #12 变更：分块 embedding → 单节点 embedding。保留祖先去重 + 锚定验证。

Step 1: 编写失败测试

覆盖：

view_node() 返回卡片内容 + 子节点概览
view_node(anchor=True) 锚定标记
search_similar() 余弦相似度 + 祖先去重
get_subtitle() 字幕查询
resolve_frame_paths() 帧路径解析
ID 索引映射（O(1) 查找）
Step 2: 运行测试确认 FAIL
Step 3: 实现 app/tree/environment.py

关键逻辑：

构造函数：接收 TreeIndex，构建 _id_to_node: dict[str, L1Node | L2Node | L3Node] 节点引用映射（O(1) 查找任意层级节点）
view_node(node_id, anchor=False)：
- 通过 _id_to_path O(1) 定位节点
- 格式化卡片字段为文本
- anchor=True 时为每个卡片字段行添加 [c1]、[s1] 锚标（从 TRM4 _node_anchored_text() 迁移）
- 列出子节点概览（ID + 时间范围 + 主描述前 120 字符）
search_similar(query, top_k, embed_fn)：
- 用 embed_fn(query) 获取 query embedding
- 与所有节点 embedding 计算余弦相似度
- 祖先去重（从 TRM4 迁移：any(s.startswith(nid + "_") for s in seen_prefixes)）
- 返回 top_k 结果列表
get_subtitle(node_id) / resolve_frame_paths(node_ids)：从 TRM4 迁移，适配 TreeIndex

算法 #12 变更记录：分块 embedding（4000 字符分块，每块独立 embedding）改为 per-node embedding（基于各节点 embedding 文本源：L3.description、L2.description、L1.summary）。理由：TreeIndex 已有 per-node embedding，分块是 flat-dict 时代的替代方案。Commit message 需标注"算法 #12 变更"。

Step 4: 运行测试确认 PASS
Step 5: 提交

git add app/tree/environment.py tests/unit/test_tree_environment.py
git commit -m "feat(tree): TreeEnvironment — 运行时数据访问 + 语义搜索(算法#12变更:分块→单节点embedding)"

Task 8: 修复模式 — 检测器

Files:

Create: app/tree/repair/__init__.py
Create: app/tree/repair/detector.py
Test: tests/unit/test_repair_detector.py
Step 1: 编写失败测试

覆盖：

L3 必填字段为空检测
L3 帧文件缺失检测
L2 无子节点检测
L2 时间空洞检测
NodeIssue dataclass 结构
Step 2: 运行测试确认 FAIL
Step 3: 实现 app/tree/repair/detector.py

@dataclass(frozen=True)
class NodeIssue:
    node_id: str
    level: int
    issue_type: str   # "empty_field" | "missing_frame" | "no_children" | "time_gap"
    details: str

def detect_issues(index: TreeIndex, frames_dir: Path | None = None) -> list[NodeIssue]:
    """扫描树，返回所有问题节点列表。"""

Step 4: 运行测试确认 PASS
Step 5: 提交

git add app/tree/repair/ tests/unit/test_repair_detector.py
git commit -m "feat(tree/repair): 缺失/低质量节点检测器"

Task 9: 修复模式 — 重生成器

Files:

Create: app/tree/repair/regenerator.py
Test: tests/unit/test_repair_regenerator.py
Step 1: 编写失败测试

用 mock VLM/LLM 测试：

L3 节点修复（VLM 重新描述帧）
L2 向上级联（LLM 从 L3 聚合）
L1 向上级联（LLM 从 L2 聚合）
RepairStats 统计
Step 2: 运行测试确认 FAIL
Step 3: 实现 app/tree/repair/regenerator.py

@dataclass(frozen=True)
class RepairStats:
    l3_repaired: int
    l2_regenerated: int
    l1_regenerated: int

async def repair_tree(
    index: TreeIndex,
    issues: list[NodeIssue],
    vlm: VLMProvider,
    llm: LLMProvider,
    frames_dir: Path,
    srt_entries: list[SRTEntry] | None = None,
) -> RepairStats:
    """修复有问题的节点，底向上级联。"""

底向上级联逻辑：

收集需修复的 L3 节点 → VLM 重新描述（复用现有 L2 描述作上下文）
收集受影响的 L2 节点（其 L3 children 被修复的）→ LLM 从 L3 聚合
收集受影响的 L1 节点 → LLM 从 L2 聚合

Step 4: 运行测试确认 PASS
Step 5: 提交

git add app/tree/repair/regenerator.py tests/unit/test_repair_regenerator.py
git commit -m "feat(tree/repair): VLM 重生成 + 底向上级联"

Task 10: 修复模式 — Q&A 反向补全

Files:

Create: app/tree/repair/supplement.py
Test: tests/unit/test_repair_supplement.py

说明: 从 TRM4 core/tree/enhance/supplement.py（401 行）迁移，适配 TreeIndex + Card。

Step 1: 编写失败测试

覆盖：

deduplicate_field() 去重
_inject_one() 单字段注入
apply_injections() 批量注入
类别白名单过滤
Step 2: 运行测试确认 FAIL
Step 3: 实现 app/tree/repair/supplement.py

从 TRM4 迁移：

_ALLOWED_CATEGORIES 白名单
deduplicate_field() — 保持不变
_inject_one() — 适配 Card frozen dataclass（创建新 Card 实例）
apply_injections() — 适配 TreeIndex
analyze_question() — LLM 调用分析缺失事实
supplement_tree() — 主入口，遍历 questions，收集注入指令，执行注入
Step 4: 运行测试确认 PASS
Step 5: 提交

git add app/tree/repair/supplement.py tests/unit/test_repair_supplement.py
git commit -m "feat(tree/repair): Q&A 反向补全 — 从 TRM4 supplement 迁移"

Task 11: 迁移工具

Files:

Create: tools/migrate_from_trm4.sh
Create: tools/convert_flat_to_treeindex.py

说明: 一次性迁移脚本。从 TRM4.zip 解压原始树 → 格式转换 → 拷贝资产 → 验收。

Step 1: 实现 tools/convert_flat_to_treeindex.py

"""一次性格式转换：TRM4 flat tree.json → TreeIndex JSON。

用法: python tools/convert_flat_to_treeindex.py <src_dir> <dst_dir>

app/core/adapters 不 import 此脚本。迁移完成后归档至 tools/archived/。
"""

核心逻辑：

读取 flat tree.json（{nodes: {id: {level, card, ...}}}）
按 level 分组：L1 → L2 → L3
按 parent_id/children_ids 重建嵌套关系
card dict → L1Card/L2Card/L3Card dataclass
组装 TreeIndex，调用 save_json()
Step 2: 实现 tools/migrate_from_trm4.sh

#!/usr/bin/env bash
# 从 TRM4.zip 迁移资产到 TRM5
# 用法: bash tools/migrate_from_trm4.sh /path/to/Video-Tree-TRM4.zip

set -euo pipefail
# 1. 解压到临时目录
# 2. 拷贝帧文件（rsync --ignore-existing）
# 3. 拷贝 SRT 字幕
# 4. 拷贝视频压缩包
# 5. 拷贝问题 JSON
# 6. 运行格式转换
# 7. 验收：检查 300 视频 + tree.json + frames 完整性
# 8. 输出报告

验收逻辑（§9.3）：

检查 300 个视频目录
每个 tree.json 可反序列化为 TreeIndex
每个 L3 的 frame_path 对应文件存在
SRT 文件数 ≥ 290
每个视频有 question JSON
缺失资产报告 + 非零缺失 exit code 1
Step 3: 手动测试迁移（在少量视频上验证）

# 仅解压一个视频测试转换
unzip -o -j /home/iomgaa/Projects/Video-Tree-TRM4.zip \
  "Video-Tree-TRM4/store/videos/wNpA02SNgUg/*" \
  -d /tmp/trm4_test/store/videos/wNpA02SNgUg/

conda activate Video-Tree-TRM & python tools/convert_flat_to_treeindex.py \
  /tmp/trm4_test/store/videos/ store/videos/ --dry-run

Step 4: 提交

git add tools/migrate_from_trm4.sh tools/convert_flat_to_treeindex.py
git commit -m "feat(tools): TRM4→TRM5 迁移工具 — 格式转换 + 资产拷贝 + 验收"

Task 12: 集成测试

Files:

Create: tests/integration/test_tree_build_e2e.py

说明: 端到端测试：mock VLM/LLM → VideoTreeBuilder 建树 → verify → subtitle 注入 → TreeEnvironment 查询 → 序列化 roundtrip。

Step 1: 编写集成测试

# tests/integration/test_tree_build_e2e.py
"""建树模块端到端集成测试。

使用 mock VLM/LLM 测试完整建树流程：
build → verify → subtitle → environment → serialize
"""

import asyncio
import json
import pytest
import numpy as np

from app.tree.index import TreeIndex
from app.tree.verify import verify_tree
from app.tree.subtitle import SRTEntry, assign_subtitles_voronoi
from app.tree.environment import TreeEnvironment


class MockVLM:
    """返回固定结构化 JSON 的 mock VLM。"""
    async def chat_with_images(self, messages, images, **kwargs):
        from core.types import LLMResponse
        n = len(images)
        cards = [
            {
                "frame_summary": f"帧{i}描述",
                "visible_entities": [f"实体{i}"],
                "ongoing_actions": [f"动作{i}"],
                "visible_text": [],
                "spatial_layout": "居中",
                "visual_attributes": {"lighting": "明亮"},
            }
            for i in range(n)
        ]
        return LLMResponse(
            content=json.dumps(cards, ensure_ascii=False),
            thinking="", model="mock", provider="mock",
            prompt_tokens=0, completion_tokens=0,
            latency_ms=0, ttft_ms=None, max_inter_token_ms=None,
            cache_hit=False, call_id="mock",
        )


class MockLLM:
    """返回固定文本的 mock LLM。"""
    async def chat(self, messages, **kwargs):
        from core.types import LLMResponse
        return LLMResponse(
            content='{"event_description":"事件","entities":[],"actions":[],"action_subjects":[],"visible_text":[],"spatial_relations":"","state_changes":null}',
            thinking="", model="mock", provider="mock",
            prompt_tokens=0, completion_tokens=0,
            latency_ms=0, ttft_ms=None, max_inter_token_ms=None,
            cache_hit=False, call_id="mock",
        )


class TestTreeBuildE2E:
    def test_verify_then_environment(self, tmp_path):
        """构造最小树 → verify → subtitle → environment 查询。"""
        from app.tree.index import (
            IndexMeta, TreeIndex, L1Node, L1Card,
            L2Node, L2Card, L3Node, L3Card,
        )
        from app.tree.verify import verify_tree
        from app.tree.subtitle import SRTEntry, assign_subtitles_voronoi
        from app.tree.environment import TreeEnvironment

        l3 = L3Node(
            id="l1_0_l2_0_l3_0",
            card=L3Card("帧描述", ["真实实体"], ["动作"], ["文字"], "居中", {}),
            timestamp=2.0,
            frame_path="frames/l1_0_l2_0_l3_0.jpg",
        )
        l2 = L2Node(
            id="l1_0_l2_0",
            card=L2Card("事件", ["真实实体", "幻觉实体"], [], [], ["文字"], "", None),
            time_range=(0.0, 10.0),
            children=[l3],
        )
        l1 = L1Node(
            id="l1_0",
            card=L1Card("场景", "", ["真实实体"], [], [], ["文字"], ""),
            time_range=(0.0, 10.0),
            children=[l2],
        )
        index = TreeIndex(metadata=IndexMeta("/test.mp4", "video"), roots=[l1])

        # verify 应删除 L2 中无 L3 出处的"幻觉实体"
        stats = verify_tree(index)
        assert "幻觉实体" not in index.roots[0].children[0].card.entities

        # subtitle 注入
        entries = [SRTEntry(1.0, 3.0, "hello")]
        assign_subtitles_voronoi(index, entries)
        assert l3.subtitle is not None

        # environment 查询
        env = TreeEnvironment(index)
        result = env.view_node("l1_0_l2_0_l3_0")
        assert "帧描述" in result

        # 序列化 roundtrip
        path = tmp_path / "tree.json"
        index.save_json(str(path))
        loaded = TreeIndex.load_json(str(path))
        assert len(loaded.roots) == 1
        assert loaded.roots[0].children[0].children[0].subtitle is not None

Step 2: 运行集成测试

conda activate Video-Tree-TRM & pytest tests/integration/test_tree_build_e2e.py -v

Step 3: 提交

git add tests/integration/test_tree_build_e2e.py
git commit -m "test(tree): 建树模块端到端集成测试"

Task 13: 最终 lint + 覆盖率

Step 1: 全量 lint

conda activate Video-Tree-TRM & ruff check app/tree/ adapters/embedding.py adapters/vlm.py --fix
conda activate Video-Tree-TRM & ruff format app/tree/ adapters/embedding.py adapters/vlm.py

Step 2: 运行全部测试 + 覆盖率

conda activate Video-Tree-TRM & pytest tests/ --cov=app/tree --cov=adapters/embedding --cov=adapters/vlm --cov-report=term-missing -v

预期：覆盖率 ≥ 80%

Step 3: 提交

git add -A
git commit -m "chore: lint + 覆盖率达标"

核心算法保真校验结果

本计划涉及 4 项核心算法：

#	算法	Task	保真状态
1	L2 轴心建树策略	Task 6	保真——逐行迁移 `_build_async()` 链式并发结构
2	VLM 批量帧描述 + JSON fallback	Task 6	保真——`_L3_BATCH_SIZE=5`、`_parse_json_descriptions()` + 逐帧 fallback
3	断点续跑机制	Task 6	保真——`progress.json` + L1 中间 JSON + cleanup
12	树环境语义搜索	Task 7	变更——分块 embedding → 单节点 embedding；祖先去重 + 锚定验证保留

算法 #4-#11、#13 不在本计划范围内（属于 harness / evolution / retriever 模块）。

41 KiB Raw Blame History Unescape Escape

建树模块竖切实现计划

文件结构总览

Task 1: TreeIndex 数据结构

Task 1.5: TreeConfig 数据类

Task 2: EmbeddingProvider Protocol + 适配器

Task 3: VLMProvider 最小可用适配器

Task 4: 字幕模块

Task 5: 质量校验

Task 6: VideoTreeBuilder [保真 #1 #2 #3]

Task 7: TreeEnvironment [保真 #12 变更]

Task 8: 修复模式 — 检测器

Task 9: 修复模式 — 重生成器

Task 10: 修复模式 — Q&A 反向补全

Task 11: 迁移工具

Task 12: 集成测试

Task 13: 最终 lint + 覆盖率

核心算法保真校验结果

41 KiB

Raw Blame History