feat(harness): momentum.py — async 慢更新动量生成

2026-07-07 13:01:04 -04:00
parent a550d39e1c
commit 886a444d1d
2 changed files with 439 additions and 0 deletions
@@ -0,0 +1,261 @@
+"""慢更新动量生成（app/harness/momentum.py）单元测试。
+
+覆盖:
+- _categorize_pair 四分类 + 缺键 KeyError
+- _format_comparison_pairs 分组排序 + 空列表
+- run_slow_momentum 正常解析 + 解析失败保留 prev_guidance
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+from unittest.mock import AsyncMock
+
+import pytest
+
+from app.harness.momentum import (
+    IMPROVED,
+    PERSISTENT_FAIL,
+    REGRESSED,
+    STABLE_SUCCESS,
+    _categorize_pair,
+    _format_comparison_pairs,
+    run_slow_momentum,
+)
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+# =========================================================================
+# _categorize_pair
+# =========================================================================
+
+
+class TestCategorizePair:
+    """_categorize_pair 四分类测试。"""
+
+    def test_categorize_pair_all_four(self) -> None:
+        """四种 correct_prev/correct_curr 组合应返回对应类别常量。"""
+        assert _categorize_pair({"correct_prev": False, "correct_curr": True}) == IMPROVED
+        assert _categorize_pair({"correct_prev": True, "correct_curr": False}) == REGRESSED
+        assert _categorize_pair({"correct_prev": False, "correct_curr": False}) == PERSISTENT_FAIL
+        assert _categorize_pair({"correct_prev": True, "correct_curr": True}) == STABLE_SUCCESS
+
+    def test_categorize_pair_truthy_values(self) -> None:
+        """非布尔真值（int / str）也能正确分类。"""
+        assert _categorize_pair({"correct_prev": 0, "correct_curr": 1}) == IMPROVED
+        assert _categorize_pair({"correct_prev": "yes", "correct_curr": ""}) == REGRESSED
+
+    def test_categorize_pair_missing_key(self) -> None:
+        """缺 correct_prev 或 correct_curr 键时抛 KeyError。"""
+        with pytest.raises(KeyError):
+            _categorize_pair({"correct_prev": True})
+        with pytest.raises(KeyError):
+            _categorize_pair({"correct_curr": False})
+        with pytest.raises(KeyError):
+            _categorize_pair({})
+
+
+# =========================================================================
+# _format_comparison_pairs
+# =========================================================================
+
+
+class TestFormatComparisonPairs:
+    """_format_comparison_pairs 分组排序测试。"""
+
+    def test_format_comparison_pairs_empty(self) -> None:
+        """空列表返回占位说明。"""
+        result = _format_comparison_pairs([])
+        assert "无可用" in result
+
+    def test_format_comparison_pairs_order(self) -> None:
+        """REGRESSED 标题应出现在 IMPROVED 标题之前（伤害信号优先）。"""
+        pairs = [
+            {
+                "question": "Q1",
+                "prev_prediction": "A",
+                "curr_prediction": "B",
+                "correct_prev": False,
+                "correct_curr": True,
+            },
+            {
+                "question": "Q2",
+                "prev_prediction": "C",
+                "curr_prediction": "D",
+                "correct_prev": True,
+                "correct_curr": False,
+            },
+        ]
+        result = _format_comparison_pairs(pairs)
+        regressed_pos = result.index("回退")
+        improved_pos = result.index("改善")
+        assert regressed_pos < improved_pos, "REGRESSED 应排在 IMPROVED 之前"
+
+    def test_format_comparison_pairs_all_categories(self) -> None:
+        """四种类别的 pair 都能被正确分组。"""
+        pairs = [
+            {"question": "Q1", "correct_prev": False, "correct_curr": True},
+            {"question": "Q2", "correct_prev": True, "correct_curr": False},
+            {"question": "Q3", "correct_prev": False, "correct_curr": False},
+            {"question": "Q4", "correct_prev": True, "correct_curr": True},
+        ]
+        result = _format_comparison_pairs(pairs)
+        assert "固定样本总数：4" in result
+        # 每个类别都应标注 1 题
+        assert "1 题" in result
+
+    def test_format_comparison_pairs_missing_key(self) -> None:
+        """pair 缺键时 KeyError 不被吞。"""
+        with pytest.raises(KeyError):
+            _format_comparison_pairs([{"question": "Q1"}])
+
+
+# =========================================================================
+# run_slow_momentum
+# =========================================================================
+
+
+def _make_mock_llm(response_content: str) -> Any:
+    """构造一个返回指定 content 的 mock LLMProvider。"""
+
+    @dataclass(frozen=True)
+    class _FakeResponse:
+        content: str
+        thinking: str = ""
+        model: str = "mock"
+        provider: str = "mock"
+        prompt_tokens: int = 0
+        completion_tokens: int = 0
+        latency_ms: int = 0
+        ttft_ms: float | None = None
+        max_inter_token_ms: float | None = None
+        cache_hit: bool = False
+        call_id: str = "test-call-id"
+
+    mock_llm = AsyncMock()
+    mock_llm.chat.return_value = _FakeResponse(content=response_content)
+    return mock_llm
+
+
+@pytest.mark.asyncio
+async def test_run_slow_momentum_basic(tmp_path: Path) -> None:
+    """正常解析时返回新的动量指导文本。"""
+    # 准备 prompt 文件
+    prompt_file = tmp_path / "slow_momentum.md"
+    prompt_file.write_text("你是一个慢更新动量裁判。", encoding="utf-8")
+
+    new_guidance_text = "新一轮的动量指导内容"
+    llm_response = json.dumps({"slow_update_content": new_guidance_text}, ensure_ascii=False)
+    mock_llm = _make_mock_llm(llm_response)
+
+    result = await run_slow_momentum(
+        llm=mock_llm,
+        diagnose_prompts_dir=tmp_path,
+        skill_content="当前 skill 正文",
+        prev_skill="上一版 skill 正文",
+        prev_guidance="旧的动量指导",
+        comparison_pairs=[
+            {
+                "question": "Q1",
+                "prev_prediction": "A",
+                "curr_prediction": "B",
+                "correct_prev": False,
+                "correct_curr": True,
+            }
+        ],
+    )
+    assert result == new_guidance_text
+    mock_llm.chat.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_run_slow_momentum_parse_failure(tmp_path: Path) -> None:
+    """LLM 返回无法解析的内容时保留 prev_guidance。"""
+    prompt_file = tmp_path / "slow_momentum.md"
+    prompt_file.write_text("你是一个慢更新动量裁判。", encoding="utf-8")
+
+    mock_llm = _make_mock_llm("这不是 JSON，无法解析")
+    prev_guidance = "应该被保留的旧动量指导"
+
+    result = await run_slow_momentum(
+        llm=mock_llm,
+        diagnose_prompts_dir=tmp_path,
+        skill_content="当前 skill",
+        prev_skill="上一版 skill",
+        prev_guidance=prev_guidance,
+        comparison_pairs=[
+            {
+                "question": "Q1",
+                "prev_prediction": "A",
+                "curr_prediction": "B",
+                "correct_prev": True,
+                "correct_curr": True,
+            }
+        ],
+    )
+    assert result == prev_guidance
+
+
+@pytest.mark.asyncio
+async def test_run_slow_momentum_missing_field(tmp_path: Path) -> None:
+    """LLM 返回合法 JSON 但缺少 slow_update_content 字段时保留 prev_guidance。"""
+    prompt_file = tmp_path / "slow_momentum.md"
+    prompt_file.write_text("你是一个慢更新动量裁判。", encoding="utf-8")
+
+    llm_response = json.dumps({"other_field": "无关内容"}, ensure_ascii=False)
+    mock_llm = _make_mock_llm(llm_response)
+    prev_guidance = "应该被保留的旧动量指导"
+
+    result = await run_slow_momentum(
+        llm=mock_llm,
+        diagnose_prompts_dir=tmp_path,
+        skill_content="当前 skill",
+        prev_skill="上一版 skill",
+        prev_guidance=prev_guidance,
+        comparison_pairs=[],
+    )
+    assert result == prev_guidance
+
+
+@pytest.mark.asyncio
+async def test_run_slow_momentum_keyerror_not_swallowed(tmp_path: Path) -> None:
+    """comparison_pairs 缺键时 KeyError 不被 ValueError 吞掉。"""
+    prompt_file = tmp_path / "slow_momentum.md"
+    prompt_file.write_text("你是一个慢更新动量裁判。", encoding="utf-8")
+
+    mock_llm = _make_mock_llm('{"slow_update_content": "ok"}')
+
+    with pytest.raises(KeyError):
+        await run_slow_momentum(
+            llm=mock_llm,
+            diagnose_prompts_dir=tmp_path,
+            skill_content="当前 skill",
+            prev_skill="上一版 skill",
+            prev_guidance="旧指导",
+            comparison_pairs=[{"question": "Q1"}],  # 缺 correct_prev/correct_curr
+        )
+
+
+@pytest.mark.asyncio
+async def test_run_slow_momentum_fenced_json(tmp_path: Path) -> None:
+    """LLM 返回 fenced code block 中的 JSON 也能正确解析。"""
+    prompt_file = tmp_path / "slow_momentum.md"
+    prompt_file.write_text("你是一个慢更新动量裁判。", encoding="utf-8")
+
+    new_guidance = "从 fenced block 中提取的指导"
+    llm_response = f'```json\n{{"slow_update_content": "{new_guidance}"}}\n```'
+    mock_llm = _make_mock_llm(llm_response)
+
+    result = await run_slow_momentum(
+        llm=mock_llm,
+        diagnose_prompts_dir=tmp_path,
+        skill_content="当前 skill",
+        prev_skill="上一版 skill",
+        prev_guidance="旧指导",
+        comparison_pairs=[],
+    )
+    assert result == new_guidance