fischer
/
fischer-agentkit
1
0
Fork 0
Code Issues Pull Requests Actions 85 Packages Projects Releases Wiki Activity

feat(iq): U7 ABTester prompt-version offline comparison (R14) 

- 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.
This commit is contained in:
Chiguyong 2026-07-06 14:02:14 +08:00
parent a2deeac0d6
commit 4e2c7c5cac
3 changed files with 370 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

118
src/agentkit/evolution/ab_tester.py
View File

@ -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),
}

62
src/agentkit/memory/episodic.py
View File

@ -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).

197
tests/unit/test_ab_tester_prompt.py Normal file
View File

@ -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"