7eae6de52d
合并两个不可分割的深化： Phase 2 — 智能作业结算统一（ADR-0012） - intelligence_jobs.execute(job_kind, campaign_id, ...) 作为结算的 唯一实现：建行 → 认领 → 校验 → generating → 落账，一处编排、 一处截断（500 字符）。两个 executor 退化为 ensure_queued / validate / work_fn 三个小 adapter。 - analysis.validate_analysis_request() 共享校验入口（活动终态 → 模型），路由捕获映射 400、executor 捕获落 failed 行，与 validate_comparison_request 先例同构。 - campaign_runner._auto_start_analysis 的跳过守卫收敛至 auto_intelligence_eligible 单一判断点。 - comparison.py 删除零调用的 build_comparison_payload； load_comparison_view 投影归位至 campaign_read_model。 - 新增 characterization 测试（认领竞争、重复触发、截断、恢复上限）。 Phase 3 — storage/repository.py 拆分 - AsyncJobRepository 及两个子类迁至 storage/async_job_repository.py（Phase 2 的 intelligence_jobs 与 comparison 必须 import 自该路径，故与 Phase 2 同 commit）。 - ExplorationSession / ExplorationMessage 迁至 storage/exploration_repository.py；repository.py 由 1180 行降至 约 814 行，grep 确认无残留符号。 - exploration 子模块与路由 import 全部更新；测试 import 跟随。 刻意不做：CAS 共享原语、app.py 五 registry 关停顺序归一 （ADR-0006 精神，等真实需求出现再议）。
101 lines
3.9 KiB
Python
101 lines
3.9 KiB
Python
"""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.exploration_repository import ExplorationSessionRepository
|
||
from agenteval.storage.repository import (
|
||
CampaignRepository,
|
||
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}
|