From 4e2c7c5cac63f1bd7e8864259134c74dc52b8bb0 Mon Sep 17 00:00:00 2001
From: Chiguyong <chiguyong@ChiGuyongs-MacBook-Pro.local>
Date: Mon, 6 Jul 2026 14:02:14 +0800
Subject: [PATCH] feat(iq): U7 ABTester prompt-version offline comparison (R14)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- EpisodicMemory.list_prompt_reflections_by_hash(task_hash): exact
  query for all prompt_reflection records matching task_hash.
  ponytail: O(N) scan with N<100 typical; GIN index upgrade path noted.
- ABTester.__init__: accepts optional episodic_memory parameter.
- ABTester.compare_prompt_versions(task_hash) -> dict: retrieves all
  versions for a task_hash, sorts by score descending, returns
  {versions, best_version, recommendation, total_versions}.
  Offline-only — no online bandit. Non-blocking on retrieval failure.
- 8 unit tests covering no-episodic, multi-version sort, single version,
  empty result, retrieval failure, field completeness, low-score
  retention, task_hash echo.
---
 src/agentkit/evolution/ab_tester.py | 118 ++++++++++++++++-
 src/agentkit/memory/episodic.py     |  62 +++++++++
 tests/unit/test_ab_tester_prompt.py | 197 ++++++++++++++++++++++++++++
 3 files changed, 370 insertions(+), 7 deletions(-)
 create mode 100644 tests/unit/test_ab_tester_prompt.py

diff --git a/src/agentkit/evolution/ab_tester.py b/src/agentkit/evolution/ab_tester.py
index 9a01572..e113987 100644
--- a/src/agentkit/evolution/ab_tester.py
+++ b/src/agentkit/evolution/ab_tester.py
@@ -12,6 +12,7 @@ from sqlalchemy.exc import DBAPIError
 
 if TYPE_CHECKING:
     from agentkit.evolution.evolution_store import InMemoryEvolutionStore
+    from agentkit.memory.episodic import EpisodicMemory
 
 logger = logging.getLogger(__name__)
 
@@ -19,6 +20,7 @@ logger = logging.getLogger(__name__)
 @dataclass
 class ABTestConfig:
     """A/B 测试配置"""
+
     test_id: str
     agent_name: str
     change_type: str  # prompt / strategy / pipeline
@@ -31,6 +33,7 @@ class ABTestConfig:
 @dataclass
 class ABTestResult:
     """A/B 测试结果"""
+
     test_id: str
     control_metric: float
     experiment_metric: float
@@ -52,11 +55,15 @@ class ABTester:
         self,
         evolution_store: "InMemoryEvolutionStore | None" = None,
         min_samples: int = 10,
+        episodic_memory: "EpisodicMemory | None" = None,
     ):
         self._tests: dict[str, ABTestConfig] = {}
         self._results: dict[str, list[tuple[str, float]]] = {}  # test_id -> [(group, metric)]
         self._evolution_store = evolution_store
         self._default_min_samples = min_samples
+        # IQ-Boost/U7 (R14): optional EpisodicMemory for prompt-version comparison.
+        # None = compare_prompt_versions() returns empty result (backward-compatible).
+        self._episodic_memory = episodic_memory
 
     def create_test(self, config: ABTestConfig) -> None:
         """创建 A/B 测试"""
@@ -115,7 +122,9 @@ class ABTester:
         experiment_metrics = [m for g, m in results if g == "experiment"]
 
         control_avg = sum(control_metrics) / len(control_metrics) if control_metrics else 0.0
-        experiment_avg = sum(experiment_metrics) / len(experiment_metrics) if experiment_metrics else 0.0
+        experiment_avg = (
+            sum(experiment_metrics) / len(experiment_metrics) if experiment_metrics else 0.0
+        )
 
         try:
             await self._evolution_store.record_ab_test_result(
@@ -144,11 +153,18 @@ class ABTester:
         control_metrics = [m for g, m in results if g == "control"]
         experiment_metrics = [m for g, m in results if g == "experiment"]
 
-        if len(control_metrics) < config.min_samples or len(experiment_metrics) < config.min_samples:
+        if (
+            len(control_metrics) < config.min_samples
+            or len(experiment_metrics) < config.min_samples
+        ):
             return ABTestResult(
                 test_id=test_id,
-                control_metric=sum(control_metrics) / len(control_metrics) if control_metrics else 0,
-                experiment_metric=sum(experiment_metrics) / len(experiment_metrics) if experiment_metrics else 0,
+                control_metric=sum(control_metrics) / len(control_metrics)
+                if control_metrics
+                else 0,
+                experiment_metric=sum(experiment_metrics) / len(experiment_metrics)
+                if experiment_metrics
+                else 0,
                 control_samples=len(control_metrics),
                 experiment_samples=len(experiment_metrics),
                 is_significant=False,
@@ -159,10 +175,16 @@ class ABTester:
         control_mean = sum(control_metrics) / len(control_metrics)
         experiment_mean = sum(experiment_metrics) / len(experiment_metrics)
 
-        control_var = sum((m - control_mean) ** 2 for m in control_metrics) / (len(control_metrics) - 1)
-        experiment_var = sum((m - experiment_mean) ** 2 for m in experiment_metrics) / (len(experiment_metrics) - 1)
+        control_var = sum((m - control_mean) ** 2 for m in control_metrics) / (
+            len(control_metrics) - 1
+        )
+        experiment_var = sum((m - experiment_mean) ** 2 for m in experiment_metrics) / (
+            len(experiment_metrics) - 1
+        )
 
-        pooled_se = math.sqrt(control_var / len(control_metrics) + experiment_var / len(experiment_metrics))
+        pooled_se = math.sqrt(
+            control_var / len(control_metrics) + experiment_var / len(experiment_metrics)
+        )
 
         # Handle zero variance case: if means differ but variance is zero,
         # the difference is clearly significant
@@ -201,3 +223,85 @@ class ABTester:
     def _normal_cdf(x: float) -> float:
         """标准正态分布 CDF 近似"""
         return 0.5 * (1 + math.erf(x / math.sqrt(2)))
+
+    # ── IQ-Boost/U7: Prompt-version offline comparison (R14) ────────────
+
+    async def compare_prompt_versions(self, task_hash: str) -> dict[str, object]:
+        """离线对比同一 task_hash 的多个 prompt 版本效果 (U7/R14).
+
+        从 EpisodicMemory 检索该 task_hash 的所有 prompt_reflection 记录，
+        按 score 降序排列，返回对比结果 + 推荐保留版本。
+
+        离线验证 — 不在线 bandit，仅基于历史 score 对比。
+
+        Returns:
+            {
+                "task_hash": str,
+                "versions": [{"version": int, "score": float, "timestamp": str,
+                              "reflection_summary": str, "improved_prompt": str}],
+                "best_version": dict | None,  # score 最高的版本
+                "recommendation": str,  # "keep_best" | "no_data"
+                "total_versions": int,
+            }
+            无 episodic_memory 或无记录时返回 "no_data" recommendation。
+        """
+        if self._episodic_memory is None:
+            return {
+                "task_hash": task_hash,
+                "versions": [],
+                "best_version": None,
+                "recommendation": "no_data",
+                "total_versions": 0,
+            }
+
+        try:
+            items = await self._episodic_memory.list_prompt_reflections_by_hash(task_hash)
+        except (DBAPIError, RuntimeError, ValueError, KeyError, OSError) as e:
+            logger.warning(f"U7: compare_prompt_versions retrieval failed: {e}")
+            return {
+                "task_hash": task_hash,
+                "versions": [],
+                "best_version": None,
+                "recommendation": "no_data",
+                "total_versions": 0,
+            }
+
+        if not items:
+            return {
+                "task_hash": task_hash,
+                "versions": [],
+                "best_version": None,
+                "recommendation": "no_data",
+                "total_versions": 0,
+            }
+
+        # Sort by score descending (best first)
+        sorted_items = sorted(items, key=lambda it: it.score, reverse=True)
+
+        versions: list[dict[str, object]] = []
+        for item in sorted_items:
+            meta = item.metadata or {}
+            value = item.value if isinstance(item.value, dict) else {}
+            reflection_text = value.get("reflection", "") if isinstance(value, dict) else ""
+            improved_prompt = value.get("output_summary", "") if isinstance(value, dict) else ""
+            versions.append(
+                {
+                    "version": meta.get("version", 1),
+                    "score": item.score or 0.0,
+                    "timestamp": meta.get("created_at", "")
+                    or (item.created_at.isoformat() if item.created_at else ""),
+                    "reflection_summary": (reflection_text[:200] if reflection_text else ""),
+                    "improved_prompt": (improved_prompt[:500] if improved_prompt else ""),
+                }
+            )
+
+        best = versions[0] if versions else None
+        recommendation = "keep_best" if best else "no_data"
+
+        return {
+            "task_hash": task_hash,
+            "versions": versions,
+            "best_version": best,
+            "recommendation": recommendation,
+            "total_versions": len(versions),
+        }
diff --git a/src/agentkit/memory/episodic.py b/src/agentkit/memory/episodic.py
index 3ced8af..ca16113 100644
--- a/src/agentkit/memory/episodic.py
+++ b/src/agentkit/memory/episodic.py
@@ -487,6 +487,68 @@ class EpisodicMemory(Memory):
             logger.warning(f"U5: failed to search prompt reflections: {e}")
             return []
 
+    async def list_prompt_reflections_by_hash(
+        self,
+        task_hash: str,
+        agent_name: str | None = None,
+    ) -> list[MemoryItem]:
+        """精确查询同一 task_hash 的所有 prompt 反思版本 (U7/R14).
+
+        用于 ABTester 离线对比同一任务的不同 prompt 版本效果。
+
+        ponytail: ceiling = O(N) 全表扫描 task_type='prompt_reflection' 记录后
+        在 Python 侧过滤 task_hash。N 通常 <100（一个 task_hash 的版本数有限）。
+        升级路径 = 在 metadata_['task_hash'] 上加 GIN 索引 + JSONB ->> 算符查询。
+
+        Returns empty list on failure (non-raising).
+        """
+        from sqlalchemy import select
+
+        async with self._session_factory() as db:
+            try:
+                Model = self._episodic_model
+                stmt = (
+                    select(Model)
+                    .where(Model.task_type == "prompt_reflection")
+                    .order_by(Model.created_at.desc())
+                    .limit(200)
+                )
+                result = await db.execute(stmt)
+                entries = result.scalars().all()
+            except (DBAPIError, ValueError, KeyError, RuntimeError, OSError) as e:
+                logger.warning(f"U7: list_prompt_reflections_by_hash failed: {e}")
+                return []
+
+        items: list[MemoryItem] = []
+        for entry in entries:
+            meta = entry.metadata_ or {}
+            if meta.get("task_hash") != task_hash:
+                continue
+            if agent_name and meta.get("agent_name") != agent_name:
+                continue
+            items.append(
+                MemoryItem(
+                    key=str(entry.id),
+                    value={
+                        "input_summary": entry.input_summary,
+                        "output_summary": entry.output_summary,
+                        "reflection": entry.reflection,
+                        "quality_score": entry.quality_score,
+                    },
+                    metadata={
+                        "agent_name": entry.agent_name,
+                        "task_type": entry.task_type,
+                        "task_hash": meta.get("task_hash", ""),
+                        "version": meta.get("version", 1),
+                        "score": entry.quality_score or 0.0,
+                        "created_at": entry.created_at.isoformat() if entry.created_at else None,
+                    },
+                    score=entry.quality_score or 0.0,
+                    created_at=entry.created_at or datetime.now(timezone.utc),
+                )
+            )
+        return items
+
     async def cleanup_expired(self, max_age_days: int = 30) -> int:
         """删除超过 max_age_days 天的记录 (U5/R15 TTL).
 
diff --git a/tests/unit/test_ab_tester_prompt.py b/tests/unit/test_ab_tester_prompt.py
new file mode 100644
index 0000000..fcf18c4
--- /dev/null
+++ b/tests/unit/test_ab_tester_prompt.py
@@ -0,0 +1,197 @@
+"""U7: ABTester prompt-version offline comparison (R14).
+
+Covers:
+- compare_prompt_versions(): no episodic_memory → "no_data"
+- Multiple versions sorted by score descending
+- Single version → that version is best_version
+- No matching versions → "no_data"
+- Retrieval failure → "no_data" (non-blocking)
+- best_version is the highest-scored
+- Low-score versions included (no cleanup in compare; recommendation only)
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from agentkit.evolution.ab_tester import ABTester
+from agentkit.memory.base import MemoryItem
+
+
+def _make_item(
+    version: int,
+    score: float,
+    task_hash: str = "abc123",
+    reflection: str = "reflection text",
+    improved_prompt: str = "improved prompt",
+    created_at: datetime | None = None,
+) -> MemoryItem:
+    """Build a MemoryItem mimicking list_prompt_reflections_by_hash output."""
+    if created_at is None:
+        created_at = datetime.now(timezone.utc)
+    return MemoryItem(
+        key=f"prompt_reflection:{task_hash}:{version}",
+        value={
+            "input_summary": "task input",
+            "output_summary": improved_prompt,
+            "reflection": reflection,
+            "quality_score": score,
+        },
+        metadata={
+            "agent_name": "test_agent",
+            "task_type": "prompt_reflection",
+            "task_hash": task_hash,
+            "version": version,
+            "score": score,
+            "created_at": created_at.isoformat(),
+        },
+        score=score,
+        created_at=created_at,
+    )
+
+
+class TestComparePromptVersions:
+    """Tests for ABTester.compare_prompt_versions()."""
+
+    @pytest.mark.asyncio
+    async def test_no_episodic_memory_returns_no_data(self):
+        """No episodic_memory wired → recommendation='no_data', empty versions."""
+        tester = ABTester(episodic_memory=None)
+        result = await tester.compare_prompt_versions("any_hash")
+
+        assert result["recommendation"] == "no_data"
+        assert result["total_versions"] == 0
+        assert result["versions"] == []
+        assert result["best_version"] is None
+        assert result["task_hash"] == "any_hash"
+
+    @pytest.mark.asyncio
+    async def test_multiple_versions_sorted_by_score_desc(self):
+        """3 versions (0.8, 0.6, 0.4) → best_version score=0.8."""
+        episodic = MagicMock()
+        episodic.list_prompt_reflections_by_hash = AsyncMock(
+            return_value=[
+                _make_item(version=1, score=0.4),
+                _make_item(version=2, score=0.8),
+                _make_item(version=3, score=0.6),
+            ]
+        )
+        tester = ABTester(episodic_memory=episodic)
+
+        result = await tester.compare_prompt_versions("abc123")
+
+        assert result["total_versions"] == 3
+        assert result["recommendation"] == "keep_best"
+        versions = result["versions"]
+        # Sorted descending by score
+        assert versions[0]["score"] == 0.8
+        assert versions[1]["score"] == 0.6
+        assert versions[2]["score"] == 0.4
+        # best_version is the highest-scored
+        assert result["best_version"]["version"] == 2
+        assert result["best_version"]["score"] == 0.8
+
+    @pytest.mark.asyncio
+    async def test_single_version_is_best(self):
+        """Only 1 version → that version is best_version."""
+        episodic = MagicMock()
+        episodic.list_prompt_reflections_by_hash = AsyncMock(
+            return_value=[_make_item(version=1, score=0.7)]
+        )
+        tester = ABTester(episodic_memory=episodic)
+
+        result = await tester.compare_prompt_versions("abc123")
+
+        assert result["total_versions"] == 1
+        assert result["recommendation"] == "keep_best"
+        assert result["best_version"]["version"] == 1
+        assert result["best_version"]["score"] == 0.7
+
+    @pytest.mark.asyncio
+    async def test_no_matching_versions_returns_no_data(self):
+        """EpisodicMemory returns empty list → 'no_data'."""
+        episodic = MagicMock()
+        episodic.list_prompt_reflections_by_hash = AsyncMock(return_value=[])
+        tester = ABTester(episodic_memory=episodic)
+
+        result = await tester.compare_prompt_versions("unknown_hash")
+
+        assert result["recommendation"] == "no_data"
+        assert result["total_versions"] == 0
+        assert result["best_version"] is None
+
+    @pytest.mark.asyncio
+    async def test_retrieval_failure_returns_no_data(self):
+        """list_prompt_reflections_by_hash raises → 'no_data' (non-blocking)."""
+        episodic = MagicMock()
+        episodic.list_prompt_reflections_by_hash = AsyncMock(
+            side_effect=RuntimeError("db connection failed")
+        )
+        tester = ABTester(episodic_memory=episodic)
+
+        result = await tester.compare_prompt_versions("abc123")
+
+        assert result["recommendation"] == "no_data"
+        assert result["total_versions"] == 0
+
+    @pytest.mark.asyncio
+    async def test_versions_include_reflection_and_prompt_fields(self):
+        """Each version dict carries reflection_summary + improved_prompt."""
+        episodic = MagicMock()
+        episodic.list_prompt_reflections_by_hash = AsyncMock(
+            return_value=[
+                _make_item(
+                    version=1,
+                    score=0.9,
+                    reflection="avoid retrying on schema error",
+                    improved_prompt="use structured output schema",
+                )
+            ]
+        )
+        tester = ABTester(episodic_memory=episodic)
+
+        result = await tester.compare_prompt_versions("abc123")
+
+        version = result["versions"][0]
+        assert "avoid retrying on schema error" in version["reflection_summary"]
+        assert "use structured output schema" in version["improved_prompt"]
+        assert version["version"] == 1
+        assert version["score"] == 0.9
+        # timestamp is a string (ISO format)
+        assert isinstance(version["timestamp"], str)
+
+    @pytest.mark.asyncio
+    async def test_low_score_versions_kept_in_list(self):
+        """Low-score versions remain in versions list (no cleanup; recommendation only)."""
+        episodic = MagicMock()
+        episodic.list_prompt_reflections_by_hash = AsyncMock(
+            return_value=[
+                _make_item(version=1, score=0.1),  # low score
+                _make_item(version=2, score=0.9),  # high score
+            ]
+        )
+        tester = ABTester(episodic_memory=episodic)
+
+        result = await tester.compare_prompt_versions("abc123")
+
+        # Both versions present — compare does not filter, only sorts
+        assert result["total_versions"] == 2
+        assert result["best_version"]["score"] == 0.9
+        # Low-score version still in list (for inspection)
+        scores = [v["score"] for v in result["versions"]]
+        assert 0.1 in scores
+        assert 0.9 in scores
+
+    @pytest.mark.asyncio
+    async def test_task_hash_echoed_in_result(self):
+        """task_hash field in result matches input."""
+        episodic = MagicMock()
+        episodic.list_prompt_reflections_by_hash = AsyncMock(return_value=[])
+        tester = ABTester(episodic_memory=episodic)
+
+        result = await tester.compare_prompt_versions("specific_hash_456")
+
+        assert result["task_hash"] == "specific_hash_456"