1317552701
Introduce 智能评估 as an evaluation paradigm parallel to static evaluation, driven by OpenClaw. The platform supplies storage, lifecycle, and reporting; OpenClaw plans and executes. - Data model: IntelligentEval + Session + Message tables (new, not reusing exploration) - Lifecycle state machine: draft → planning → pending_approval → executing → completed/cancelled/failed - Session API: create/message (channel-forwarded)/close with turn accounting - Report API: pydantic-validated structured report, executing → completed, Markdown export (pure renderer) - Alembic migration for the three tables; domain glossary added to CONTEXT.md
277 lines
10 KiB
Python
277 lines
10 KiB
Python
"""Intelligent eval lifecycle (状态机 + 领域操作).
|
||
|
||
状态机：
|
||
draft → planning → pending_approval → executing → completed
|
||
→ cancelled
|
||
→ failed
|
||
pending_approval 可打回 → planning（附反馈）
|
||
executing 可取消 → cancelled
|
||
|
||
非法转换抛 IntelligentEvalTransitionError，路由层映射为 409。
|
||
"""
|
||
|
||
import json
|
||
from typing import Any
|
||
|
||
from sqlmodel import Session
|
||
|
||
from agenteval.channels.factory import ChannelFactory
|
||
from agenteval.config import get_settings
|
||
from agenteval.intelligent_eval.models import (
|
||
IntelligentEval,
|
||
IntelligentEvalMessage,
|
||
IntelligentEvalSession,
|
||
IntelligentEvalSessionStatus,
|
||
IntelligentEvalStatus,
|
||
)
|
||
from agenteval.intelligent_eval.repository import (
|
||
IntelligentEvalMessageRepository,
|
||
IntelligentEvalRepository,
|
||
IntelligentEvalSessionRepository,
|
||
)
|
||
from agenteval.storage.db import utc_now
|
||
from agenteval.storage.repository import TargetRepository
|
||
|
||
# 合法转换表：当前状态 → 允许的目标状态集合
|
||
_TRANSITIONS: dict[IntelligentEvalStatus, set[IntelligentEvalStatus]] = {
|
||
IntelligentEvalStatus.DRAFT: {IntelligentEvalStatus.PLANNING},
|
||
IntelligentEvalStatus.PLANNING: {IntelligentEvalStatus.PENDING_APPROVAL},
|
||
IntelligentEvalStatus.PENDING_APPROVAL: {IntelligentEvalStatus.EXECUTING, IntelligentEvalStatus.PLANNING, IntelligentEvalStatus.CANCELLED},
|
||
IntelligentEvalStatus.EXECUTING: {IntelligentEvalStatus.COMPLETED, IntelligentEvalStatus.CANCELLED, IntelligentEvalStatus.FAILED},
|
||
IntelligentEvalStatus.COMPLETED: set(),
|
||
IntelligentEvalStatus.CANCELLED: set(),
|
||
IntelligentEvalStatus.FAILED: set(),
|
||
}
|
||
|
||
|
||
class IntelligentEvalNotFoundError(Exception):
|
||
pass
|
||
|
||
|
||
class IntelligentEvalTransitionError(Exception):
|
||
def __init__(self, reason: str):
|
||
self.reason = reason
|
||
super().__init__(reason)
|
||
|
||
|
||
class IntelligentEvalChannelError(Exception):
|
||
pass
|
||
|
||
|
||
def coerce_reply_text(content: Any) -> str:
|
||
"""Flatten a reply payload to text (tutu returns msgBody as a parsed object)."""
|
||
if isinstance(content, str):
|
||
return content
|
||
if isinstance(content, dict):
|
||
for key in ("content", "text", "message"):
|
||
value = content.get(key)
|
||
if isinstance(value, str) and value:
|
||
return value
|
||
if content is None:
|
||
return ""
|
||
return json.dumps(content, ensure_ascii=False)
|
||
|
||
|
||
def _get_or_raise(repo: IntelligentEvalRepository, eval_id: str) -> IntelligentEval:
|
||
ev = repo.get(eval_id)
|
||
if ev is None:
|
||
raise IntelligentEvalNotFoundError(f"intelligent eval {eval_id} not found")
|
||
return ev
|
||
|
||
|
||
def _transition(repo: IntelligentEvalRepository, ev: IntelligentEval, target: IntelligentEvalStatus) -> IntelligentEval:
|
||
allowed = _TRANSITIONS.get(ev.status, set())
|
||
if target not in allowed:
|
||
raise IntelligentEvalTransitionError(
|
||
f"cannot transition from {ev.status.value} to {target.value}"
|
||
)
|
||
return repo.transition_status(ev.id, target)
|
||
|
||
|
||
def create_eval(
|
||
session: Session,
|
||
*,
|
||
name: str,
|
||
target_id: str,
|
||
goal: str,
|
||
seeds: dict[str, Any],
|
||
intent: str,
|
||
role_description: str,
|
||
time_window_hours: int = 24,
|
||
) -> IntelligentEval:
|
||
"""创建智能评估并直接进入 planning 状态（draft → planning 一步完成）。"""
|
||
repo = IntelligentEvalRepository(session)
|
||
ev = repo.create(IntelligentEval(
|
||
name=name,
|
||
target_id=target_id,
|
||
status=IntelligentEvalStatus.DRAFT,
|
||
goal=goal,
|
||
seeds=seeds,
|
||
intent=intent,
|
||
role_description=role_description,
|
||
time_window_hours=time_window_hours,
|
||
))
|
||
return _transition(repo, ev, IntelligentEvalStatus.PLANNING)
|
||
|
||
|
||
def submit_plan(session: Session, eval_id: str, plan: dict[str, Any]) -> IntelligentEval:
|
||
"""OpenClaw 提交粗计划：planning → pending_approval。"""
|
||
repo = IntelligentEvalRepository(session)
|
||
ev = _get_or_raise(repo, eval_id)
|
||
ev.plan = plan
|
||
ev.plan_feedback = None
|
||
ev = repo.update(ev)
|
||
return _transition(repo, ev, IntelligentEvalStatus.PENDING_APPROVAL)
|
||
|
||
|
||
def approve(session: Session, eval_id: str) -> IntelligentEval:
|
||
"""用户批准：pending_approval → executing。"""
|
||
repo = IntelligentEvalRepository(session)
|
||
ev = _get_or_raise(repo, eval_id)
|
||
return _transition(repo, ev, IntelligentEvalStatus.EXECUTING)
|
||
|
||
|
||
def reject(session: Session, eval_id: str, feedback: str) -> IntelligentEval:
|
||
"""用户打回：pending_approval → planning（附反馈）。"""
|
||
repo = IntelligentEvalRepository(session)
|
||
ev = _get_or_raise(repo, eval_id)
|
||
ev.plan_feedback = feedback
|
||
ev = repo.update(ev)
|
||
return _transition(repo, ev, IntelligentEvalStatus.PLANNING)
|
||
|
||
|
||
def cancel(session: Session, eval_id: str) -> IntelligentEval:
|
||
"""用户取消：pending_approval / executing → cancelled。"""
|
||
repo = IntelligentEvalRepository(session)
|
||
ev = _get_or_raise(repo, eval_id)
|
||
return _transition(repo, ev, IntelligentEvalStatus.CANCELLED)
|
||
|
||
|
||
def submit_report(session: Session, eval_id: str, report: dict[str, Any]) -> IntelligentEval:
|
||
"""OpenClaw 提交结构化报告：executing → completed。
|
||
|
||
报告结构校验在路由层（pydantic），此处只负责落库与状态迁移。
|
||
"""
|
||
repo = IntelligentEvalRepository(session)
|
||
ev = _get_or_raise(repo, eval_id)
|
||
ev.report = report
|
||
ev = repo.update(ev)
|
||
return _transition(repo, ev, IntelligentEvalStatus.COMPLETED)
|
||
|
||
|
||
def get_eval(session: Session, eval_id: str) -> IntelligentEval:
|
||
repo = IntelligentEvalRepository(session)
|
||
return _get_or_raise(repo, eval_id)
|
||
|
||
|
||
def list_evals(session: Session) -> list[IntelligentEval]:
|
||
return IntelligentEvalRepository(session).list_all()
|
||
|
||
|
||
def _get_session_or_raise(repo: IntelligentEvalSessionRepository, session_id: str) -> IntelligentEvalSession:
|
||
obj = repo.get(session_id)
|
||
if obj is None:
|
||
raise IntelligentEvalNotFoundError(f"intelligent eval session {session_id} not found")
|
||
return obj
|
||
|
||
|
||
def open_session(
|
||
session: Session,
|
||
*,
|
||
eval_id: str,
|
||
persona: dict[str, Any],
|
||
goal: str,
|
||
dimension: str | None = None,
|
||
) -> IntelligentEvalSession:
|
||
"""创建虚拟用户会话：仅 executing 状态的评估可创建。"""
|
||
eval_repo = IntelligentEvalRepository(session)
|
||
ev = _get_or_raise(eval_repo, eval_id)
|
||
if ev.status != IntelligentEvalStatus.EXECUTING:
|
||
raise IntelligentEvalTransitionError(
|
||
f"评估不在执行中（当前 {ev.status.value}），无法创建会话"
|
||
)
|
||
repo = IntelligentEvalSessionRepository(session)
|
||
return repo.create(IntelligentEvalSession(
|
||
eval_id=ev.id,
|
||
target_id=ev.target_id,
|
||
persona=persona,
|
||
goal=goal,
|
||
dimension=dimension,
|
||
))
|
||
|
||
|
||
async def conduct_turn(session: Session, *, session_id: str, content: str) -> dict[str, Any]:
|
||
"""一轮完整问答：状态检查 → 通道往返 → 双条消息落库 → 轮次自增。"""
|
||
repo = IntelligentEvalSessionRepository(session)
|
||
obj = _get_session_or_raise(repo, session_id)
|
||
if obj.status != IntelligentEvalSessionStatus.RUNNING:
|
||
raise IntelligentEvalTransitionError("会话不在进行中，拒收消息")
|
||
|
||
target = TargetRepository(session).get(obj.target_id)
|
||
if not target:
|
||
raise IntelligentEvalNotFoundError("session target not found")
|
||
|
||
channel = ChannelFactory.create(target)
|
||
sent_at = utc_now()
|
||
try:
|
||
send_result = await channel.send(content)
|
||
except Exception as exc: # channel adapters raise transport-specific errors
|
||
raise IntelligentEvalChannelError(f"评测对象通道发送失败: {exc}") from exc
|
||
if not send_result.ok:
|
||
raise IntelligentEvalChannelError(f"评测对象通道发送失败: {send_result.error}")
|
||
|
||
message_repo = IntelligentEvalMessageRepository(session)
|
||
message_repo.create(IntelligentEvalMessage(
|
||
session_id=obj.id, role="user", content=content, created_at=sent_at
|
||
))
|
||
repo.increment_turns(obj.id)
|
||
|
||
try:
|
||
reply = await channel.poll_reply(
|
||
send_result.question_msg_id,
|
||
timeout=get_settings().poll_reply_timeout,
|
||
)
|
||
except Exception as exc:
|
||
raise IntelligentEvalChannelError(f"等待评测对象回复失败: {exc}") from exc
|
||
if reply is None:
|
||
raise IntelligentEvalChannelError("等待评测对象回复超时")
|
||
|
||
received_at = utc_now()
|
||
latency_ms = int((received_at - sent_at).total_seconds() * 1000)
|
||
reply_text = coerce_reply_text(reply.content)
|
||
message_repo.create(IntelligentEvalMessage(
|
||
session_id=obj.id,
|
||
role="assistant",
|
||
content=reply_text,
|
||
latency_ms=latency_ms,
|
||
created_at=received_at,
|
||
))
|
||
return {"reply": reply_text, "latency_ms": latency_ms, "turn_count": obj.turn_count + 1}
|
||
|
||
|
||
def close_session(session: Session, *, session_id: str, verdict: dict[str, Any]) -> IntelligentEvalSession:
|
||
"""关闭会话并记录结论（verdict）；仅 running 会话可关闭。"""
|
||
repo = IntelligentEvalSessionRepository(session)
|
||
obj = _get_session_or_raise(repo, session_id)
|
||
if obj.status != IntelligentEvalSessionStatus.RUNNING:
|
||
raise IntelligentEvalTransitionError("会话不在进行中，无法关闭")
|
||
closed = repo.close(obj.id, verdict)
|
||
if closed is None:
|
||
raise IntelligentEvalNotFoundError(f"intelligent eval session {session_id} not found")
|
||
return closed
|
||
|
||
|
||
def get_session_by_id(session: Session, session_id: str) -> IntelligentEvalSession:
|
||
return _get_session_or_raise(IntelligentEvalSessionRepository(session), session_id)
|
||
|
||
|
||
def list_sessions(session: Session, eval_id: str) -> list[IntelligentEvalSession]:
|
||
eval_repo = IntelligentEvalRepository(session)
|
||
_get_or_raise(eval_repo, eval_id)
|
||
return IntelligentEvalSessionRepository(session).list_by_eval(eval_id)
|
||
|
||
|
||
def list_messages(session: Session, session_id: str) -> list[IntelligentEvalMessage]:
|
||
_get_session_or_raise(IntelligentEvalSessionRepository(session), session_id)
|
||
return IntelligentEvalMessageRepository(session).list_by_session(session_id)
|