200 lines
7.0 KiB
Python
200 lines
7.0 KiB
Python
"""Read projections for intelligent evaluations.
|
|
|
|
The lifecycle module owns writes and state transitions. This module is the
|
|
read seam used by HTTP adapters: it turns domain entities into stable,
|
|
purpose-specific projections and keeps session-count/detail assembly out of
|
|
routers. Message bodies are intentionally not part of the detail projection;
|
|
transcripts remain available through the dedicated messages endpoint.
|
|
"""
|
|
|
|
from collections.abc import Sequence
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, Field
|
|
from sqlmodel import Session
|
|
|
|
from agenteval.intelligent_eval.models import (
|
|
IntelligentEval,
|
|
IntelligentEvalMessage,
|
|
IntelligentEvalSession,
|
|
IntelligentEvalSessionStatus,
|
|
IntelligentEvalStatus,
|
|
)
|
|
from agenteval.intelligent_eval.repository import (
|
|
IntelligentEvalMessageRepository,
|
|
IntelligentEvalRepository,
|
|
IntelligentEvalSessionRepository,
|
|
)
|
|
|
|
|
|
class IntelligentEvalProjection(BaseModel):
|
|
"""Fields shared by list and detail representations."""
|
|
|
|
id: str
|
|
name: str
|
|
target_id: str
|
|
status: IntelligentEvalStatus
|
|
goal: str
|
|
seeds: dict[str, Any] = Field(default_factory=dict)
|
|
intent: str
|
|
role_description: str
|
|
plan: dict[str, Any] | None = None
|
|
plan_feedback: str | None = None
|
|
time_window_hours: int
|
|
report: dict[str, Any] | None = None
|
|
created_at: datetime | None = None
|
|
updated_at: datetime | None = None
|
|
started_at: datetime | None = None
|
|
completed_at: datetime | None = None
|
|
|
|
|
|
class IntelligentEvalListItem(IntelligentEvalProjection):
|
|
"""Compact projection used by collection and mutation responses."""
|
|
|
|
session_count: int = 0
|
|
completed_sessions: int = 0
|
|
|
|
|
|
class IntelligentEvalSessionSummary(BaseModel):
|
|
"""Session metadata shown in an evaluation detail view.
|
|
|
|
This deliberately contains no chat messages. Callers that need a
|
|
transcript must use ``GET /{eval_id}/sessions/{session_id}/messages``.
|
|
"""
|
|
|
|
id: str
|
|
eval_id: str
|
|
target_id: str
|
|
persona: dict[str, Any] = Field(default_factory=dict)
|
|
goal: str
|
|
dimension: str | None = None
|
|
status: IntelligentEvalSessionStatus
|
|
verdict: dict[str, Any] | None = None
|
|
turn_count: int = 0
|
|
created_at: datetime | None = None
|
|
closed_at: datetime | None = None
|
|
|
|
|
|
class IntelligentEvalDetail(IntelligentEvalListItem):
|
|
"""Full evaluation projection, including session metadata only."""
|
|
|
|
sessions: list[IntelligentEvalSessionSummary] = Field(default_factory=list)
|
|
|
|
|
|
def _require_id(eval_obj: IntelligentEval) -> str:
|
|
if eval_obj.id is None:
|
|
raise ValueError("intelligent evaluation projection requires an id")
|
|
return eval_obj.id
|
|
|
|
|
|
def _base_fields(eval_obj: IntelligentEval) -> dict[str, Any]:
|
|
"""Copy domain data into the stable projection field set."""
|
|
|
|
return {
|
|
"id": _require_id(eval_obj),
|
|
"name": eval_obj.name,
|
|
"target_id": eval_obj.target_id,
|
|
"status": eval_obj.status,
|
|
"goal": eval_obj.goal,
|
|
"seeds": eval_obj.seeds,
|
|
"intent": eval_obj.intent,
|
|
"role_description": eval_obj.role_description,
|
|
"plan": eval_obj.plan,
|
|
"plan_feedback": eval_obj.plan_feedback,
|
|
"time_window_hours": eval_obj.time_window_hours,
|
|
"report": eval_obj.report,
|
|
"created_at": eval_obj.created_at,
|
|
"updated_at": eval_obj.updated_at,
|
|
"started_at": eval_obj.started_at,
|
|
"completed_at": eval_obj.completed_at,
|
|
}
|
|
|
|
|
|
def _session_summary(session_obj: IntelligentEvalSession) -> IntelligentEvalSessionSummary:
|
|
if session_obj.id is None:
|
|
raise ValueError("intelligent evaluation session projection requires an id")
|
|
return IntelligentEvalSessionSummary.model_validate(session_obj.model_dump())
|
|
|
|
|
|
def project_list_item(
|
|
eval_obj: IntelligentEval,
|
|
sessions: Sequence[IntelligentEvalSession],
|
|
) -> IntelligentEvalListItem:
|
|
"""Build a list projection from an evaluation and its session metadata."""
|
|
|
|
return IntelligentEvalListItem(
|
|
**_base_fields(eval_obj),
|
|
session_count=len(sessions),
|
|
completed_sessions=sum(1 for item in sessions if item.status is IntelligentEvalSessionStatus.COMPLETED),
|
|
)
|
|
|
|
|
|
def project_detail(
|
|
eval_obj: IntelligentEval,
|
|
sessions: Sequence[IntelligentEvalSession],
|
|
) -> IntelligentEvalDetail:
|
|
"""Build a detail projection without embedding transcript messages."""
|
|
|
|
item = project_list_item(eval_obj, sessions)
|
|
return IntelligentEvalDetail(
|
|
**item.model_dump(),
|
|
sessions=[_session_summary(item) for item in sessions],
|
|
)
|
|
|
|
|
|
class IntelligentEvalReadModel:
|
|
"""Read interface for stable intelligent-evaluation projections.
|
|
|
|
The current implementation uses the existing session repository. Query
|
|
batching can be added behind this seam without changing Router callers.
|
|
"""
|
|
|
|
def __init__(self, session: Session):
|
|
self._evals = IntelligentEvalRepository(session)
|
|
self._sessions = IntelligentEvalSessionRepository(session)
|
|
self._messages = IntelligentEvalMessageRepository(session)
|
|
|
|
def list_item(self, eval_obj: IntelligentEval) -> IntelligentEvalListItem:
|
|
return project_list_item(eval_obj, self._sessions.list_by_eval(_require_id(eval_obj)))
|
|
|
|
def list_items(self, evals: Sequence[IntelligentEval]) -> list[IntelligentEvalListItem]:
|
|
"""Project a collection after one batched session lookup."""
|
|
|
|
sessions_by_eval = self._sessions.list_by_evals([_require_id(item) for item in evals])
|
|
return [project_list_item(item, sessions_by_eval[item.id]) for item in evals if item.id is not None]
|
|
|
|
def detail(self, eval_obj: IntelligentEval) -> IntelligentEvalDetail:
|
|
return project_detail(eval_obj, self._sessions.list_by_eval(_require_id(eval_obj)))
|
|
|
|
def detail_by_id(self, eval_id: str) -> IntelligentEvalDetail | None:
|
|
"""Read one evaluation and its summaries from the same session snapshot."""
|
|
|
|
eval_obj = self._evals.get(eval_id)
|
|
return self.detail(eval_obj) if eval_obj is not None else None
|
|
|
|
def sessions_by_eval(self, eval_id: str) -> list[IntelligentEvalSession] | None:
|
|
"""Return session summaries, or ``None`` when the parent is unknown."""
|
|
|
|
if self._evals.get(eval_id) is None:
|
|
return None
|
|
return self._sessions.list_by_eval(eval_id)
|
|
|
|
def messages_by_session(self, eval_id: str, session_id: str) -> list[IntelligentEvalMessage] | None:
|
|
"""Read a transcript only when the session belongs to the evaluation."""
|
|
|
|
if self._evals.get(eval_id) is None:
|
|
return None
|
|
session_obj = self._sessions.get(session_id)
|
|
if session_obj is None or session_obj.eval_id != eval_id:
|
|
return None
|
|
return self._messages.list_by_session(session_id)
|
|
|
|
def report_by_eval(self, eval_id: str) -> tuple[str, dict[str, Any]] | None:
|
|
"""Return the report payload with its evaluation name for renderers."""
|
|
|
|
eval_obj = self._evals.get(eval_id)
|
|
if eval_obj is None or eval_obj.report is None:
|
|
return None
|
|
return eval_obj.name, eval_obj.report
|