Compare commits

..

No commits in common. "df76edcf55e7d3d63e5c0b96c8648c8ecbeeb133" and "a665b496b0e9ed316ce8576505aa5f0fdd384f63" have entirely different histories.

17 changed files with 298 additions and 847 deletions

View File

@ -13,7 +13,7 @@ from typing import Any, Awaitable, Callable, Optional
from sqlmodel import Session from sqlmodel import Session
from agenteval.evaluation.report import load_campaign_report from agenteval.evaluation.report import generate_campaign_report
from agenteval.exploration.summary import summarize_campaign_exploration from agenteval.exploration.summary import summarize_campaign_exploration
from agenteval.model_gateway import ModelGateway from agenteval.model_gateway import ModelGateway
from agenteval.models import Campaign, ModelCapability, RunStatus from agenteval.models import Campaign, ModelCapability, RunStatus
@ -28,6 +28,7 @@ from agenteval.storage.repository import (
CampaignAnalysisRepository, CampaignAnalysisRepository,
CampaignRepository, CampaignRepository,
RunRepository, RunRepository,
ScenarioRepository,
) )
from agenteval.utils.llm import extract_reply_text, parse_json_from_llm_text from agenteval.utils.llm import extract_reply_text, parse_json_from_llm_text
@ -305,7 +306,8 @@ async def execute_campaign_analysis(
try: try:
client = chat_client or gateway_chat_client(runtime) client = chat_client or gateway_chat_client(runtime)
runs = RunRepository(session).list_by_campaign(campaign_id) runs = RunRepository(session).list_by_campaign(campaign_id)
report = load_campaign_report(session, campaign) scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
report = generate_campaign_report(campaign, runs, scenario_names=scenario_names)
result = await analyze_campaign( result = await analyze_campaign(
campaign=campaign, campaign=campaign,
report=report, report=report,

View File

@ -180,7 +180,7 @@ async def advance_campaign(
summary = campaign.summary or CampaignSummary() summary = campaign.summary or CampaignSummary()
summary.scheduler = SchedulerState(spawned_indices=sorted(spawned), errors=errors) summary.scheduler = SchedulerState(spawned_indices=sorted(spawned), errors=errors)
campaign.summary = summary campaign.summary = summary
repo.save_scheduler_state(campaign_id, summary) repo.update(campaign)
return result return result

View File

@ -15,13 +15,15 @@ from typing import Any, Optional
from sqlmodel import Session from sqlmodel import Session
from agenteval.evaluation.analysis import ChatClient, gateway_chat_client, resolve_analysis_model from agenteval.evaluation.analysis import ChatClient, gateway_chat_client, resolve_analysis_model
from agenteval.evaluation.report import load_campaign_report from agenteval.evaluation.report import generate_campaign_report
from agenteval.models import Campaign from agenteval.models import Campaign, EvalRun
from agenteval.storage.db import get_session, iso_utc, utc_now from agenteval.storage.db import get_session, iso_utc, utc_now
from agenteval.storage.repository import ( from agenteval.storage.repository import (
CampaignAnalysisRepository, CampaignAnalysisRepository,
CampaignPeriodComparisonRepository, CampaignPeriodComparisonRepository,
CampaignRepository, CampaignRepository,
RunRepository,
ScenarioRepository,
) )
from agenteval.utils.llm import parse_json_from_llm_text from agenteval.utils.llm import parse_json_from_llm_text
@ -152,14 +154,21 @@ def compute_metric_diff(
return {"overall": overall, "scenarios": scenarios} return {"overall": overall, "scenarios": scenarios}
def build_campaign_report_dict(campaign: Campaign, session: Session) -> dict[str, Any]:
"""为周期对比现算一期活动的报告 dict读路径不重算聚合口径"""
runs: list[EvalRun] = RunRepository(session).list_by_campaign(campaign.id)
scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
return generate_campaign_report(campaign, runs, scenario_names=scenario_names)
def build_comparison_payload(campaign: Campaign, session: Session) -> dict[str, Any]: def build_comparison_payload(campaign: Campaign, session: Session) -> dict[str, Any]:
"""GET 返回体:自动基线信息 + 机械 diff无基线时两者均为 null""" """GET 返回体:自动基线信息 + 机械 diff无基线时两者均为 null"""
baseline = resolve_auto_baseline(campaign, session) baseline = resolve_auto_baseline(campaign, session)
if baseline is None: if baseline is None:
return {"auto_baseline": None, "metric_diff": None} return {"auto_baseline": None, "metric_diff": None}
diff = compute_metric_diff( diff = compute_metric_diff(
load_campaign_report(session, baseline), build_campaign_report_dict(baseline, session),
load_campaign_report(session, campaign), build_campaign_report_dict(campaign, session),
) )
return { return {
"auto_baseline": { "auto_baseline": {
@ -349,8 +358,8 @@ async def execute_campaign_comparison(
) )
try: try:
diff = compute_metric_diff( diff = compute_metric_diff(
load_campaign_report(session, baseline), build_campaign_report_dict(baseline, session),
load_campaign_report(session, campaign), build_campaign_report_dict(campaign, session),
) )
valid_scenario_ids = {s["scenario_id"] for s in diff["scenarios"]} valid_scenario_ids = {s["scenario_id"] for s in diff["scenarios"]}
result = await narrate_period_comparison( result = await narrate_period_comparison(

View File

@ -9,8 +9,6 @@ from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from typing import Any, Optional from typing import Any, Optional
from sqlmodel import Session
from agenteval.evaluation.case_verdict import CaseEvidence, resolve_case_verdicts from agenteval.evaluation.case_verdict import CaseEvidence, resolve_case_verdicts
from agenteval.evaluation.metrics import aggregate_runs from agenteval.evaluation.metrics import aggregate_runs
from agenteval.evaluation.report_render import render_html, render_json, render_markdown from agenteval.evaluation.report_render import render_html, render_json, render_markdown
@ -361,12 +359,6 @@ def generate_campaign_report(
} }
def load_campaign_report(session: Session, campaign: Campaign) -> dict[str, Any]:
"""取数 + 聚合一步完成:报告 / 分析 / 对比 / 导出共用的活动报告 dict 取法。"""
runs = RunRepository(session).list_by_campaign(campaign.id)
return generate_campaign_report(campaign, runs, scenario_names=ScenarioRepository(session).name_map())
def save_report(run_id: str, fmt: str = "html", output_dir: Optional[Path] = None) -> Path: def save_report(run_id: str, fmt: str = "html", output_dir: Optional[Path] = None) -> Path:
"""Generate a run report and save it to disk in the requested format.""" """Generate a run report and save it to disk in the requested format."""
output_dir = output_dir or DATA_DIR / "reports" output_dir = output_dir or DATA_DIR / "reports"

View File

@ -1,26 +0,0 @@
"""Domain errors for exploratory evaluation (不依赖 fastapi).
领域违规用类型化异常表达router 统一映射 HTTP 状态码
NotFound 404Guardrail 409reason 即给常驻代理的反馈
Channel 502
"""
class ExplorationNotFoundError(Exception):
"""探索域实体(活动 / 评测对象 / 会话)不存在。"""
class ExplorationGuardrailError(Exception):
"""平台账本规则违规(状态机 / 预算),拒绝本身即反馈。"""
def __init__(self, reason: str):
self.reason = reason
super().__init__(reason)
class ExplorationChannelError(Exception):
"""评测对象通道收发失败。"""
def __init__(self, reason: str):
self.reason = reason
super().__init__(reason)

View File

@ -1,187 +0,0 @@
"""Exploration session lifecycle (探索会话生命周期).
会话三态的领域操作open_session / conduct_turn / close_session
平台账本预算状态机轮次记账全部在此执行违规抛类型化领域
异常errors.py router 映射 HTTP 状态码通道收发沿用
ChannelFactory 既有接缝
"""
import json
from typing import Any, Optional
from sqlmodel import Session
from agenteval.channels.factory import ChannelFactory
from agenteval.config import get_settings
from agenteval.exploration.errors import (
ExplorationChannelError,
ExplorationGuardrailError,
ExplorationNotFoundError,
)
from agenteval.exploration.judge import start_judge_review
from agenteval.exploration.models import (
ExplorationBudget,
ExplorationMessage,
ExplorationSession,
ExplorationSessionStatus,
ExplorationTrigger,
normalize_experience,
resolve_budget,
)
from agenteval.models import CampaignStatus
from agenteval.storage.db import as_utc, utc_now
from agenteval.storage.repository import (
CampaignRepository,
ExplorationMessageRepository,
ExplorationSessionRepository,
TargetRepository,
)
def coerce_reply_text(content: Any) -> str:
"""Flatten a reply payload to text; tutu returns msgBody as a parsed object,
and str(dict) would leak a Python repr into the view."""
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 _check_creation_guardrails(
campaign,
triggered_by: ExplorationTrigger,
budget: ExplorationBudget,
repo: ExplorationSessionRepository,
) -> None:
if campaign.status != CampaignStatus.RUNNING:
raise ExplorationGuardrailError("活动不在进行中,无法创建探索会话")
if campaign.time_scale != 1 and triggered_by != ExplorationTrigger.MANUAL:
raise ExplorationGuardrailError("加速调试线仅允许手动创建探索会话(时间压缩与拟真相冲突)")
sessions = repo.list_by_campaign(campaign.id)
if len(sessions) >= budget.max_sessions:
raise ExplorationGuardrailError(f"探索会话数超出预算:本活动窗口最多 {budget.max_sessions} 个会话")
if sessions:
latest = max(s.created_at for s in sessions if s.created_at)
elapsed = (utc_now() - as_utc(latest)).total_seconds()
if elapsed < budget.min_interval_seconds:
wait_minutes = budget.min_interval_seconds // 60
raise ExplorationGuardrailError(f"相邻探索会话间隔不足:最小间隔 {wait_minutes} 分钟,请稍后再试")
def open_session(
db_session: Session,
*,
campaign_id: str,
persona: dict[str, Any],
goal: str,
seed_ref: Optional[dict[str, Any]] = None,
triggered_by: ExplorationTrigger = ExplorationTrigger.AUTO,
) -> ExplorationSession:
"""按账本规则开一个 running 会话;违规抛领域异常。"""
campaign = CampaignRepository(db_session).get(campaign_id)
if not campaign:
raise ExplorationNotFoundError("campaign not found")
if not TargetRepository(db_session).get(campaign.target_id):
raise ExplorationNotFoundError("campaign target not found")
budget = resolve_budget(campaign)
repo = ExplorationSessionRepository(db_session)
_check_creation_guardrails(campaign, triggered_by, budget, repo)
return repo.create(ExplorationSession(
campaign_id=campaign.id,
target_id=campaign.target_id,
persona=persona,
goal=goal,
seed_ref=seed_ref,
triggered_by=triggered_by,
))
async def conduct_turn(db_session: Session, *, session_id: str, content: str) -> dict[str, Any]:
"""一轮完整问答:状态机拒收 → 轮次预算记账 → 通道往返 → 双条落库。
消息已送达即消耗一轮预算平台账本先落用户消息再等回复
"""
repo = ExplorationSessionRepository(db_session)
session_obj = repo.get(session_id)
if not session_obj:
raise ExplorationNotFoundError("exploration session not found")
if session_obj.status != ExplorationSessionStatus.RUNNING:
raise ExplorationGuardrailError("探索会话不在进行中,拒收消息")
campaign = CampaignRepository(db_session).get(session_obj.campaign_id)
budget = resolve_budget(campaign) if campaign else ExplorationBudget()
if session_obj.turn_count >= budget.max_turns:
raise ExplorationGuardrailError(f"会话轮数超出预算:单会话最多 {budget.max_turns}")
target = TargetRepository(db_session).get(session_obj.target_id)
if not target:
raise ExplorationNotFoundError("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 ExplorationChannelError(f"评测对象通道发送失败: {exc}") from exc
if not send_result.ok:
raise ExplorationChannelError(f"评测对象通道发送失败: {send_result.error}")
message_repo = ExplorationMessageRepository(db_session)
round_index = session_obj.turn_count + 1
message_repo.save_message(
ExplorationMessage(
session_id=session_obj.id, round_index=round_index, role="user", content=content, created_at=sent_at
)
)
session_obj.turn_count = round_index
repo.update(session_obj)
try:
reply = await channel.poll_reply(
send_result.question_msg_id,
timeout=get_settings().poll_reply_timeout,
)
except Exception as exc:
raise ExplorationChannelError(f"等待评测对象回复失败: {exc}") from exc
if reply is None:
raise ExplorationChannelError("等待评测对象回复超时")
received_at = utc_now()
latency_ms = int((received_at - sent_at).total_seconds() * 1000)
reply_text = coerce_reply_text(reply.content)
message_repo.save_message(
ExplorationMessage(
session_id=session_obj.id,
round_index=round_index,
role="assistant",
content=reply_text,
latency_ms=latency_ms,
created_at=received_at,
)
)
return {"reply": reply_text, "latency_ms": latency_ms, "turn_count": round_index}
def close_session(db_session: Session, *, session_id: str, experience: dict[str, Any]) -> ExplorationSession:
"""关闭会话:体验记录归一化 → 状态迁移 → 触发 judge 抽样复核。"""
repo = ExplorationSessionRepository(db_session)
session_obj = repo.get(session_id)
if not session_obj:
raise ExplorationNotFoundError("exploration session not found")
if session_obj.status != ExplorationSessionStatus.RUNNING:
raise ExplorationGuardrailError("探索会话不在进行中,无法关闭")
session_obj.experience = normalize_experience(experience)
session_obj.status = ExplorationSessionStatus.COMPLETED
session_obj.closed_at = utc_now()
updated = repo.update(session_obj)
start_judge_review(session_obj.id)
return updated

View File

@ -1,100 +0,0 @@
"""Exploration patrol (探索巡检).
常驻智能体的无状态巡检筛出参与探索的正式线活动running
time_scale == 1有种子集汇报水位以来的新增结果与剩余探索预算
构建完响应后推进水位保证下次只报增量
"""
from datetime import datetime
from typing import Any
from sqlmodel import Session
from agenteval.evaluation.report import generate_campaign_report
from agenteval.exploration.models import resolve_budget
from agenteval.models import Campaign, CampaignStatus, EvalRun
from agenteval.storage.db import as_utc, iso_utc, utc_now
from agenteval.storage.repository import (
CampaignRepository,
ExplorationSessionRepository,
RunRepository,
ScenarioRepository,
TargetRepository,
)
def _new_runs_since(runs: list[EvalRun], watermark: datetime | None) -> list[EvalRun]:
fresh = []
for run in runs:
if run.completed_at is None:
continue
if watermark is not None and as_utc(run.completed_at) <= watermark:
continue
fresh.append(run)
return fresh
def patrol_report(db_session: Session) -> dict[str, Any]:
"""一次巡检的完整读模型patrolled_at + 逐活动条目列表。
水位取构建响应之后的时刻查询与持久化之间完成的结果不会在下次重复上报
"""
campaign_repo = CampaignRepository(db_session)
run_repo = RunRepository(db_session)
exploration_repo = ExplorationSessionRepository(db_session)
scenario_names = ScenarioRepository(db_session).name_map()
target_names = {t.id: t.name for t in TargetRepository(db_session).list_all()}
patrolled_at = utc_now()
entries: list[dict[str, Any]] = []
patrolled_campaigns: list[Campaign] = []
for campaign in campaign_repo.list_all():
if campaign.status != CampaignStatus.RUNNING:
continue
if campaign.time_scale != 1:
continue
if campaign.exploration_seeds is None:
continue
watermark = as_utc(campaign.last_patrolled_at) if campaign.last_patrolled_at else None
fresh = _new_runs_since(run_repo.list_by_campaign(campaign.id), watermark)
new_results = None
if fresh:
report = generate_campaign_report(campaign, fresh, scenario_names=scenario_names)
new_results = {
"summary": report["summary"],
"capability_summary": report["capability_summary"],
}
budget = resolve_budget(campaign)
sessions = exploration_repo.list_by_campaign(campaign.id)
seconds_since_last_session = None
if sessions:
latest = max(as_utc(s.created_at) for s in sessions if s.created_at)
seconds_since_last_session = int((patrolled_at - latest).total_seconds())
entries.append(
{
"campaign_id": campaign.id,
"campaign_name": campaign.name,
"target_id": campaign.target_id,
"target_name": target_names.get(campaign.target_id),
"last_patrolled_at": iso_utc(campaign.last_patrolled_at),
"new_results": new_results,
"budget": {
"max_sessions": budget.max_sessions,
"sessions_used": len(sessions),
"remaining_sessions": max(0, budget.max_sessions - len(sessions)),
"max_turns": budget.max_turns,
"min_interval_seconds": budget.min_interval_seconds,
"seconds_since_last_session": seconds_since_last_session,
},
}
)
patrolled_campaigns.append(campaign)
watermark_at = utc_now()
for campaign in patrolled_campaigns:
campaign_repo.touch_patrol_watermark(campaign.id, watermark_at)
return {"patrolled_at": iso_utc(watermark_at), "campaigns": entries}

View File

@ -1,12 +1,11 @@
"""Repository layer for database access.""" """Repository layer for database access."""
from datetime import datetime
from typing import Generic, Optional, TypeVar from typing import Generic, Optional, TypeVar
from sqlmodel import Session, select from sqlmodel import Session, select
from agenteval.exploration.models import ExplorationMessage, ExplorationSession from agenteval.exploration.models import ExplorationMessage, ExplorationSession
from agenteval.models import Campaign, CampaignStatus, CampaignSummary, Case, EvalResult, EvalRun, EvalTarget, Scenario from agenteval.models import Campaign, Case, EvalResult, EvalRun, EvalTarget, Scenario
from agenteval.services.model_configs import ModelConfigService from agenteval.services.model_configs import ModelConfigService
from agenteval.storage.db import ( from agenteval.storage.db import (
CampaignAnalysisDB, CampaignAnalysisDB,
@ -248,10 +247,6 @@ class ScenarioRepository(BaseRepository[Scenario, ScenarioDB]):
raise raise
return True return True
def name_map(self) -> dict[str, str]:
"""scenario_id → 名称映射:报告 / 时间线 / 列表等读路径共用的场景名取法。"""
return {sid: name for sid, name in self.session.exec(select(ScenarioDB.id, ScenarioDB.name)).all()}
class RunRepository(BaseRepository[EvalRun, EvalRunDB]): class RunRepository(BaseRepository[EvalRun, EvalRunDB]):
"""Repository for evaluation runs.""" """Repository for evaluation runs."""
@ -412,38 +407,6 @@ class CampaignRepository(BaseRepository[Campaign, CampaignDB]):
self.session.refresh(existing) self.session.refresh(existing)
return self._from_db(existing) return self._from_db(existing)
def touch_patrol_watermark(self, campaign_id: str, at: datetime) -> None:
"""窄口径原子更新:只写巡检水位,不覆写并发的 status / summary 变更。"""
db = self.session.get(CampaignDB, campaign_id)
if not db:
return
db.last_patrolled_at = at
self.session.add(db)
self.session.commit()
self.session.refresh(db)
def mark_cancelled(self, campaign_id: str, at: datetime) -> Optional[Campaign]:
"""窄口径终态迁移:只写 status + completed_at不抹掉水位与调度进度。"""
db = self.session.get(CampaignDB, campaign_id)
if not db:
return None
db.status = CampaignStatus.CANCELLED.value
db.completed_at = at
self.session.add(db)
self.session.commit()
self.session.refresh(db)
return self._from_db(db)
def save_scheduler_state(self, campaign_id: str, summary: CampaignSummary) -> None:
"""窄口径调度进度持久化:只写 summary 列,不覆写并发的状态 / 水位变更。"""
db = self.session.get(CampaignDB, campaign_id)
if not db:
return
db.set_summary(summary.model_dump(mode="json"))
self.session.add(db)
self.session.commit()
self.session.refresh(db)
class CampaignAnalysisRepository: class CampaignAnalysisRepository:
"""Repository for campaign analysis rows (one per campaign, upserted).""" """Repository for campaign analysis rows (one per campaign, upserted)."""

View File

@ -14,6 +14,7 @@ from sqlmodel import Session
from agenteval.evaluation.analysis import resolve_analysis_model, start_campaign_analysis from agenteval.evaluation.analysis import resolve_analysis_model, start_campaign_analysis
from agenteval.evaluation.campaign_runner import campaign_progress, request_cancel, start_campaign from agenteval.evaluation.campaign_runner import campaign_progress, request_cancel, start_campaign
from agenteval.evaluation.comparison import ( from agenteval.evaluation.comparison import (
build_campaign_report_dict,
build_comparison_payload, build_comparison_payload,
compute_metric_diff, compute_metric_diff,
resolve_auto_baseline, resolve_auto_baseline,
@ -21,7 +22,7 @@ from agenteval.evaluation.comparison import (
) )
from agenteval.evaluation.report import ( from agenteval.evaluation.report import (
build_campaign_timeline, build_campaign_timeline,
load_campaign_report, generate_campaign_report,
summarize_campaign_progress, summarize_campaign_progress,
) )
from agenteval.evaluation.report_render import render_campaign_markdown from agenteval.evaluation.report_render import render_campaign_markdown
@ -117,7 +118,9 @@ async def cancel_campaign(campaign_id: str, session: Session = Depends(get_db))
if campaign.status not in (CampaignStatus.PLANNED, CampaignStatus.RUNNING): if campaign.status not in (CampaignStatus.PLANNED, CampaignStatus.RUNNING):
raise HTTPException(status_code=400, detail="campaign is not in a cancellable state") raise HTTPException(status_code=400, detail="campaign is not in a cancellable state")
campaign = repo.mark_cancelled(campaign_id, utc_now()) or campaign campaign.status = CampaignStatus.CANCELLED
campaign.completed_at = utc_now()
repo.update(campaign)
request_cancel(campaign_id) request_cancel(campaign_id)
settle_campaign_sessions(campaign_id, session) settle_campaign_sessions(campaign_id, session)
return campaign.model_dump() return campaign.model_dump()
@ -128,7 +131,9 @@ async def get_campaign_report(campaign_id: str, session: Session = Depends(get_d
campaign = CampaignRepository(session).get(campaign_id) campaign = CampaignRepository(session).get(campaign_id)
if not campaign: if not campaign:
raise HTTPException(status_code=404, detail="campaign not found") raise HTTPException(status_code=404, detail="campaign not found")
report = load_campaign_report(session, campaign) runs = RunRepository(session).list_by_campaign(campaign_id)
scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
report = generate_campaign_report(campaign, runs, scenario_names=scenario_names)
exploration = summarize_campaign_exploration(session, campaign_id) exploration = summarize_campaign_exploration(session, campaign_id)
if exploration is not None: if exploration is not None:
report["exploration"] = exploration report["exploration"] = exploration
@ -140,8 +145,8 @@ async def get_campaign_report_markdown(campaign_id: str, session: Session = Depe
campaign = CampaignRepository(session).get(campaign_id) campaign = CampaignRepository(session).get(campaign_id)
if not campaign: if not campaign:
raise HTTPException(status_code=404, detail="campaign not found") raise HTTPException(status_code=404, detail="campaign not found")
report = load_campaign_report(session, campaign) runs = RunRepository(session).list_by_campaign(campaign_id)
scenario_names = ScenarioRepository(session).name_map() scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
analysis_row = CampaignAnalysisRepository(session).get_by_campaign(campaign_id) analysis_row = CampaignAnalysisRepository(session).get_by_campaign(campaign_id)
analysis = analysis_row.get_result() if analysis_row and analysis_row.status == "completed" else None analysis = analysis_row.get_result() if analysis_row and analysis_row.status == "completed" else None
target = TargetRepository(session).get(campaign.target_id) target = TargetRepository(session).get(campaign.target_id)
@ -153,8 +158,8 @@ async def get_campaign_report_markdown(campaign_id: str, session: Session = Depe
baseline = CampaignRepository(session).get(cmp_row.baseline_campaign_id) baseline = CampaignRepository(session).get(cmp_row.baseline_campaign_id)
metric_diff = ( metric_diff = (
compute_metric_diff( compute_metric_diff(
load_campaign_report(session, baseline), build_campaign_report_dict(baseline, session),
load_campaign_report(session, campaign), build_campaign_report_dict(campaign, session),
) )
if baseline is not None if baseline is not None
else None else None
@ -173,7 +178,7 @@ async def get_campaign_report_markdown(campaign_id: str, session: Session = Depe
} }
md = render_campaign_markdown( md = render_campaign_markdown(
report, generate_campaign_report(campaign, runs, scenario_names=scenario_names),
analysis=analysis, analysis=analysis,
comparison=comparison, comparison=comparison,
exploration=summarize_campaign_exploration(session, campaign_id), exploration=summarize_campaign_exploration(session, campaign_id),
@ -193,7 +198,7 @@ async def get_campaign_timeline(campaign_id: str, session: Session = Depends(get
if not campaign: if not campaign:
raise HTTPException(status_code=404, detail="campaign not found") raise HTTPException(status_code=404, detail="campaign not found")
runs = RunRepository(session).list_by_campaign(campaign_id) runs = RunRepository(session).list_by_campaign(campaign_id)
scenario_names = ScenarioRepository(session).name_map() scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
return {"entries": build_campaign_timeline(campaign, runs, scenario_names=scenario_names)} return {"entries": build_campaign_timeline(campaign, runs, scenario_names=scenario_names)}
@ -249,8 +254,8 @@ async def get_campaign_comparison(campaign_id: str, session: Session = Depends(g
baseline = CampaignRepository(session).get(row.baseline_campaign_id) baseline = CampaignRepository(session).get(row.baseline_campaign_id)
metric_diff = ( metric_diff = (
compute_metric_diff( compute_metric_diff(
load_campaign_report(session, baseline), build_campaign_report_dict(baseline, session),
load_campaign_report(session, campaign), build_campaign_report_dict(campaign, session),
) )
if baseline is not None if baseline is not None
else None else None

View File

@ -1,28 +1,45 @@
"""API routes for exploratory evaluation sessions (探索式评测, v0.9). """API routes for exploratory evaluation sessions (探索式评测, v0.9).
领域逻辑账本规则状态机巡检 exploration/lifecycle.py The virtual user (OpenClaw) drives these sessions through plain HTTP: create a
patrol.py本层只做 HTTP 翻译解析请求调用领域操作把领域异常 running session against a campaign, converse with the target through its real
映射为状态码NotFound404Guardrail409Channel502预算违规 channel, then close with a structured self-reported experience record.
与状态违规的拒绝文案由领域层给出拒绝本身就是对常驻智能体的反馈
Guardrails are a platform ledger enforced here on every call, never trusted
to client self-discipline. Budget overruns and state violations are rejected
with 409 plus a readable reason, so the rejection itself is feedback to the
resident agent.
""" """
import json
from datetime import datetime
from typing import Any from typing import Any
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from sqlmodel import Session from sqlmodel import Session
from agenteval.exploration import lifecycle, patrol from agenteval.channels.factory import ChannelFactory
from agenteval.exploration.errors import ( from agenteval.config import get_settings
ExplorationChannelError, from agenteval.evaluation.report import generate_campaign_report
ExplorationGuardrailError, from agenteval.exploration.judge import start_judge_review
ExplorationNotFoundError, from agenteval.exploration.models import (
ExplorationBudget,
ExplorationMessage,
ExplorationSession,
ExplorationSessionStatus,
ExplorationTrigger,
normalize_experience,
resolve_budget,
) )
from agenteval.exploration.models import ExplorationTrigger from agenteval.models import Campaign, CampaignStatus, EvalRun
from agenteval.storage.db import as_utc, iso_utc, utc_now
from agenteval.storage.repository import ( from agenteval.storage.repository import (
CampaignRepository, CampaignRepository,
ExplorationMessageRepository, ExplorationMessageRepository,
ExplorationSessionRepository, ExplorationSessionRepository,
RunRepository,
ScenarioRepository,
TargetRepository,
) )
from agenteval.web.deps import get_db from agenteval.web.deps import get_db
@ -45,18 +62,133 @@ class CloseSessionRequest(BaseModel):
experience: dict[str, Any] experience: dict[str, Any]
def _translate(exc: Exception) -> HTTPException: def _coerce_reply_text(content: Any) -> str:
if isinstance(exc, ExplorationNotFoundError): """Flatten a reply payload to text; tutu returns msgBody as a parsed object,
return HTTPException(status_code=404, detail=str(exc)) and str(dict) would leak a Python repr into the view."""
if isinstance(exc, ExplorationGuardrailError): if isinstance(content, str):
return HTTPException(status_code=409, detail=exc.reason) return content
return HTTPException(status_code=502, detail=exc.reason) 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 _check_creation_guardrails(
campaign: Campaign,
triggered_by: ExplorationTrigger,
budget: ExplorationBudget,
repo: ExplorationSessionRepository,
) -> None:
if campaign.status != CampaignStatus.RUNNING:
raise HTTPException(status_code=409, detail="活动不在进行中,无法创建探索会话")
if campaign.time_scale != 1 and triggered_by != ExplorationTrigger.MANUAL:
raise HTTPException(
status_code=409,
detail="加速调试线仅允许手动创建探索会话(时间压缩与拟真相冲突)",
)
sessions = repo.list_by_campaign(campaign.id)
if len(sessions) >= budget.max_sessions:
raise HTTPException(
status_code=409,
detail=f"探索会话数超出预算:本活动窗口最多 {budget.max_sessions} 个会话",
)
if sessions:
latest = max(s.created_at for s in sessions if s.created_at)
elapsed = (utc_now() - as_utc(latest)).total_seconds()
if elapsed < budget.min_interval_seconds:
wait_minutes = budget.min_interval_seconds // 60
raise HTTPException(
status_code=409,
detail=f"相邻探索会话间隔不足:最小间隔 {wait_minutes} 分钟,请稍后再试",
)
def _new_runs_since(runs: list[EvalRun], watermark: datetime | None) -> list[EvalRun]:
fresh = []
for run in runs:
if run.completed_at is None:
continue
if watermark is not None and as_utc(run.completed_at) <= watermark:
continue
fresh.append(run)
return fresh
@router.get("/patrol") @router.get("/patrol")
async def patrol_endpoint(session: Session = Depends(get_db)) -> dict: async def patrol(session: Session = Depends(get_db)) -> dict:
"""Stateless patrol for the resident agent — see exploration/patrol.py.""" """Stateless patrol for the resident agent.
return patrol.patrol_report(session)
Reports every running production-line (time_scale == 1) campaign that
participates in exploration (has a seed set), with new results since the
last watermark and the remaining exploration budget. Advances each
patrolled campaign's watermark after building the response, so the next
call only reports increments.
"""
campaign_repo = CampaignRepository(session)
run_repo = RunRepository(session)
exploration_repo = ExplorationSessionRepository(session)
scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
target_names = {t.id: t.name for t in TargetRepository(session).list_all()}
patrolled_at = utc_now()
entries: list[dict[str, Any]] = []
patrolled_campaigns: list[Campaign] = []
for campaign in campaign_repo.list_all():
if campaign.status != CampaignStatus.RUNNING:
continue
if campaign.time_scale != 1:
continue
if campaign.exploration_seeds is None:
continue
watermark = as_utc(campaign.last_patrolled_at) if campaign.last_patrolled_at else None
fresh = _new_runs_since(run_repo.list_by_campaign(campaign.id), watermark)
new_results = None
if fresh:
report = generate_campaign_report(campaign, fresh, scenario_names=scenario_names)
new_results = {
"summary": report["summary"],
"capability_summary": report["capability_summary"],
}
budget = resolve_budget(campaign)
sessions = exploration_repo.list_by_campaign(campaign.id)
seconds_since_last_session = None
if sessions:
latest = max(as_utc(s.created_at) for s in sessions if s.created_at)
seconds_since_last_session = int((patrolled_at - latest).total_seconds())
entries.append(
{
"campaign_id": campaign.id,
"campaign_name": campaign.name,
"target_id": campaign.target_id,
"target_name": target_names.get(campaign.target_id),
"last_patrolled_at": iso_utc(campaign.last_patrolled_at),
"new_results": new_results,
"budget": {
"max_sessions": budget.max_sessions,
"sessions_used": len(sessions),
"remaining_sessions": max(0, budget.max_sessions - len(sessions)),
"max_turns": budget.max_turns,
"min_interval_seconds": budget.min_interval_seconds,
"seconds_since_last_session": seconds_since_last_session,
},
}
)
patrolled_campaigns.append(campaign)
# 水位取构建响应之后的时刻:查询与持久化之间完成的结果不会在下次重复上报。
watermark_at = utc_now()
for campaign in patrolled_campaigns:
campaign.last_patrolled_at = watermark_at
campaign_repo.update(campaign)
return {"patrolled_at": iso_utc(watermark_at), "campaigns": entries}
@router.post("/sessions") @router.post("/sessions")
@ -64,18 +196,25 @@ async def create_session(
request: CreateSessionRequest, request: CreateSessionRequest,
session: Session = Depends(get_db), session: Session = Depends(get_db),
) -> dict: ) -> dict:
try: campaign = CampaignRepository(session).get(request.campaign_id)
session_obj = lifecycle.open_session( if not campaign:
session, raise HTTPException(status_code=404, detail="campaign not found")
campaign_id=request.campaign_id, if not TargetRepository(session).get(campaign.target_id):
persona=request.persona, raise HTTPException(status_code=404, detail="campaign target not found")
goal=request.goal,
seed_ref=request.seed_ref, budget = resolve_budget(campaign)
triggered_by=request.triggered_by, repo = ExplorationSessionRepository(session)
) _check_creation_guardrails(campaign, request.triggered_by, budget, repo)
except (ExplorationNotFoundError, ExplorationGuardrailError) as exc:
raise _translate(exc) from exc session_obj = ExplorationSession(
return session_obj.model_dump(mode="json") campaign_id=campaign.id,
target_id=campaign.target_id,
persona=request.persona,
goal=request.goal,
seed_ref=request.seed_ref,
triggered_by=request.triggered_by,
)
return repo.create(session_obj).model_dump(mode="json")
@router.get("/campaigns/{campaign_id}/sessions") @router.get("/campaigns/{campaign_id}/sessions")
@ -106,10 +245,72 @@ async def send_session_message(
request: SendMessageRequest, request: SendMessageRequest,
session: Session = Depends(get_db), session: Session = Depends(get_db),
) -> dict: ) -> dict:
repo = ExplorationSessionRepository(session)
session_obj = repo.get(session_id)
if not session_obj:
raise HTTPException(status_code=404, detail="exploration session not found")
if session_obj.status != ExplorationSessionStatus.RUNNING:
raise HTTPException(status_code=409, detail="探索会话不在进行中,拒收消息")
campaign = CampaignRepository(session).get(session_obj.campaign_id)
budget = resolve_budget(campaign) if campaign else ExplorationBudget()
if session_obj.turn_count >= budget.max_turns:
raise HTTPException(
status_code=409,
detail=f"会话轮数超出预算:单会话最多 {budget.max_turns}",
)
target = TargetRepository(session).get(session_obj.target_id)
if not target:
raise HTTPException(status_code=404, detail="session target not found")
channel = ChannelFactory.create(target)
sent_at = utc_now()
try: try:
return await lifecycle.conduct_turn(session, session_id=session_id, content=request.content) send_result = await channel.send(request.content)
except (ExplorationNotFoundError, ExplorationGuardrailError, ExplorationChannelError) as exc: except Exception as exc: # channel adapters raise transport-specific errors
raise _translate(exc) from exc raise HTTPException(status_code=502, detail=f"评测对象通道发送失败: {exc}")
if not send_result.ok:
raise HTTPException(
status_code=502,
detail=f"评测对象通道发送失败: {send_result.error}",
)
# 消息已送达即消耗一轮预算(平台账本):先落用户消息,再等回复。
message_repo = ExplorationMessageRepository(session)
round_index = session_obj.turn_count + 1
message_repo.save_message(
ExplorationMessage(
session_id=session_obj.id, round_index=round_index, role="user", content=request.content, created_at=sent_at
)
)
session_obj.turn_count = round_index
repo.update(session_obj)
try:
reply = await channel.poll_reply(
send_result.question_msg_id,
timeout=get_settings().poll_reply_timeout,
)
except Exception as exc:
raise HTTPException(status_code=502, detail=f"等待评测对象回复失败: {exc}")
if reply is None:
raise HTTPException(status_code=502, detail="等待评测对象回复超时")
received_at = utc_now()
latency_ms = int((received_at - sent_at).total_seconds() * 1000)
reply_text = _coerce_reply_text(reply.content)
message_repo.save_message(
ExplorationMessage(
session_id=session_obj.id,
round_index=round_index,
role="assistant",
content=reply_text,
latency_ms=latency_ms,
created_at=received_at,
)
)
return {"reply": reply_text, "latency_ms": latency_ms, "turn_count": round_index}
@router.post("/sessions/{session_id}/close") @router.post("/sessions/{session_id}/close")
@ -118,8 +319,16 @@ async def close_session(
request: CloseSessionRequest, request: CloseSessionRequest,
session: Session = Depends(get_db), session: Session = Depends(get_db),
) -> dict: ) -> dict:
try: repo = ExplorationSessionRepository(session)
updated = lifecycle.close_session(session, session_id=session_id, experience=request.experience) session_obj = repo.get(session_id)
except (ExplorationNotFoundError, ExplorationGuardrailError) as exc: if not session_obj:
raise _translate(exc) from exc raise HTTPException(status_code=404, detail="exploration session not found")
if session_obj.status != ExplorationSessionStatus.RUNNING:
raise HTTPException(status_code=409, detail="探索会话不在进行中,无法关闭")
session_obj.experience = normalize_experience(request.experience)
session_obj.status = ExplorationSessionStatus.COMPLETED
session_obj.closed_at = utc_now()
updated = repo.update(session_obj)
start_judge_review(session_obj.id)
return updated.model_dump(mode="json") return updated.model_dump(mode="json")

View File

@ -68,7 +68,7 @@ async def _run_evaluation(
@router.get("") @router.get("")
async def list_runs(session: Session = Depends(get_db)) -> list[dict]: async def list_runs(session: Session = Depends(get_db)) -> list[dict]:
scenario_names = ScenarioRepository(session).name_map() scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
target_names = {t.id: t.name for t in TargetRepository(session).list_all()} target_names = {t.id: t.name for t in TargetRepository(session).list_all()}
return [ return [
{ {

View File

@ -36,10 +36,11 @@ def _settled(runs: list[EvalRun]) -> list[EvalRun]:
@router.get("/dashboard") @router.get("/dashboard")
def dashboard(session: Session = Depends(get_db)) -> dict: def dashboard(session: Session = Depends(get_db)) -> dict:
targets = TargetRepository(session).list_all() targets = TargetRepository(session).list_all()
scenario_names = ScenarioRepository(session).name_map() scenarios = ScenarioRepository(session).list_all()
runs = RunRepository(session).list_all() runs = RunRepository(session).list_all()
model_configs = ModelConfigRepository(session).list_all() model_configs = ModelConfigRepository(session).list_all()
scenario_names = {s.id: s.name for s in scenarios}
target_names = {t.id: t.name for t in targets} target_names = {t.id: t.name for t in targets}
settled_runs = _settled(runs) settled_runs = _settled(runs)
@ -81,7 +82,7 @@ def dashboard(session: Session = Depends(get_db)) -> dict:
return { return {
"targets_count": len(targets), "targets_count": len(targets),
"scenarios_count": len(scenario_names), "scenarios_count": len(scenarios),
"runs_count": len(runs), "runs_count": len(runs),
"model_configs_count": len(model_configs), "model_configs_count": len(model_configs),
"today_runs": today_runs, "today_runs": today_runs,

View File

@ -55,9 +55,9 @@ def seeded_db(db_session, monkeypatch):
app.dependency_overrides[get_db] = _test_get_db app.dependency_overrides[get_db] = _test_get_db
# 隔离后台 judge 复核任务API 测试不真正派发后台任务 # 隔离后台 judge 复核任务API 测试不真正派发后台任务
from agenteval.exploration import lifecycle as exploration_lifecycle from agenteval.web.routers import exploration as exploration_router
monkeypatch.setattr(exploration_lifecycle, "start_judge_review", lambda session_id: None) monkeypatch.setattr(exploration_router, "start_judge_review", lambda session_id: None)
target = EvalTarget( target = EvalTarget(
id="t-1", id="t-1",
@ -75,15 +75,15 @@ def seeded_db(db_session, monkeypatch):
def _stub_channel_factory(monkeypatch, channel) -> None: def _stub_channel_factory(monkeypatch, channel) -> None:
"""Point the exploration lifecycle's ChannelFactory at a test channel.""" """Point the exploration router's ChannelFactory at a test channel."""
from agenteval.exploration import lifecycle as lifecycle_module from agenteval.web.routers import exploration as exploration_module
class _StubFactory: class _StubFactory:
@staticmethod @staticmethod
def create(target): def create(target):
return channel return channel
monkeypatch.setattr(lifecycle_module, "ChannelFactory", _StubFactory) monkeypatch.setattr(exploration_module, "ChannelFactory", _StubFactory)
@pytest.fixture() @pytest.fixture()
@ -178,7 +178,6 @@ async def test_full_lifecycle_create_message_close(seeded_db, mock_channel, clie
async def test_dict_reply_content_is_flattened_to_text(seeded_db, monkeypatch, client): async def test_dict_reply_content_is_flattened_to_text(seeded_db, monkeypatch, client):
"""通道回复 content 为对象(如 tutu msgBody时应提取文本而非存 str(dict)。""" """通道回复 content 为对象(如 tutu msgBody时应提取文本而非存 str(dict)。"""
from agenteval.channels.base import Reply from agenteval.channels.base import Reply
from tests.unit.mock_channel import MockChannel from tests.unit.mock_channel import MockChannel
class _DictReplyChannel(MockChannel): class _DictReplyChannel(MockChannel):
@ -322,7 +321,7 @@ async def test_poll_timeout_consumes_turn_budget(seeded_db, monkeypatch, client)
"""消息已送达但等不到回复:账本仍计一轮(超时不可绕过轮数预算)。""" """消息已送达但等不到回复:账本仍计一轮(超时不可绕过轮数预算)。"""
from types import SimpleNamespace from types import SimpleNamespace
from agenteval.exploration import lifecycle as exploration_module from agenteval.web.routers import exploration as exploration_module
from tests.unit.mock_channel import MockChannel from tests.unit.mock_channel import MockChannel
@ -368,10 +367,10 @@ async def test_close_twice_rejected(seeded_db, mock_channel, client):
async def test_close_triggers_judge_review(seeded_db, mock_channel, client, monkeypatch): async def test_close_triggers_judge_review(seeded_db, mock_channel, client, monkeypatch):
from agenteval.exploration import lifecycle as exploration_lifecycle from agenteval.web.routers import exploration as exploration_router
started: list[str] = [] started: list[str] = []
monkeypatch.setattr(exploration_lifecycle, "start_judge_review", started.append) monkeypatch.setattr(exploration_router, "start_judge_review", started.append)
session_id = (await _create_session(client)).json()["id"] session_id = (await _create_session(client)).json()["id"]
resp = await client.post( resp = await client.post(

View File

@ -48,10 +48,10 @@ def seeded_db(db_session, monkeypatch):
target + scenario needed by both the loop path and the cancel path.""" target + scenario needed by both the loop path and the cancel path."""
from agenteval.channels import factory as factory_module from agenteval.channels import factory as factory_module
from agenteval.evaluation import engine as engine_module from agenteval.evaluation import engine as engine_module
from agenteval.exploration import lifecycle as exploration_lifecycle
from agenteval.storage import db as db_module from agenteval.storage import db as db_module
from agenteval.storage import repository as repo_module from agenteval.storage import repository as repo_module
from agenteval.web import app as app_module from agenteval.web import app as app_module
from agenteval.web.routers import exploration as exploration_router
monkeypatch.setattr(app_module, "init_db", lambda: None) monkeypatch.setattr(app_module, "init_db", lambda: None)
@ -68,7 +68,7 @@ def seeded_db(db_session, monkeypatch):
channel = MockChannel(reply_delay=0.0) channel = MockChannel(reply_delay=0.0)
monkeypatch.setattr(factory_module.ChannelFactory, "create", lambda target: channel) monkeypatch.setattr(factory_module.ChannelFactory, "create", lambda target: channel)
monkeypatch.setattr(exploration_lifecycle, "start_judge_review", lambda session_id: None) monkeypatch.setattr(exploration_router, "start_judge_review", lambda session_id: None)
from agenteval.web.deps import get_db from agenteval.web.deps import get_db

View File

@ -1,102 +0,0 @@
"""CampaignRepository 窄口径原子更新直测(架构保养候选 1
全行 update 逼着调用方承诺持有最新完整快照巡检水位 / 取消 /
调度循环并发写入会互相覆盖三个窄方法各自只动自己的列
touch_patrol_watermark水位mark_cancelled终态迁移
save_scheduler_state调度进度
"""
from datetime import datetime, timezone
import pytest
from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, CampaignSummary, SchedulerState
from agenteval.storage.db import as_utc
from agenteval.storage.repository import CampaignRepository
from sqlmodel import Session, SQLModel, create_engine
T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
T1 = datetime(2026, 1, 1, 1, 0, 0, tzinfo=timezone.utc)
@pytest.fixture()
def db_session(tmp_path):
from agenteval.storage.db import ( # noqa: F401
CampaignDB,
EvalResultDB,
EvalRunDB,
EvalTargetDB,
FileCategoryDB,
FileRecordDB,
ScenarioDB,
TurnDB,
)
engine = create_engine(
f"sqlite:///{tmp_path / 'narrow.db'}",
connect_args={"check_same_thread": False},
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
try:
yield session
finally:
session.close()
engine.dispose()
def _seed(db_session, status=CampaignStatus.RUNNING) -> Campaign:
campaign = CampaignRepository(db_session).create(Campaign(
name="cycle", target_id="t-1", window_seconds=3600,
plan=[CampaignPlanEntry(scenario_id="s-a", offset_seconds=0, count=1)],
status=status, started_at=T0,
summary=CampaignSummary(scheduler=SchedulerState(spawned_indices=[0])),
))
return campaign
def test_touch_patrol_watermark_only_touches_watermark(db_session):
campaign = _seed(db_session)
CampaignRepository(db_session).touch_patrol_watermark(campaign.id, T1)
fresh = CampaignRepository(db_session).get(campaign.id)
assert as_utc(fresh.last_patrolled_at) == T1
# 其余列原样:窄口径不覆写并发写入的 status / summary
assert fresh.status == CampaignStatus.RUNNING
assert fresh.summary.scheduler.spawned_indices == [0]
assert fresh.name == "cycle" and as_utc(fresh.started_at) == T0
def test_mark_cancelled_only_touches_status_and_completed_at(db_session):
campaign = _seed(db_session)
CampaignRepository(db_session).touch_patrol_watermark(campaign.id, T1)
updated = CampaignRepository(db_session).mark_cancelled(campaign.id, T1)
assert updated.status == CampaignStatus.CANCELLED
assert as_utc(updated.completed_at) == T1
fresh = CampaignRepository(db_session).get(campaign.id)
assert fresh.status == CampaignStatus.CANCELLED
assert as_utc(fresh.completed_at) == T1
# 水位与调度进度不被终态迁移抹掉
assert as_utc(fresh.last_patrolled_at) == T1
assert fresh.summary.scheduler.spawned_indices == [0]
def test_save_scheduler_state_only_touches_summary(db_session):
campaign = _seed(db_session)
CampaignRepository(db_session).touch_patrol_watermark(campaign.id, T1)
new_summary = CampaignSummary(scheduler=SchedulerState(spawned_indices=[0, 1, 2]))
CampaignRepository(db_session).save_scheduler_state(campaign.id, new_summary)
fresh = CampaignRepository(db_session).get(campaign.id)
assert fresh.summary.scheduler.spawned_indices == [0, 1, 2]
# 状态与水位不被调度进度覆写
assert fresh.status == CampaignStatus.RUNNING
assert as_utc(fresh.last_patrolled_at) == T1
def test_narrow_updates_ignore_missing_campaign(db_session):
repo = CampaignRepository(db_session)
repo.touch_patrol_watermark("missing", T1)
repo.save_scheduler_state("missing", CampaignSummary())
assert repo.mark_cancelled("missing", T1) is None

View File

@ -1,96 +0,0 @@
"""活动报告读模型 load_campaign_report 与 ScenarioRepository.name_map 直测。
读模型是报告 / 分析 / 对比 / 导出共用的取数 + 聚合单一出口
返回结果必须与手写三件套list_by_campaign + scenario_names + generate_campaign_report完全一致
"""
from datetime import datetime, timedelta, timezone
import pytest
from agenteval.evaluation.report import generate_campaign_report, load_campaign_report
from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, Case, CaseType, EvalRun, RunStatus, Scenario
from agenteval.storage.repository import CampaignRepository, RunRepository, ScenarioRepository
from sqlmodel import Session, SQLModel, create_engine
T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
@pytest.fixture()
def db_session(tmp_path):
from agenteval.storage.db import ( # noqa: F401
CampaignDB,
EvalResultDB,
EvalRunDB,
EvalTargetDB,
FileCategoryDB,
FileRecordDB,
ScenarioDB,
TurnDB,
)
engine = create_engine(
f"sqlite:///{tmp_path / 'loader.db'}",
connect_args={"check_same_thread": False},
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
try:
yield session
finally:
session.close()
engine.dispose()
def _seed(session) -> Campaign:
ScenarioRepository(session).create(Scenario(
id="s-a", name="夜间问诊",
cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
))
campaign = CampaignRepository(session).create(Campaign(
name="cycle", target_id="t-1", window_seconds=12, time_scale=1.0,
plan=[CampaignPlanEntry(scenario_id="s-a", offset_seconds=0, count=1)],
))
campaign.status = CampaignStatus.RUNNING
campaign.started_at = T0
campaign = CampaignRepository(session).update(campaign)
RunRepository(session).create(EvalRun(
target_id="t-1", scenario_id="s-a", campaign_id=campaign.id,
status=RunStatus.COMPLETED, started_at=T0 + timedelta(seconds=1),
summary={"total_cases": 1, "passed_cases": 1, "pass_rate": 1.0, "avg_latency_ms": 100},
))
return campaign
def test_load_campaign_report_matches_manual_trio(db_session):
campaign = _seed(db_session)
report = load_campaign_report(db_session, campaign)
runs = RunRepository(db_session).list_by_campaign(campaign.id)
scenario_names = {s.id: s.name for s in ScenarioRepository(db_session).list_all()}
expected = generate_campaign_report(campaign, runs, scenario_names=scenario_names)
assert report == expected
cap = next(c for c in report["capability_summary"] if c["scenario_id"] == "s-a")
assert cap["scenario_name"] == "夜间问诊"
def test_load_campaign_report_empty_campaign(db_session):
campaign = CampaignRepository(db_session).create(Campaign(
name="empty", target_id="t-1", window_seconds=12,
plan=[CampaignPlanEntry(scenario_id="s-x", offset_seconds=0, count=1)],
))
report = load_campaign_report(db_session, campaign)
assert report["summary"]["total_runs"] == 0
assert report["capability_summary"] == []
def test_scenario_name_map(db_session):
_seed(db_session)
ScenarioRepository(db_session).create(Scenario(
id="s-b", name="缴费引导",
cases=[Case(id="c2", type=CaseType.SINGLE, messages=["缴费"])],
))
assert ScenarioRepository(db_session).name_map() == {"s-a": "夜间问诊", "s-b": "缴费引导"}
def test_scenario_name_map_empty(db_session):
assert ScenarioRepository(db_session).name_map() == {}

View File

@ -1,218 +0,0 @@
"""探索生命周期领域模块直测(架构保养候选 3
账本规则与状态机从 HTTP 层落入 exploration/lifecycle.py patrol.py
用类型化领域异常NotFound / Guardrail / Channel表达违规
脱离 TestClient 即可单测
"""
from datetime import datetime, timedelta, timezone
from uuid import uuid4
import pytest
from agenteval.exploration import lifecycle, patrol
from agenteval.exploration.errors import ExplorationGuardrailError, ExplorationNotFoundError
from agenteval.exploration.models import ExplorationSessionStatus, ExplorationTrigger
from agenteval.models import (
Campaign,
CampaignPlanEntry,
CampaignStatus,
EvalRun,
EvalTarget,
ExplorationBudgetConfig,
ExplorationSeeds,
RunStatus,
)
from agenteval.storage.repository import (
CampaignRepository,
ExplorationSessionRepository,
RunRepository,
TargetRepository,
)
from sqlmodel import Session, SQLModel, create_engine
T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
@pytest.fixture()
def db_session(tmp_path):
from agenteval.storage.db import ( # noqa: F401
CampaignDB,
EvalResultDB,
EvalRunDB,
EvalTargetDB,
ExplorationMessageDB,
ExplorationSessionDB,
FileCategoryDB,
FileRecordDB,
ScenarioDB,
TurnDB,
)
engine = create_engine(
f"sqlite:///{tmp_path / 'lifecycle.db'}",
connect_args={"check_same_thread": False},
)
SQLModel.metadata.create_all(engine)
session = Session(engine)
try:
yield session
finally:
session.close()
engine.dispose()
def _seed_campaign(
db_session,
*,
status=CampaignStatus.RUNNING,
time_scale=1.0,
budget=None,
seeds=None,
) -> Campaign:
target = TargetRepository(db_session).create(EvalTarget(id=f"t-{uuid4().hex[:8]}", name="数字员工"))
campaign = CampaignRepository(db_session).create(Campaign(
name="cycle", target_id=target.id, window_seconds=3600, time_scale=time_scale,
plan=[CampaignPlanEntry(scenario_id="s-a", offset_seconds=0, count=1)],
status=status, started_at=T0,
exploration_budget=budget, exploration_seeds=seeds,
))
return campaign
def _open(db_session, campaign, triggered_by=ExplorationTrigger.AUTO, goal="查询账单"):
return lifecycle.open_session(
db_session,
campaign_id=campaign.id,
persona={"name": "急性子用户"},
goal=goal,
triggered_by=triggered_by,
)
# ── 账本四规则 ────────────────────────────────────────────────────────────
def test_open_rejects_non_running_campaign(db_session):
campaign = _seed_campaign(db_session, status=CampaignStatus.PLANNED)
with pytest.raises(ExplorationGuardrailError, match="活动不在进行中"):
_open(db_session, campaign)
def test_open_rejects_auto_on_accelerated_line_but_allows_manual(db_session):
campaign = _seed_campaign(db_session, time_scale=10.0)
with pytest.raises(ExplorationGuardrailError, match="加速调试线仅允许手动"):
_open(db_session, campaign)
session_obj = _open(db_session, campaign, triggered_by=ExplorationTrigger.MANUAL)
assert session_obj.status == ExplorationSessionStatus.RUNNING
def test_open_enforces_max_sessions_budget(db_session):
campaign = _seed_campaign(db_session, budget=ExplorationBudgetConfig(max_sessions=1))
_open(db_session, campaign, triggered_by=ExplorationTrigger.MANUAL)
with pytest.raises(ExplorationGuardrailError, match="探索会话数超出预算"):
_open(db_session, campaign, triggered_by=ExplorationTrigger.MANUAL)
def test_open_enforces_min_interval(db_session):
campaign = _seed_campaign(db_session) # 默认间隔 30 分钟
_open(db_session, campaign)
with pytest.raises(ExplorationGuardrailError, match="相邻探索会话间隔不足"):
_open(db_session, campaign)
def test_open_missing_campaign_or_target_raises_not_found(db_session):
with pytest.raises(ExplorationNotFoundError):
lifecycle.open_session(
db_session, campaign_id="missing", persona={}, goal="x",
)
campaign = _seed_campaign(db_session)
campaign.target_id = "missing-target"
CampaignRepository(db_session).update(campaign)
with pytest.raises(ExplorationNotFoundError):
lifecycle.open_session(
db_session, campaign_id=campaign.id, persona={}, goal="x",
)
# ── 关闭状态机 ────────────────────────────────────────────────────────────
def test_close_completes_session_and_normalizes_experience(db_session, monkeypatch):
judged: list[str] = []
monkeypatch.setattr(lifecycle, "start_judge_review", lambda sid: judged.append(sid))
campaign = _seed_campaign(db_session)
session_obj = _open(db_session, campaign)
updated = lifecycle.close_session(
db_session, session_id=session_obj.id,
experience={"goal_achieved": True, "blockers": ["入口难找"], "emotion": "weird"},
)
assert updated.status == ExplorationSessionStatus.COMPLETED
assert updated.closed_at is not None
assert updated.experience["emotion"] == "neutral" # 非法情绪归一
assert updated.experience["blockers"] == ["入口难找"]
assert judged == [session_obj.id]
def test_close_rejects_non_running_session(db_session, monkeypatch):
monkeypatch.setattr(lifecycle, "start_judge_review", lambda sid: None)
campaign = _seed_campaign(db_session)
session_obj = _open(db_session, campaign)
lifecycle.close_session(db_session, session_id=session_obj.id, experience={"goal_achieved": False})
with pytest.raises(ExplorationGuardrailError, match="不在进行中"):
lifecycle.close_session(db_session, session_id=session_obj.id, experience={"goal_achieved": False})
def test_close_missing_session_raises_not_found(db_session):
with pytest.raises(ExplorationNotFoundError):
lifecycle.close_session(db_session, session_id="missing", experience={})
# ── 对话轮账本(拒收路径;通道往返由集成测试覆盖)────────────────────────
async def test_turn_rejects_non_running_session(db_session):
campaign = _seed_campaign(db_session)
session_obj = _open(db_session, campaign)
lifecycle.close_session(db_session, session_id=session_obj.id, experience={"goal_achieved": True})
with pytest.raises(ExplorationGuardrailError, match="拒收消息"):
await lifecycle.conduct_turn(db_session, session_id=session_obj.id, content="你好")
async def test_turn_enforces_turn_budget(db_session):
campaign = _seed_campaign(db_session, budget=ExplorationBudgetConfig(max_turns=1, min_interval_seconds=1))
session_obj = _open(db_session, campaign)
repo = ExplorationSessionRepository(db_session)
session_obj.turn_count = 1
repo.update(session_obj)
with pytest.raises(ExplorationGuardrailError, match="会话轮数超出预算"):
await lifecycle.conduct_turn(db_session, session_id=session_obj.id, content="你好")
async def test_turn_missing_session_raises_not_found(db_session):
with pytest.raises(ExplorationNotFoundError):
await lifecycle.conduct_turn(db_session, session_id="missing", content="你好")
# ── 巡检:筛选 + 增量 + 水位 ──────────────────────────────────────────────
def test_patrol_filters_and_advances_watermark(db_session):
eligible = _seed_campaign(db_session, seeds=ExplorationSeeds(personas=["急性子用户"], goals=["查账单"]))
# 不符合巡检条件:加速线 / 无种子集
_seed_campaign(db_session, time_scale=10.0, seeds=ExplorationSeeds(personas=["慢用户"], goals=["y"]))
RunRepository(db_session).create(EvalRun(
target_id="t-1", scenario_id="s-a", campaign_id=eligible.id,
status=RunStatus.COMPLETED, started_at=T0, completed_at=T0 + timedelta(seconds=10),
summary={"total_cases": 1, "passed_cases": 1, "pass_rate": 1.0, "avg_latency_ms": 100},
))
first = patrol.patrol_report(db_session)
assert [e["campaign_id"] for e in first["campaigns"]] == [eligible.id]
assert first["campaigns"][0]["new_results"] is not None
# 水位推进后无增量;且只动水位列(状态不被覆写)
second = patrol.patrol_report(db_session)
assert second["campaigns"][0]["new_results"] is None
fresh = CampaignRepository(db_session).get(eligible.id)
assert fresh.last_patrolled_at is not None
assert fresh.status == CampaignStatus.RUNNING