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 精神，等真实需求出现再议）。
433 lines
16 KiB
Python
433 lines
16 KiB
Python
"""Durable runtime for Campaign intelligence jobs.
|
||
|
||
智能分析与周期对比是两个领域工作 adapter；本 module 的 ``execute``
|
||
统一掌握它们的结算契约：建行/认领/校验/generating/failed/completed。
|
||
数据库行是耐久权威，TaskRegistry 只保存当前进程中的任务句柄。
|
||
"""
|
||
|
||
import logging
|
||
from collections.abc import Awaitable, Callable
|
||
from typing import Any, Optional
|
||
|
||
from sqlmodel import Session
|
||
|
||
from agenteval.storage.async_job_repository import CampaignAnalysisRepository, CampaignPeriodComparisonRepository
|
||
from agenteval.storage.db import get_session
|
||
from agenteval.task_registry import TaskRegistry
|
||
|
||
_registry = TaskRegistry()
|
||
_logger = logging.getLogger("agenteval")
|
||
MAX_QUEUED_RECOVERY_ATTEMPTS = 3
|
||
_ERROR_TRUNCATE_LEN = 500
|
||
|
||
# adapter 契约：
|
||
# ensure_queued(session) -> row | None（建行或确认已有行；None = 静默放弃）
|
||
# validate(session) -> 落账附加字段（认领后校验，失败抛 JobValidationError）
|
||
# work_fn(session) -> (result, 落账附加字段)
|
||
EnsureQueuedFn = Callable[[Session], Any]
|
||
ValidateFn = Callable[[Session], Optional[dict]]
|
||
WorkFn = Callable[[Session], Awaitable[tuple[dict, dict]]]
|
||
|
||
|
||
class JobValidationError(RuntimeError):
|
||
"""认领后校验失败：携带已确定的落账字段（如对比的生效基线）进 failed 行。"""
|
||
|
||
def __init__(self, message: str, **meta: Any):
|
||
super().__init__(message)
|
||
self.meta = meta
|
||
|
||
|
||
def _job_key(kind: str, campaign_id: str) -> str:
|
||
return f"{kind}:{campaign_id}"
|
||
|
||
|
||
def _launch_analysis(campaign_id: str, *, triggered_by: str) -> None:
|
||
async def run(_cancel) -> None:
|
||
await execute_campaign_analysis_job(campaign_id, triggered_by=triggered_by)
|
||
|
||
_registry.launch(_job_key("analysis", campaign_id), run)
|
||
|
||
|
||
def _launch_comparison(
|
||
campaign_id: str,
|
||
*,
|
||
triggered_by: str,
|
||
baseline_campaign_id: str,
|
||
) -> None:
|
||
async def run(_cancel) -> None:
|
||
await execute_campaign_comparison_job(
|
||
campaign_id,
|
||
triggered_by=triggered_by,
|
||
baseline_campaign_id=baseline_campaign_id,
|
||
)
|
||
|
||
_registry.launch(_job_key("comparison", campaign_id), run)
|
||
|
||
|
||
async def execute(
|
||
job_kind: str,
|
||
campaign_id: str,
|
||
*,
|
||
triggered_by: str,
|
||
repo_cls,
|
||
ensure_queued: EnsureQueuedFn,
|
||
validate: Optional[ValidateFn] = None,
|
||
work_fn: WorkFn,
|
||
session_factory: Optional[Callable[[], Session]] = None,
|
||
) -> None:
|
||
"""结算一个耐久智能作业：建行 → 认领 → 校验 → generating → 落账。
|
||
|
||
结算契约（两个 adapter 共用，只此一处）：
|
||
|
||
- ``ensure_queued`` 幂等建行（或确认已有行）；返回 None 表示静默
|
||
放弃（如活动不存在），不写任何行。
|
||
- ``claim_queued`` 是耐久幂等权威：抢不到（重复触发 / 已在执行）
|
||
静默返回，绝不重复跑领域工作。
|
||
- ``validate`` 在认领后校验并产出附加字段（模型、基线），随
|
||
generating/结算落账；抛 ``JobValidationError`` → 落 failed 行
|
||
（异常携带的已确定字段一并落账）。
|
||
- ``work_fn`` 抛任何异常 → failed 行，error 截断到 500 字符（一处）。
|
||
"""
|
||
session = (session_factory or get_session)()
|
||
repo = repo_cls(session)
|
||
try:
|
||
row = ensure_queued(session)
|
||
if row is None:
|
||
return
|
||
if not repo.claim_queued(campaign_id).claimed:
|
||
return
|
||
effective_trigger = row.triggered_by or triggered_by
|
||
|
||
meta: dict[str, Any] = {}
|
||
if validate is not None:
|
||
try:
|
||
meta = validate(session) or {}
|
||
except JobValidationError as exc:
|
||
repo.upsert(
|
||
campaign_id,
|
||
status="failed",
|
||
triggered_by=effective_trigger,
|
||
error=str(exc),
|
||
**exc.meta,
|
||
)
|
||
return
|
||
|
||
repo.upsert(campaign_id, status="generating", triggered_by=effective_trigger, **meta)
|
||
work_meta: dict[str, Any] = {}
|
||
try:
|
||
result, work_meta = await work_fn(session)
|
||
except Exception as exc:
|
||
_logger.warning("%s 作业 %s 失败: %s", job_kind, campaign_id, exc)
|
||
repo.upsert(
|
||
campaign_id,
|
||
status="failed",
|
||
error=str(exc)[:_ERROR_TRUNCATE_LEN],
|
||
triggered_by=effective_trigger,
|
||
**meta,
|
||
**work_meta,
|
||
)
|
||
return
|
||
repo.upsert(
|
||
campaign_id,
|
||
status="completed",
|
||
result=result,
|
||
triggered_by=effective_trigger,
|
||
**meta,
|
||
**work_meta,
|
||
)
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
async def execute_campaign_analysis_job(
|
||
campaign_id: str,
|
||
*,
|
||
triggered_by: str,
|
||
chat_client: Any = None,
|
||
session_factory: Optional[Callable[[], Session]] = None,
|
||
) -> None:
|
||
"""Claim and settle one intelligent-analysis job (work adapter)."""
|
||
|
||
def ensure_queued(session: Session):
|
||
analyses = CampaignAnalysisRepository(session)
|
||
row = analyses.get_by_campaign(campaign_id)
|
||
return row or analyses.enqueue(campaign_id, triggered_by=triggered_by)
|
||
|
||
def validate(session: Session) -> dict:
|
||
from agenteval.evaluation.analysis import AnalysisError, validate_analysis_request
|
||
from agenteval.storage.repository import CampaignRepository
|
||
|
||
campaign = CampaignRepository(session).get(campaign_id)
|
||
if campaign is None:
|
||
raise JobValidationError("campaign not found")
|
||
try:
|
||
runtime = validate_analysis_request(session, campaign)
|
||
except AnalysisError as exc:
|
||
raise JobValidationError(str(exc)) from exc
|
||
return {"model_config_id": runtime.id}
|
||
|
||
async def work(session: Session) -> tuple[dict, dict]:
|
||
from agenteval.evaluation.analysis import (
|
||
analyze_campaign,
|
||
collect_failure_samples,
|
||
gateway_chat_client,
|
||
resolve_analysis_model,
|
||
)
|
||
from agenteval.evaluation.report import load_campaign_view
|
||
from agenteval.storage.repository import CampaignRepository, RunRepository
|
||
|
||
campaign = CampaignRepository(session).get(campaign_id)
|
||
runtime = resolve_analysis_model(campaign, session)
|
||
runs = RunRepository(session).list_by_campaign(campaign_id)
|
||
view = load_campaign_view(session, campaign)
|
||
result = await analyze_campaign(
|
||
campaign=campaign,
|
||
report=view["report"],
|
||
failure_samples=collect_failure_samples(campaign_id, session),
|
||
valid_run_ids={run.id for run in runs if run.id},
|
||
chat_client=chat_client or gateway_chat_client(runtime),
|
||
exploration_summary=view["exploration"],
|
||
)
|
||
return result, {}
|
||
|
||
await execute(
|
||
"analysis",
|
||
campaign_id,
|
||
triggered_by=triggered_by,
|
||
repo_cls=CampaignAnalysisRepository,
|
||
ensure_queued=ensure_queued,
|
||
validate=validate,
|
||
work_fn=work,
|
||
session_factory=session_factory,
|
||
)
|
||
await _maybe_enqueue_auto_comparison(campaign_id, session_factory=session_factory)
|
||
|
||
|
||
def auto_intelligence_eligible(campaign, session: Session) -> bool:
|
||
"""自动触发智能分析/周期对比的唯一跳过守卫：正式线 + 分析模型可解析。
|
||
|
||
活动完成时的自动分析与分析完成后的自动对比共用这一判断点；
|
||
加速线与未配置模型一律静默跳过（增强能力，不阻断主流程）。
|
||
"""
|
||
from agenteval.evaluation.analysis import resolve_analysis_model
|
||
|
||
return bool(campaign.id) and campaign.time_scale == 1 and resolve_analysis_model(campaign, session) is not None
|
||
|
||
|
||
async def _maybe_enqueue_auto_comparison(
|
||
campaign_id: str,
|
||
*,
|
||
session_factory: Optional[Callable[[], Session]] = None,
|
||
) -> None:
|
||
"""分析完成后：存在自动基线 → 入队周期对比。任何一步不满足或出错都静默跳过。"""
|
||
from agenteval.evaluation.comparison import resolve_auto_baseline
|
||
from agenteval.storage.repository import CampaignRepository
|
||
|
||
session = (session_factory or get_session)()
|
||
try:
|
||
campaign = CampaignRepository(session).get(campaign_id)
|
||
if campaign is None or not auto_intelligence_eligible(campaign, session):
|
||
return
|
||
row = CampaignAnalysisRepository(session).get_by_campaign(campaign_id)
|
||
if row is None or row.status != "completed":
|
||
return
|
||
baseline = resolve_auto_baseline(campaign, session)
|
||
if baseline is None:
|
||
return
|
||
enqueue_campaign_comparison(
|
||
campaign.id,
|
||
triggered_by="auto",
|
||
baseline_campaign_id=baseline.id,
|
||
session=session,
|
||
)
|
||
except Exception as exc:
|
||
_logger.warning("活动 %s 自动周期对比跳过: %s", campaign_id, exc)
|
||
finally:
|
||
session.close()
|
||
|
||
|
||
async def execute_campaign_comparison_job(
|
||
campaign_id: str,
|
||
*,
|
||
triggered_by: str,
|
||
baseline_campaign_id: Optional[str] = None,
|
||
chat_client: Any = None,
|
||
session_factory: Optional[Callable[[], Session]] = None,
|
||
) -> None:
|
||
"""Claim and settle one period-comparison job (work adapter)."""
|
||
from agenteval.evaluation.comparison import ComparisonError
|
||
|
||
def ensure_queued(session: Session):
|
||
from agenteval.evaluation.comparison import validate_comparison_request
|
||
from agenteval.storage.repository import CampaignRepository
|
||
|
||
comparisons = CampaignPeriodComparisonRepository(session)
|
||
campaign = CampaignRepository(session).get(campaign_id)
|
||
if campaign is None:
|
||
return None
|
||
row = comparisons.get_by_campaign(campaign_id)
|
||
if row is not None:
|
||
return row
|
||
try:
|
||
initial_baseline = validate_comparison_request(
|
||
session,
|
||
campaign,
|
||
explicit_baseline_id=baseline_campaign_id,
|
||
)
|
||
except ComparisonError as exc:
|
||
comparisons.upsert(
|
||
campaign_id,
|
||
status="failed",
|
||
triggered_by=triggered_by,
|
||
error=str(exc),
|
||
)
|
||
return None
|
||
return comparisons.enqueue(
|
||
campaign_id,
|
||
baseline_campaign_id=initial_baseline.id,
|
||
triggered_by=triggered_by,
|
||
)
|
||
|
||
def validate(session: Session) -> dict:
|
||
from agenteval.evaluation.analysis import resolve_analysis_model
|
||
from agenteval.evaluation.comparison import (
|
||
ComparisonError,
|
||
validate_comparison_request,
|
||
)
|
||
from agenteval.storage.repository import CampaignRepository
|
||
|
||
campaign = CampaignRepository(session).get(campaign_id)
|
||
row = CampaignPeriodComparisonRepository(session).get_by_campaign(campaign_id)
|
||
effective_baseline_id = baseline_campaign_id or (row.baseline_campaign_id if row else None)
|
||
try:
|
||
baseline = validate_comparison_request(
|
||
session,
|
||
campaign,
|
||
explicit_baseline_id=effective_baseline_id,
|
||
)
|
||
except ComparisonError as exc:
|
||
raise JobValidationError(str(exc), baseline_campaign_id=effective_baseline_id) from exc
|
||
runtime = resolve_analysis_model(campaign, session)
|
||
return {"baseline_campaign_id": baseline.id, "model_config_id": runtime.id}
|
||
|
||
async def work(session: Session) -> tuple[dict, dict]:
|
||
from agenteval.evaluation.analysis import gateway_chat_client, resolve_analysis_model
|
||
from agenteval.evaluation.comparison import compute_metric_diff, narrate_period_comparison
|
||
from agenteval.evaluation.report import load_campaign_report
|
||
from agenteval.storage.repository import CampaignRepository
|
||
|
||
campaign = CampaignRepository(session).get(campaign_id)
|
||
row = CampaignPeriodComparisonRepository(session).get_by_campaign(campaign_id)
|
||
effective_baseline_id = baseline_campaign_id or (row.baseline_campaign_id if row else None)
|
||
baseline = CampaignRepository(session).get(effective_baseline_id)
|
||
runtime = resolve_analysis_model(campaign, session)
|
||
baseline_analysis = CampaignAnalysisRepository(session).get_by_campaign(baseline.id)
|
||
current_analysis = CampaignAnalysisRepository(session).get_by_campaign(campaign_id)
|
||
diff = compute_metric_diff(
|
||
load_campaign_report(session, baseline),
|
||
load_campaign_report(session, campaign),
|
||
)
|
||
result = await narrate_period_comparison(
|
||
baseline_analysis=baseline_analysis.get_result(),
|
||
current_analysis=current_analysis.get_result(),
|
||
metric_diff=diff,
|
||
valid_scenario_ids={item["scenario_id"] for item in diff["scenarios"]},
|
||
chat_client=chat_client or gateway_chat_client(runtime),
|
||
)
|
||
return result, {}
|
||
|
||
await execute(
|
||
"comparison",
|
||
campaign_id,
|
||
triggered_by=triggered_by,
|
||
repo_cls=CampaignPeriodComparisonRepository,
|
||
ensure_queued=ensure_queued,
|
||
validate=validate,
|
||
work_fn=work,
|
||
session_factory=session_factory,
|
||
)
|
||
|
||
|
||
def enqueue_campaign_analysis(
|
||
campaign_id: str,
|
||
*,
|
||
triggered_by: str,
|
||
session: Optional[Session] = None,
|
||
) -> None:
|
||
"""Persist an analysis job, then launch its process-local worker."""
|
||
owns_session = session is None
|
||
active_session = session or get_session()
|
||
try:
|
||
row = CampaignAnalysisRepository(active_session).enqueue(
|
||
campaign_id,
|
||
triggered_by=triggered_by,
|
||
)
|
||
finally:
|
||
if owns_session:
|
||
active_session.close()
|
||
if row.status == "queued":
|
||
_launch_analysis(campaign_id, triggered_by=row.triggered_by or triggered_by)
|
||
|
||
|
||
def enqueue_campaign_comparison(
|
||
campaign_id: str,
|
||
*,
|
||
triggered_by: str,
|
||
baseline_campaign_id: str,
|
||
session: Optional[Session] = None,
|
||
) -> None:
|
||
"""Persist a comparison job, then launch its process-local worker."""
|
||
owns_session = session is None
|
||
active_session = session or get_session()
|
||
try:
|
||
row = CampaignPeriodComparisonRepository(active_session).enqueue(
|
||
campaign_id,
|
||
baseline_campaign_id=baseline_campaign_id,
|
||
triggered_by=triggered_by,
|
||
)
|
||
finally:
|
||
if owns_session:
|
||
active_session.close()
|
||
if row.status == "queued":
|
||
_launch_comparison(
|
||
campaign_id,
|
||
triggered_by=row.triggered_by or triggered_by,
|
||
baseline_campaign_id=row.baseline_campaign_id,
|
||
)
|
||
|
||
|
||
def recover_campaign_intelligence_jobs(session: Session) -> tuple[int, int]:
|
||
"""Fail interrupted work and relaunch every durably queued job."""
|
||
analyses = CampaignAnalysisRepository(session)
|
||
comparisons = CampaignPeriodComparisonRepository(session)
|
||
interrupted = analyses.mark_orphans_failed() + comparisons.mark_orphans_failed()
|
||
|
||
queued_analyses = analyses.prepare_queued_recovery(
|
||
MAX_QUEUED_RECOVERY_ATTEMPTS,
|
||
"服务重启恢复次数超过上限，分析任务已终止",
|
||
)
|
||
for row in queued_analyses:
|
||
_launch_analysis(row.campaign_id, triggered_by=row.triggered_by or "manual")
|
||
|
||
queued_comparisons = comparisons.prepare_queued_recovery(
|
||
MAX_QUEUED_RECOVERY_ATTEMPTS,
|
||
"服务重启恢复次数超过上限，周期对比任务已终止",
|
||
)
|
||
for row in queued_comparisons:
|
||
_launch_comparison(
|
||
row.campaign_id,
|
||
triggered_by=row.triggered_by or "manual",
|
||
baseline_campaign_id=row.baseline_campaign_id,
|
||
)
|
||
return interrupted, len(queued_analyses) + len(queued_comparisons)
|
||
|
||
|
||
def is_intelligence_job_running(kind: str, campaign_id: str) -> bool:
|
||
"""Expose process-local liveness without exposing registry internals."""
|
||
return _registry.is_running(_job_key(kind, campaign_id))
|
||
|
||
|
||
async def shutdown_campaign_intelligence_jobs() -> None:
|
||
"""Stop every live analysis and comparison worker."""
|
||
await _registry.shutdown_all()
|