wehub-resource-sync
6c9c7fe7f3
CI / integration tests (3.13) (push) Failing after 1s
Commit lint / pull request title (push) Has been skipped
Docs / links (push) Failing after 1s
CI / unit tests (3.13) (push) Failing after 1s
CI / lint (push) Failing after 1s
CI / integration tests (push) Failing after 1s
CI / package build (push) Failing after 1s
Commit lint / commit messages (push) Failing after 1s
CI / unit tests (push) Failing after 1s
811 lines
28 KiB
Python
811 lines
28 KiB
Python
"""Reflection E2E test -- validates the full Reflection pipeline with real
|
|
LLM, real embedder, and LoCoMo conversation data.
|
|
|
|
Usage:
|
|
python tests/test_reflection_e2e.py # run all TCs
|
|
python tests/test_reflection_e2e.py --tc 1,2,14 # run selected TCs
|
|
python tests/test_reflection_e2e.py --verbose # verbose output
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
# benchmarks/run.py is the benchmark runner; add repo root to sys.path so
|
|
# the benchmarks package is importable from any working directory.
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from benchmarks.run import (
|
|
ANSWER_PROMPT,
|
|
JUDGE_SYSTEM_PROMPT,
|
|
JUDGE_USER_PROMPT,
|
|
EverosClient,
|
|
LLMClientPool,
|
|
_build_context,
|
|
_extract_final_answer,
|
|
_extract_json,
|
|
_parse_session_timestamp,
|
|
print_section,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Constants — session indices per storyline and golden data
|
|
# ---------------------------------------------------------------------------
|
|
|
|
DATA_PATH = Path(__file__).resolve().parent.parent / "data" / "locomo10.json"
|
|
|
|
ADOPTION_INIT_SESSIONS = [2, 8, 13, 17]
|
|
ADOPTION_UPDATE_SESSIONS = [19]
|
|
LGBTQ_INIT_SESSIONS = [1, 3, 5, 12]
|
|
LGBTQ_UPDATE_SESSIONS = [14]
|
|
PET_INIT_SESSIONS = [1, 5, 12, 24]
|
|
PET_UPDATE_SESSIONS = [27, 28]
|
|
HEALTH_INIT_SESSIONS = [2, 4, 8, 10, 13, 14]
|
|
HEALTH_UPDATE_SESSIONS = [16, 20]
|
|
|
|
QUERIES = {
|
|
"adoption": "What steps has Caroline taken toward adoption?",
|
|
"lgbtq": "How has Caroline dealt with discrimination?",
|
|
"pet": "How many pets does Andrew have and what are their names?",
|
|
"health": "How has Sam's diet and health journey been going?",
|
|
}
|
|
|
|
GOLDEN_FACTS = {
|
|
"adoption": [
|
|
"research",
|
|
"adoption council",
|
|
"applied",
|
|
"mentor",
|
|
"interview",
|
|
],
|
|
"lgbtq": [
|
|
"support group",
|
|
"school",
|
|
"pride",
|
|
"discriminat",
|
|
"apolog",
|
|
],
|
|
"pet": [
|
|
"no pet",
|
|
"toby",
|
|
"buddy",
|
|
"scout",
|
|
],
|
|
"health": [
|
|
"doctor",
|
|
"diet",
|
|
"before and after",
|
|
"snack",
|
|
"gastritis",
|
|
"weight watchers",
|
|
"struggl",
|
|
],
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data parsing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def load_locomo() -> list[dict[str, Any]]:
|
|
"""Load the LoCoMo dataset from the project data directory."""
|
|
with open(DATA_PATH) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def parse_sessions(
|
|
conv: dict[str, Any],
|
|
session_indices: list[int],
|
|
conv_index: int,
|
|
) -> list[dict[str, Any]]:
|
|
"""Parse LoCoMo sessions into the everos /add message format.
|
|
|
|
Returns a list of dicts, each with ``session_idx``, ``session_id``,
|
|
and ``messages`` (ready for the ``/api/v1/memory/add`` payload).
|
|
"""
|
|
raw = conv["conversation"]
|
|
results: list[dict[str, Any]] = []
|
|
for idx in session_indices:
|
|
key = f"session_{idx}"
|
|
if key not in raw:
|
|
raise ValueError(f"session {key} not found in conv {conv_index}")
|
|
date_key = f"{key}_date_time"
|
|
base_ts = _parse_session_timestamp(raw.get(date_key, ""))
|
|
session_id = f"refl_conv{conv_index}_s{idx}"
|
|
messages: list[dict[str, Any]] = []
|
|
for i, dia in enumerate(raw[key]):
|
|
messages.append(
|
|
{
|
|
"sender_id": f"{dia['speaker'].lower()}_conv{conv_index}",
|
|
"sender_name": dia["speaker"],
|
|
"role": "user",
|
|
"timestamp": base_ts + i * 30,
|
|
"content": [{"type": "text", "text": dia["text"]}],
|
|
}
|
|
)
|
|
results.append(
|
|
{
|
|
"session_idx": idx,
|
|
"session_id": session_id,
|
|
"messages": messages,
|
|
}
|
|
)
|
|
return results
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Infrastructure helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
_SYSTEM_DB = DATA_PATH.parent.parent / ".everos" / ".index" / "sqlite" / "system.db"
|
|
|
|
|
|
def print_episode_locations(
|
|
owner_id: str,
|
|
episodes: list[dict[str, Any]],
|
|
) -> None:
|
|
"""Print md paths for human review of merged vs source episodes."""
|
|
merged = [e for e in episodes if e.get("session_id") is None]
|
|
original = [e for e in episodes if e.get("session_id") is not None]
|
|
print(f"\n episode locations ({owner_id}):")
|
|
if merged:
|
|
for ep in merged:
|
|
print(f" [MERGED] {ep.get('id', '?')}")
|
|
if original:
|
|
for ep in original[:3]:
|
|
print(f" [source] {ep.get('id', '?')} session={ep.get('session_id')}")
|
|
if len(original) > 3:
|
|
print(f" ... and {len(original) - 3} more sources")
|
|
root = str(DATA_PATH.parent.parent / ".everos")
|
|
print(f" md root: {root}")
|
|
|
|
|
|
def _owner_id(speaker: str, conv_index: int) -> str:
|
|
"""Build the canonical owner_id for a speaker in a conversation."""
|
|
return f"{speaker.lower()}_conv{conv_index}"
|
|
|
|
|
|
def count_reflection_reports(owner_id: str) -> int:
|
|
"""Query SQLite directly to count reflection reports for an owner."""
|
|
import sqlite3
|
|
|
|
conn = sqlite3.connect(str(_SYSTEM_DB))
|
|
try:
|
|
cur = conn.execute(
|
|
"SELECT count(*) FROM reflection_report WHERE owner_id = ?",
|
|
(owner_id,),
|
|
)
|
|
return cur.fetchone()[0]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def count_deprecated_episodes(owner_id: str) -> int:
|
|
"""Query LanceDB via search with a special filter is not possible from
|
|
outside the server. Instead check reflection_report source_count as proxy."""
|
|
import sqlite3
|
|
|
|
conn = sqlite3.connect(str(_SYSTEM_DB))
|
|
try:
|
|
cur = conn.execute(
|
|
"SELECT coalesce(sum(source_count), 0) "
|
|
"FROM reflection_report WHERE owner_id = ?",
|
|
(owner_id,),
|
|
)
|
|
return cur.fetchone()[0]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def add_and_flush(
|
|
client: EverosClient,
|
|
sessions: list[dict[str, Any]],
|
|
*,
|
|
quiet: bool = True,
|
|
) -> None:
|
|
"""Ingest sessions: /add all messages first, then /flush each session."""
|
|
for sess in sessions:
|
|
payload = {"session_id": sess["session_id"], "messages": sess["messages"]}
|
|
status, _ = client.post("/api/v1/memory/add", payload, quiet=quiet)
|
|
assert status == 200, f"add failed for {sess['session_id']}: {status}"
|
|
|
|
for sess in sessions:
|
|
status, _ = client.post(
|
|
"/api/v1/memory/flush",
|
|
{"session_id": sess["session_id"]},
|
|
quiet=quiet,
|
|
)
|
|
assert status == 200, f"flush failed for {sess['session_id']}: {status}"
|
|
|
|
|
|
def wait_pipeline(seconds: int = 180) -> None:
|
|
"""Wait for cascade + OME pipeline to settle after flush."""
|
|
print(f" waiting {seconds}s for pipeline to settle...")
|
|
time.sleep(seconds) # tz-noqa — wall-clock delay, not a datetime
|
|
print(" pipeline wait done")
|
|
|
|
|
|
def trigger_reflection(
|
|
client: EverosClient,
|
|
*,
|
|
timeout: float = 120.0,
|
|
) -> None:
|
|
"""Trigger Reflection via HTTP endpoint on the running server."""
|
|
print(" triggering reflection via HTTP...")
|
|
status, resp = client.post(
|
|
"/api/v1/ome/trigger",
|
|
{"name": "reflect_episodes", "timeout": timeout, "force": True},
|
|
quiet=True,
|
|
)
|
|
result_status = resp.get("status", "unknown") if isinstance(resp, dict) else "error"
|
|
print(f" trigger response: status={result_status}")
|
|
if status != 200 or result_status != "ok":
|
|
raise RuntimeError(f"reflection trigger failed: HTTP {status}, {resp}")
|
|
|
|
|
|
def search_episodes(
|
|
client: EverosClient,
|
|
query: str,
|
|
owner_id: str,
|
|
*,
|
|
method: str = "hybrid",
|
|
top_k: int = 10,
|
|
) -> dict[str, Any]:
|
|
"""Run a memory search and return the ``data`` payload."""
|
|
payload = {
|
|
"query": query,
|
|
"method": method,
|
|
"top_k": top_k,
|
|
"user_id": owner_id,
|
|
}
|
|
status, resp = client.post("/api/v1/memory/search", payload, quiet=True)
|
|
assert status == 200, f"search failed: {status}"
|
|
return resp.get("data", {})
|
|
|
|
|
|
def answer_and_judge(
|
|
query: str,
|
|
search_data: dict[str, Any],
|
|
golden_answer: str,
|
|
*,
|
|
speaker_a: str,
|
|
speaker_b: str,
|
|
llm_client: LLMClientPool,
|
|
llm_model: str,
|
|
) -> dict[str, Any]:
|
|
"""Generate an answer from search results and judge correctness.
|
|
|
|
Returns a dict with ``answer``, ``judge_score`` (0 or 1), and
|
|
``episodes_count``.
|
|
"""
|
|
context = _build_context(
|
|
search_data.get("episodes", []),
|
|
search_data.get("profiles", []),
|
|
speaker_a,
|
|
speaker_b,
|
|
)
|
|
prompt = ANSWER_PROMPT.format(context=context, question=query)
|
|
try:
|
|
resp = llm_client.chat.completions.create(
|
|
model=llm_model,
|
|
messages=[{"role": "user", "content": prompt}],
|
|
temperature=0.0,
|
|
)
|
|
answer = _extract_final_answer(resp.choices[0].message.content or "")
|
|
except Exception as e:
|
|
answer = f"[error: {e}]"
|
|
|
|
try:
|
|
judge_resp = llm_client.chat.completions.create(
|
|
model=llm_model,
|
|
messages=[
|
|
{"role": "system", "content": JUDGE_SYSTEM_PROMPT},
|
|
{
|
|
"role": "user",
|
|
"content": JUDGE_USER_PROMPT.format(
|
|
question=query,
|
|
golden_answer=golden_answer,
|
|
generated_answer=answer,
|
|
),
|
|
},
|
|
],
|
|
temperature=0.0,
|
|
)
|
|
judge_text = judge_resp.choices[0].message.content or ""
|
|
raw_json = _extract_json(judge_text)
|
|
if raw_json:
|
|
parsed = json.loads(raw_json)
|
|
is_correct = parsed.get("label", "").upper() == "CORRECT"
|
|
else:
|
|
is_correct = False
|
|
except Exception:
|
|
logger.warning("judge evaluation failed", exc_info=True)
|
|
is_correct = False
|
|
|
|
return {
|
|
"answer": answer,
|
|
"judge_score": 1 if is_correct else 0,
|
|
"episodes_count": len(search_data.get("episodes", [])),
|
|
}
|
|
|
|
|
|
def compute_fact_coverage(text: str, facts: list[str]) -> float:
|
|
"""Compute fraction of golden facts found (case-insensitive substring)."""
|
|
text_lower = text.lower()
|
|
hits = sum(1 for f in facts if f.lower() in text_lower)
|
|
return hits / len(facts) if facts else 0.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TCResult — lightweight per-test-case assertion tracker
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TCResult:
|
|
"""Accumulate pass/fail checks for a single test case."""
|
|
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
self.passed: list[str] = []
|
|
self.failed: list[str] = []
|
|
|
|
def check(self, condition: bool, description: str) -> None:
|
|
(self.passed if condition else self.failed).append(description)
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return len(self.failed) == 0
|
|
|
|
def print_summary(self) -> None:
|
|
status = "PASS" if self.ok else "FAIL"
|
|
print(f"\n {self.name}: {status}")
|
|
for p in self.passed:
|
|
print(f" [ok] {p}")
|
|
for f in self.failed:
|
|
print(f" [FAIL] {f}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test cases (TC1-TC8) — INIT + UPDATE per storyline
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def tc1_adoption_init(client: EverosClient) -> TCResult:
|
|
tc = TCResult("TC1: Adoption INIT")
|
|
print_section("TC1: Adoption INIT (conv0, sessions 2,8,13,17)")
|
|
owner = _owner_id("caroline", 0)
|
|
data = load_locomo()
|
|
sessions = parse_sessions(data[0], ADOPTION_INIT_SESSIONS, 0)
|
|
add_and_flush(client, sessions)
|
|
wait_pipeline()
|
|
trigger_reflection(client)
|
|
# Positive: reflection report was written (deprecation completed)
|
|
reports = count_reflection_reports(owner)
|
|
tc.check(reports >= 1, f"reflection report created ({reports} found)")
|
|
dep_count = count_deprecated_episodes(owner)
|
|
tc.check(dep_count >= 1, f"source episodes deprecated ({dep_count} source_count)")
|
|
# Search: merged episode visible, deprecated filtered out
|
|
result = search_episodes(client, QUERIES["adoption"], owner)
|
|
episodes = result.get("episodes", [])
|
|
tc.check(len(episodes) > 0, "search returns episodes")
|
|
merged = [e for e in episodes if e.get("session_id") is None]
|
|
tc.check(len(merged) >= 1, "merged episode exists (session_id=None)")
|
|
print_episode_locations(owner, episodes)
|
|
tc.print_summary()
|
|
return tc
|
|
|
|
|
|
def tc2_adoption_update(client: EverosClient) -> TCResult:
|
|
tc = TCResult("TC2: Adoption UPDATE")
|
|
print_section("TC2: Adoption UPDATE (conv0, session 19)")
|
|
owner = _owner_id("caroline", 0)
|
|
reports_before = count_reflection_reports(owner)
|
|
data = load_locomo()
|
|
sessions = parse_sessions(data[0], ADOPTION_UPDATE_SESSIONS, 0)
|
|
add_and_flush(client, sessions)
|
|
wait_pipeline()
|
|
trigger_reflection(client)
|
|
reports_after = count_reflection_reports(owner)
|
|
tc.check(
|
|
reports_after > reports_before,
|
|
f"report count up ({reports_before}->{reports_after})",
|
|
)
|
|
result = search_episodes(client, QUERIES["adoption"], owner)
|
|
merged = [e for e in result.get("episodes", []) if e.get("session_id") is None]
|
|
tc.check(len(merged) >= 1, "merged episode exists after update")
|
|
print_episode_locations(owner, result.get("episodes", []))
|
|
tc.print_summary()
|
|
return tc
|
|
|
|
|
|
def tc3_lgbtq_init(client: EverosClient) -> TCResult:
|
|
tc = TCResult("TC3: LGBTQ+Conflict INIT")
|
|
print_section("TC3: LGBTQ+Conflict INIT (conv0, sessions 1,3,5,12)")
|
|
owner = _owner_id("caroline", 0)
|
|
data = load_locomo()
|
|
sessions = parse_sessions(data[0], LGBTQ_INIT_SESSIONS, 0)
|
|
add_and_flush(client, sessions)
|
|
wait_pipeline()
|
|
trigger_reflection(client)
|
|
reports = count_reflection_reports(owner)
|
|
tc.check(reports >= 1, f"reflection report(s) exist ({reports})")
|
|
result = search_episodes(client, QUERIES["lgbtq"], owner)
|
|
merged = [e for e in result.get("episodes", []) if e.get("session_id") is None]
|
|
tc.check(len(merged) >= 1, "merged episode exists")
|
|
print_episode_locations(owner, result.get("episodes", []))
|
|
tc.print_summary()
|
|
return tc
|
|
|
|
|
|
def tc4_lgbtq_update(client: EverosClient) -> TCResult:
|
|
tc = TCResult("TC4: LGBTQ+Conflict UPDATE")
|
|
print_section("TC4: LGBTQ+Conflict UPDATE (conv0, session 14)")
|
|
owner = _owner_id("caroline", 0)
|
|
reports_before = count_reflection_reports(owner)
|
|
data = load_locomo()
|
|
sessions = parse_sessions(data[0], LGBTQ_UPDATE_SESSIONS, 0)
|
|
add_and_flush(client, sessions)
|
|
wait_pipeline()
|
|
trigger_reflection(client)
|
|
reports_after = count_reflection_reports(owner)
|
|
tc.check(
|
|
reports_after > reports_before,
|
|
f"report count up ({reports_before}->{reports_after})",
|
|
)
|
|
result = search_episodes(client, QUERIES["lgbtq"], owner)
|
|
merged = [e for e in result.get("episodes", []) if e.get("session_id") is None]
|
|
tc.check(len(merged) >= 1, "merged episode exists after update")
|
|
print_episode_locations(owner, result.get("episodes", []))
|
|
tc.print_summary()
|
|
return tc
|
|
|
|
|
|
def tc5_pet_init(client: EverosClient) -> TCResult:
|
|
tc = TCResult("TC5: Pet Count INIT")
|
|
print_section("TC5: Pet Count INIT (conv5, sessions 1,5,12,24)")
|
|
owner = _owner_id("andrew", 5)
|
|
data = load_locomo()
|
|
sessions = parse_sessions(data[5], PET_INIT_SESSIONS, 5)
|
|
add_and_flush(client, sessions)
|
|
wait_pipeline()
|
|
trigger_reflection(client)
|
|
reports = count_reflection_reports(owner)
|
|
tc.check(reports >= 1, f"reflection report created ({reports})")
|
|
result = search_episodes(client, QUERIES["pet"], owner)
|
|
merged = [e for e in result.get("episodes", []) if e.get("session_id") is None]
|
|
tc.check(len(merged) >= 1, "merged episode exists")
|
|
print_episode_locations(owner, result.get("episodes", []))
|
|
tc.print_summary()
|
|
return tc
|
|
|
|
|
|
def tc6_pet_update(client: EverosClient) -> TCResult:
|
|
tc = TCResult("TC6: Pet Count UPDATE")
|
|
print_section("TC6: Pet Count UPDATE (conv5, sessions 27,28)")
|
|
owner = _owner_id("andrew", 5)
|
|
reports_before = count_reflection_reports(owner)
|
|
data = load_locomo()
|
|
sessions = parse_sessions(data[5], PET_UPDATE_SESSIONS, 5)
|
|
add_and_flush(client, sessions)
|
|
wait_pipeline()
|
|
trigger_reflection(client)
|
|
reports_after = count_reflection_reports(owner)
|
|
tc.check(
|
|
reports_after > reports_before,
|
|
f"report count up ({reports_before}->{reports_after})",
|
|
)
|
|
result = search_episodes(client, QUERIES["pet"], owner)
|
|
merged = [e for e in result.get("episodes", []) if e.get("session_id") is None]
|
|
tc.check(len(merged) >= 1, "merged episode exists after update")
|
|
print_episode_locations(owner, result.get("episodes", []))
|
|
tc.print_summary()
|
|
return tc
|
|
|
|
|
|
def tc7_health_init(client: EverosClient) -> TCResult:
|
|
tc = TCResult("TC7: Health Relapse INIT")
|
|
print_section("TC7: Health INIT (conv8, sessions 2,4,8,10,13,14)")
|
|
owner = _owner_id("sam", 8)
|
|
data = load_locomo()
|
|
sessions = parse_sessions(data[8], HEALTH_INIT_SESSIONS, 8)
|
|
add_and_flush(client, sessions)
|
|
wait_pipeline()
|
|
trigger_reflection(client)
|
|
reports = count_reflection_reports(owner)
|
|
tc.check(reports >= 1, f"reflection report created ({reports})")
|
|
result = search_episodes(client, QUERIES["health"], owner)
|
|
merged = [e for e in result.get("episodes", []) if e.get("session_id") is None]
|
|
tc.check(len(merged) >= 1, "merged episode exists")
|
|
print_episode_locations(owner, result.get("episodes", []))
|
|
tc.print_summary()
|
|
return tc
|
|
|
|
|
|
def tc8_health_update(client: EverosClient) -> TCResult:
|
|
tc = TCResult("TC8: Health Relapse UPDATE")
|
|
print_section("TC8: Health UPDATE (conv8, sessions 16,20)")
|
|
owner = _owner_id("sam", 8)
|
|
reports_before = count_reflection_reports(owner)
|
|
data = load_locomo()
|
|
sessions = parse_sessions(data[8], HEALTH_UPDATE_SESSIONS, 8)
|
|
add_and_flush(client, sessions)
|
|
wait_pipeline()
|
|
trigger_reflection(client)
|
|
reports_after = count_reflection_reports(owner)
|
|
tc.check(
|
|
reports_after > reports_before,
|
|
f"report count up ({reports_before}->{reports_after})",
|
|
)
|
|
result = search_episodes(client, QUERIES["health"], owner)
|
|
merged = [e for e in result.get("episodes", []) if e.get("session_id") is None]
|
|
tc.check(len(merged) >= 1, "merged episode exists after update")
|
|
print_episode_locations(owner, result.get("episodes", []))
|
|
tc.print_summary()
|
|
return tc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test cases (TC9, TC11-TC14) — cross-cutting validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def tc9_search_visibility(client: EverosClient) -> TCResult:
|
|
tc = TCResult("TC9: Search Visibility")
|
|
print_section("TC9: Search Visibility")
|
|
checks = [
|
|
(QUERIES["adoption"], _owner_id("caroline", 0)),
|
|
(QUERIES["lgbtq"], _owner_id("caroline", 0)),
|
|
(QUERIES["pet"], _owner_id("andrew", 5)),
|
|
(QUERIES["health"], _owner_id("sam", 8)),
|
|
]
|
|
for query, owner in checks:
|
|
# Positive: deprecation actually happened
|
|
reports = count_reflection_reports(owner)
|
|
tc.check(reports >= 1, f"reports exist for {owner} ({reports})")
|
|
# Search: merged visible, deprecated filtered
|
|
data = search_episodes(client, query, owner)
|
|
episodes = data.get("episodes", [])
|
|
merged = [e for e in episodes if e.get("session_id") is None]
|
|
tc.check(len(merged) >= 1, f"merged present for '{query[:40]}...'")
|
|
tc.print_summary()
|
|
return tc
|
|
|
|
|
|
def tc11_idempotency(client: EverosClient) -> TCResult:
|
|
tc = TCResult("TC11: Idempotency")
|
|
print_section("TC11: Idempotency")
|
|
owner = _owner_id("caroline", 0)
|
|
before = search_episodes(client, QUERIES["adoption"], owner)
|
|
merged_before = [
|
|
e for e in before.get("episodes", []) if e.get("session_id") is None
|
|
]
|
|
count_before = len(merged_before)
|
|
trigger_reflection(client)
|
|
after = search_episodes(client, QUERIES["adoption"], owner)
|
|
merged_after = [e for e in after.get("episodes", []) if e.get("session_id") is None]
|
|
tc.check(
|
|
len(merged_after) == count_before,
|
|
f"merged count unchanged ({count_before} -> {len(merged_after)})",
|
|
)
|
|
if merged_before and merged_after:
|
|
tc.check(
|
|
merged_before[0].get("id") == merged_after[0].get("id"),
|
|
"same merged episode ID (no duplicate)",
|
|
)
|
|
tc.print_summary()
|
|
return tc
|
|
|
|
|
|
def tc12_atomic_facts(client: EverosClient) -> TCResult:
|
|
tc = TCResult("TC12: Atomic Facts Re-extraction")
|
|
print_section("TC12: Atomic Facts Re-extraction")
|
|
owner = _owner_id("caroline", 0)
|
|
result = search_episodes(client, QUERIES["adoption"], owner)
|
|
merged = [e for e in result.get("episodes", []) if e.get("session_id") is None]
|
|
tc.check(len(merged) >= 1, "merged episode exists")
|
|
if merged:
|
|
facts = merged[0].get("atomic_facts", [])
|
|
tc.check(len(facts) > 0, f"merged has atomic facts ({len(facts)} found)")
|
|
tc.print_summary()
|
|
return tc
|
|
|
|
|
|
def tc13_topic_isolation(client: EverosClient) -> TCResult:
|
|
tc = TCResult("TC13: Cross-topic Isolation")
|
|
print_section("TC13: Cross-topic Isolation")
|
|
owner = _owner_id("caroline", 0)
|
|
adoption = search_episodes(client, QUERIES["adoption"], owner)
|
|
lgbtq = search_episodes(client, QUERIES["lgbtq"], owner)
|
|
a_merged = [e for e in adoption.get("episodes", []) if e.get("session_id") is None]
|
|
l_merged = [e for e in lgbtq.get("episodes", []) if e.get("session_id") is None]
|
|
tc.check(len(a_merged) >= 1, "adoption has merged episode")
|
|
tc.check(len(l_merged) >= 1, "lgbtq has merged episode")
|
|
if a_merged and l_merged:
|
|
tc.check(
|
|
a_merged[0].get("id") != l_merged[0].get("id"),
|
|
"different merged episode IDs",
|
|
)
|
|
a_text = a_merged[0].get("episode", "").lower()
|
|
l_text = l_merged[0].get("episode", "").lower()
|
|
tc.check(
|
|
"discriminat" not in a_text and "hike" not in a_text,
|
|
"adoption text has no discrimination content",
|
|
)
|
|
tc.check(
|
|
"agenc" not in l_text and "adoption council" not in l_text,
|
|
"lgbtq text has no adoption process content",
|
|
)
|
|
tc.print_summary()
|
|
return tc
|
|
|
|
|
|
def tc14_answer_judge(
|
|
client: EverosClient,
|
|
llm_client: LLMClientPool,
|
|
llm_model: str,
|
|
) -> TCResult:
|
|
tc = TCResult("TC14: Answer+Judge Quality")
|
|
print_section("TC14: Answer+Judge Quality Comparison")
|
|
|
|
golden_answers = {
|
|
"adoption": (
|
|
"Caroline researched adoption agencies, attended an adoption council "
|
|
"meeting, applied to multiple agencies, contacted her mentor for advice, "
|
|
"and passed the adoption agency interviews."
|
|
),
|
|
"lgbtq": (
|
|
"Caroline dealt with discrimination by attending LGBTQ support groups, "
|
|
"speaking at her school, participating in a Pride parade. When she "
|
|
"encountered discrimination on a hike from religious conservatives, "
|
|
"she later wrote an apology letter to reconcile."
|
|
),
|
|
"pet": (
|
|
"Andrew has three dogs: Toby, Buddy, and Scout. He initially had no "
|
|
"pets, then adopted Toby, followed by Buddy from a shelter, and most "
|
|
"recently Scout."
|
|
),
|
|
"health": (
|
|
"Sam's journey has been non-linear. After a doctor warned about his "
|
|
"weight, he started dieting with good results. But he relapsed by "
|
|
"buying unhealthy snacks, then had a gastritis emergency. He recovered "
|
|
"to become a Weight Watchers coach, but later struggled again."
|
|
),
|
|
}
|
|
owner_map = {
|
|
"adoption": (_owner_id("caroline", 0), "Caroline", "Melanie"),
|
|
"lgbtq": (_owner_id("caroline", 0), "Caroline", "Melanie"),
|
|
"pet": (_owner_id("andrew", 5), "Audrey", "Andrew"),
|
|
"health": (_owner_id("sam", 8), "Evan", "Sam"),
|
|
}
|
|
|
|
total_score = 0
|
|
for topic, query in QUERIES.items():
|
|
owner, speaker_a, speaker_b = owner_map[topic]
|
|
data = search_episodes(client, query, owner)
|
|
result = answer_and_judge(
|
|
query,
|
|
data,
|
|
golden_answers[topic],
|
|
speaker_a=speaker_a,
|
|
speaker_b=speaker_b,
|
|
llm_client=llm_client,
|
|
llm_model=llm_model,
|
|
)
|
|
score = result["judge_score"]
|
|
total_score += score
|
|
merged = [e for e in data.get("episodes", []) if e.get("session_id") is None]
|
|
merged_text = merged[0].get("episode", "") if merged else ""
|
|
coverage = compute_fact_coverage(merged_text, GOLDEN_FACTS[topic])
|
|
status = "CORRECT" if score else "WRONG"
|
|
print(f" {topic}: {status} | fact_coverage={coverage:.0%}")
|
|
print(f" answer: {result['answer'][:120]}...")
|
|
tc.check(score == 1, f"{topic} answered correctly")
|
|
|
|
print(f"\n Overall: {total_score}/{len(QUERIES)}")
|
|
tc.print_summary()
|
|
return tc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
TC_REGISTRY: dict[int, tuple[str, Any]] = {
|
|
1: ("Adoption INIT", lambda c, **kw: tc1_adoption_init(c)),
|
|
2: ("Adoption UPDATE", lambda c, **kw: tc2_adoption_update(c)),
|
|
3: ("LGBTQ INIT", lambda c, **kw: tc3_lgbtq_init(c)),
|
|
4: ("LGBTQ UPDATE", lambda c, **kw: tc4_lgbtq_update(c)),
|
|
5: ("Pet INIT", lambda c, **kw: tc5_pet_init(c)),
|
|
6: ("Pet UPDATE", lambda c, **kw: tc6_pet_update(c)),
|
|
7: ("Health INIT", lambda c, **kw: tc7_health_init(c)),
|
|
8: ("Health UPDATE", lambda c, **kw: tc8_health_update(c)),
|
|
9: ("Search Visibility", lambda c, **kw: tc9_search_visibility(c)),
|
|
# TC10 removed: was a duplicate of TC9 visibility checks.
|
|
11: ("Idempotency", lambda c, **kw: tc11_idempotency(c)),
|
|
12: ("Atomic Facts", lambda c, **kw: tc12_atomic_facts(c)),
|
|
13: ("Topic Isolation", lambda c, **kw: tc13_topic_isolation(c)),
|
|
14: (
|
|
"Answer+Judge",
|
|
lambda c, **kw: tc14_answer_judge(c, kw["llm_client"], kw["llm_model"]),
|
|
),
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
import os
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
parser = argparse.ArgumentParser(description="Reflection E2E Test")
|
|
parser.add_argument(
|
|
"--tc",
|
|
type=str,
|
|
default=None,
|
|
help="Comma-separated TC numbers (e.g. '1,2,14'). Default: all.",
|
|
)
|
|
parser.add_argument("--base-url", default="http://localhost:8000")
|
|
parser.add_argument("--llm-model", default=None)
|
|
parser.add_argument("--verbose", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
logging.basicConfig(
|
|
level=logging.DEBUG if args.verbose else logging.WARNING,
|
|
format="%(levelname)s %(name)s: %(message)s",
|
|
)
|
|
|
|
tc_ids = (
|
|
[int(x.strip()) for x in args.tc.split(",")]
|
|
if args.tc
|
|
else sorted(TC_REGISTRY.keys())
|
|
)
|
|
|
|
client = EverosClient(base_url=args.base_url)
|
|
llm_model = args.llm_model or os.getenv("EVEROS_LLM__MODEL", "openai/gpt-4.1-mini")
|
|
api_key = os.getenv("EVEROS_LLM__API_KEY", "")
|
|
base_url = os.getenv("EVEROS_LLM__BASE_URL", "https://openrouter.ai/api/v1")
|
|
llm_client = LLMClientPool(api_keys=[api_key], base_url=base_url)
|
|
|
|
print_section("Reflection E2E Test")
|
|
print(f" TCs: {tc_ids}")
|
|
print(f" Server: {args.base_url}")
|
|
print(f" LLM: {llm_model}")
|
|
|
|
results: list[TCResult] = []
|
|
for tc_id in tc_ids:
|
|
if tc_id not in TC_REGISTRY:
|
|
print(f" WARNING: TC{tc_id} not found, skipping")
|
|
continue
|
|
name, func = TC_REGISTRY[tc_id]
|
|
try:
|
|
r = func(client, llm_client=llm_client, llm_model=llm_model)
|
|
results.append(r)
|
|
except Exception as e:
|
|
print(f"\n TC{tc_id} ({name}) CRASHED: {e}")
|
|
tc = TCResult(f"TC{tc_id}: {name}")
|
|
tc.check(False, f"crashed: {e}")
|
|
results.append(tc)
|
|
|
|
print_section("SUMMARY")
|
|
passed = sum(1 for r in results if r.ok)
|
|
for r in results:
|
|
status = "PASS" if r.ok else "FAIL"
|
|
checks = f"{len(r.passed)}/{len(r.passed) + len(r.failed)}"
|
|
print(f" {status} {r.name} ({checks} checks)")
|
|
print(f"\n Total: {passed}/{len(results)} TCs passed")
|
|
sys.exit(0 if passed == len(results) else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|