e6f98aaa6d
- Add config_snapshot.py with save/list/get/compare functions - Auto-save snapshots on eval creation and plan submission - Implement snapshot query APIs (list, get single) - Implement snapshot comparison API (diff two snapshots) - Add 8 unit tests and 7 integration tests Snapshots track config changes over time (created/plan_submitted/config_updated). All 813 tests passing.
110 lines
3.1 KiB
Python
110 lines
3.1 KiB
Python
"""Config snapshot management for intelligent evaluations (配置快照管理).
|
|
|
|
Automatically saves config snapshots when:
|
|
- Eval is created (snapshot_type: created)
|
|
- Plan is submitted (snapshot_type: plan_submitted)
|
|
- Config is updated (snapshot_type: config_updated)
|
|
"""
|
|
|
|
from typing import Any, Optional
|
|
|
|
from sqlmodel import Session, select
|
|
|
|
from agenteval.storage.db import IntelligentEvalConfigSnapshotDB, IntelligentEvalDB
|
|
|
|
|
|
def save_snapshot(
|
|
eval_db: IntelligentEvalDB,
|
|
snapshot_type: str,
|
|
created_by: str,
|
|
session: Session,
|
|
) -> IntelligentEvalConfigSnapshotDB:
|
|
"""Save a config snapshot.
|
|
|
|
Args:
|
|
eval_db: Evaluation database record
|
|
snapshot_type: Type of snapshot (created / plan_submitted / config_updated)
|
|
created_by: Who created the snapshot (user / openclaw)
|
|
session: Database session
|
|
|
|
Returns:
|
|
Created snapshot
|
|
"""
|
|
snapshot = IntelligentEvalConfigSnapshotDB(
|
|
eval_id=eval_db.id,
|
|
snapshot_type=snapshot_type,
|
|
goal=eval_db.goal,
|
|
seeds=eval_db.seeds,
|
|
intent=eval_db.intent,
|
|
role_description=eval_db.role_description,
|
|
time_window_hours=eval_db.time_window_hours,
|
|
plan=eval_db.plan,
|
|
created_by=created_by,
|
|
)
|
|
session.add(snapshot)
|
|
session.commit()
|
|
session.refresh(snapshot)
|
|
return snapshot
|
|
|
|
|
|
def list_snapshots(eval_id: str, session: Session) -> list[IntelligentEvalConfigSnapshotDB]:
|
|
"""List all snapshots for an evaluation.
|
|
|
|
Returns:
|
|
List of snapshots, ordered by created_at descending (newest first)
|
|
"""
|
|
snapshots = session.exec(
|
|
select(IntelligentEvalConfigSnapshotDB)
|
|
.where(IntelligentEvalConfigSnapshotDB.eval_id == eval_id)
|
|
.order_by(IntelligentEvalConfigSnapshotDB.created_at.desc())
|
|
).all()
|
|
return list(snapshots)
|
|
|
|
|
|
def get_snapshot(snapshot_id: str, session: Session) -> Optional[IntelligentEvalConfigSnapshotDB]:
|
|
"""Get a single snapshot by ID.
|
|
|
|
Returns:
|
|
Snapshot, or None if not found
|
|
"""
|
|
return session.get(IntelligentEvalConfigSnapshotDB, snapshot_id)
|
|
|
|
|
|
def compare_snapshots(
|
|
snapshot1: IntelligentEvalConfigSnapshotDB,
|
|
snapshot2: IntelligentEvalConfigSnapshotDB,
|
|
) -> dict[str, Any]:
|
|
"""Compare two snapshots and return differences.
|
|
|
|
Returns:
|
|
Dict with differences, format:
|
|
{
|
|
"field_name": {
|
|
"old": value1,
|
|
"new": value2,
|
|
}
|
|
}
|
|
"""
|
|
diffs = {}
|
|
|
|
# Compare simple fields
|
|
fields = ["goal", "intent", "role_description", "time_window_hours"]
|
|
for field in fields:
|
|
val1 = getattr(snapshot1, field)
|
|
val2 = getattr(snapshot2, field)
|
|
if val1 != val2:
|
|
diffs[field] = {"old": val1, "new": val2}
|
|
|
|
# Compare JSON fields
|
|
seeds1 = snapshot1.get_seeds()
|
|
seeds2 = snapshot2.get_seeds()
|
|
if seeds1 != seeds2:
|
|
diffs["seeds"] = {"old": seeds1, "new": seeds2}
|
|
|
|
plan1 = snapshot1.get_plan()
|
|
plan2 = snapshot2.get_plan()
|
|
if plan1 != plan2:
|
|
diffs["plan"] = {"old": plan1, "new": plan2}
|
|
|
|
return diffs
|