AgentEvalTool/backend/agenteval/intelligent_eval/decision.py
sinohqb 975ed7a6ff
Some checks failed
CI / test (push) Failing after 4m21s
refactor(intelligent-eval): converge scheduling domain (S1) + stuck-task settlement (S4)
P1 deepening (issue #7):

S1:  is the single source of truth for time-slot parsing,
   slot-due checks, session deficit, priority, attention reason, and
   high-severity detection.  and  now delegate
   their internal helpers to  while keeping the same signatures
   (tests continue to pass via the thin wrappers).

S4:  encapsulates the stuck-cron
   task settlement (fail current task + enqueue retry).
   calls it instead of the previous runtime import of .

No observable behaviour change — 873 passed + 5 xfailed unchanged.
2026-08-13 10:04:24 +08:00

209 lines
6.6 KiB
Python

"""Decision logic for intelligent eval workers (决策逻辑).
Worker analyzes current situation and decides what to do:
- execute_session: Execute a new session
- wait: Wait for next tick
- start_analysis: Start analysis (all sessions completed)
"""
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional
from sqlmodel import Session, select
from agenteval.intelligent_eval.models import IntelligentEvalStatus
from agenteval.storage.db import (
IntelligentEvalDB,
IntelligentEvalSessionDB,
as_utc,
utc_now,
)
class DecisionType(str, Enum):
"""Decision types for worker."""
EXECUTE_SESSION = "execute_session"
WAIT = "wait"
START_ANALYSIS = "start_analysis"
class Decision:
"""A decision made by a worker."""
def __init__(self, decision_type: DecisionType, reason: str, context: dict):
self.decision_type = decision_type
self.reason = reason
self.context = context
def to_dict(self) -> dict:
return {
"decision_type": self.decision_type.value,
"reason": self.reason,
"context": self.context,
}
def _parse_time_slot(time_slot: str) -> Optional[tuple[int, int]]:
"""Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.parse_time_slot`."""
from agenteval.intelligent_eval.domain import parse_time_slot as _impl
return _impl(time_slot)
def _get_current_slot(time_distribution: list[dict], current_offset: timedelta) -> Optional[dict]:
"""Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.get_current_slot`."""
from agenteval.intelligent_eval.domain import get_current_slot as _impl
return _impl(time_distribution, current_offset)
def _count_sessions_in_slot(
eval_id: str, slot: dict, eval_started_at: datetime, session: Session
) -> int:
"""Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.count_sessions_in_slot`."""
from agenteval.intelligent_eval.domain import count_sessions_in_slot as _impl
return _impl(eval_id, slot, eval_started_at, session)
def _has_high_severity_issues(eval_id: str, session: Session) -> bool:
"""Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.has_high_severity_issues`."""
from agenteval.intelligent_eval.domain import has_high_severity_issues as _impl
return _impl(eval_id, session)
def make_decision(eval_db: IntelligentEvalDB, session: Session) -> Decision:
"""Make a decision based on current evaluation state.
Args:
eval_db: Evaluation database record
session: Database session
Returns:
Decision object
"""
# Check if eval is still executing
if eval_db.status != IntelligentEvalStatus.EXECUTING.value:
return Decision(
DecisionType.WAIT,
f"评估状态为 {eval_db.status},不在执行中",
{"status": eval_db.status},
)
# Check if eval has plan and started_at
if not eval_db.plan or not eval_db.started_at:
return Decision(
DecisionType.WAIT,
"评估缺少计划或未开始",
{"has_plan": bool(eval_db.plan), "has_started_at": bool(eval_db.started_at)},
)
plan = eval_db.get_plan()
time_distribution = plan.get("time_distribution", [])
estimated_sessions = plan.get("estimated_sessions", 0)
# Calculate current offset
current_offset = utc_now() - as_utc(eval_db.started_at)
current_hours = current_offset.total_seconds() / 3600
# Get all sessions
all_sessions = session.exec(
select(IntelligentEvalSessionDB).where(IntelligentEvalSessionDB.eval_id == eval_db.id)
).all()
completed_sessions = [s for s in all_sessions if s.status == "completed"]
# Check if all sessions completed and estimated reached
if len(completed_sessions) >= estimated_sessions:
return Decision(
DecisionType.START_ANALYSIS,
f"所有 {estimated_sessions} 个会话已完成,开始分析",
{
"completed_sessions": len(completed_sessions),
"estimated_sessions": estimated_sessions,
},
)
# Get current time slot
current_slot = _get_current_slot(time_distribution, current_offset)
if current_slot is None:
return Decision(
DecisionType.WAIT,
f"当前时间偏移 {current_hours:.1f}h 不在任何时段内",
{"current_offset_hours": current_hours},
)
# Check if current slot has deficit
slot_name = current_slot.get("time_slot", "")
expected_sessions = current_slot.get("sessions", 0)
current_sessions = _count_sessions_in_slot(eval_db.id, current_slot, eval_db.started_at, session)
deficit = expected_sessions - current_sessions
if deficit > 0:
return Decision(
DecisionType.EXECUTE_SESSION,
f"时段 {slot_name} 欠账 {deficit} 个会话",
{
"current_slot": slot_name,
"expected_sessions": expected_sessions,
"current_sessions": current_sessions,
"deficit": deficit,
},
)
# Check if any high severity issues found
if _has_high_severity_issues(eval_db.id, session):
return Decision(
DecisionType.EXECUTE_SESSION,
"发现高严重度问题,需要深入挖掘",
{"has_high_severity": True},
)
# No deficit, no high severity issues, wait
return Decision(
DecisionType.WAIT,
f"时段 {slot_name} 无欠账,等待下一时段",
{
"current_slot": slot_name,
"expected_sessions": expected_sessions,
"current_sessions": current_sessions,
},
)
def is_eval_completed(eval_db: IntelligentEvalDB, session: Session) -> bool:
"""Check if evaluation is completed.
An eval is completed when:
1. All sessions are completed
2. Report is submitted
Returns:
True if eval is completed
"""
if eval_db.status != IntelligentEvalStatus.EXECUTING.value:
return False
if not eval_db.plan:
return False
plan = eval_db.get_plan()
estimated_sessions = plan.get("estimated_sessions", 0)
# Check if all sessions completed
all_sessions = session.exec(
select(IntelligentEvalSessionDB).where(IntelligentEvalSessionDB.eval_id == eval_db.id)
).all()
completed_sessions = [s for s in all_sessions if s.status == "completed"]
if len(completed_sessions) < estimated_sessions:
return False
# Check if report submitted
if not eval_db.report:
return False
return True