feat(campaigns): auto-trigger analysis and include it in markdown export

The scheduler loop enqueues the analysis task when a realtime campaign
completes; accelerated or cancelled campaigns and a missing analysis
model skip silently. The campaign markdown export appends the analysis
appendix (overall, problems, narratives, suggestions) when a completed
analysis exists.
This commit is contained in:
sinohqb 2026-08-03 02:25:07 +08:00
parent 6d9e49768c
commit 5d04455664
6 changed files with 305 additions and 3 deletions

View File

@ -20,6 +20,7 @@ from typing import Optional
from sqlmodel import Session from sqlmodel import Session
from agenteval.evaluation.analysis import resolve_analysis_model, start_campaign_analysis
from agenteval.evaluation.campaign_scheduler import ( from agenteval.evaluation.campaign_scheduler import (
TickAction, TickAction,
clock_offset, clock_offset,
@ -183,6 +184,23 @@ async def advance_campaign(
return result return result
def _auto_start_analysis(campaign: Campaign, session: Session) -> None:
"""Enqueue 智能分析 when a 正式线 campaign completes.
Accelerated campaigns, an unresolvable analysis model, and any enqueue
failure all skip silently — the analysis is an enhancement and must never
block or break campaign completion.
"""
if campaign.time_scale != 1 or not campaign.id:
return
try:
if resolve_analysis_model(campaign, session) is None:
return
start_campaign_analysis(campaign.id, triggered_by="auto")
except Exception as exc:
_logger.warning("活动 %s 自动分析触发失败(已跳过): %s", campaign.id, exc)
# ── durable scheduler loop ────────────────────────────────────────────────── # ── durable scheduler loop ──────────────────────────────────────────────────
@ -232,6 +250,7 @@ async def run_campaign_loop(
current.status = CampaignStatus.COMPLETED current.status = CampaignStatus.COMPLETED
current.completed_at = utc_now() current.completed_at = utc_now()
repo.update(current) repo.update(current)
_auto_start_analysis(current, session)
return return
try: try:

View File

@ -173,8 +173,57 @@ def render_markdown(report: dict[str, Any]) -> str:
return "\n".join(lines) return "\n".join(lines)
def render_campaign_markdown(report: dict[str, Any]) -> str: _SEVERITY_LABELS = {"high": "高", "medium": "中", "low": "低"}
"""Render a dual-axis campaign report dict as Markdown."""
def _render_analysis_lines(analysis: dict[str, Any], scenario_names: dict[str, str]) -> list[str]:
"""Appendix for the campaign export: the structured analysis blocks."""
def _name(sid: str) -> str:
return scenario_names.get(sid, sid[:8])
lines = ["", "## 智能分析", "", "### 总体结论", "", str(analysis.get("overall") or "—"), ""]
problems = analysis.get("problems") or []
if problems:
lines += ["### 问题诊断", ""]
for p in problems:
severity = _SEVERITY_LABELS.get(p.get("severity"), "中")
names = "、".join(_name(sid) for sid in p.get("scenario_ids") or [])
head = f"- **[{severity}] {p.get('title', '')}**"
if names:
head += f"(场景:{names})"
lines.append(head)
if p.get("description"):
lines.append(f" {p['description']}")
evidence = p.get("evidence_run_ids") or []
if evidence:
lines.append(" 证据 run:" + " ".join(f"`{rid}`" for rid in evidence))
lines.append("")
narratives = analysis.get("scenario_narratives") or []
if narratives:
lines += ["### 分场景叙述", ""]
for n in narratives:
lines.append(f"- **{_name(str(n.get('scenario_id', '')))}**:{n.get('narrative', '')}")
lines.append("")
suggestions = sorted(analysis.get("suggestions") or [], key=lambda s: s.get("priority", 0))
if suggestions:
lines += ["### 改善建议", ""]
lines += [f"{i}. {s.get('text', '')}" for i, s in enumerate(suggestions, 1)]
lines.append("")
return lines
def render_campaign_markdown(
report: dict[str, Any],
*,
analysis: Optional[dict[str, Any]] = None,
scenario_names: Optional[dict[str, str]] = None,
) -> str:
"""Render a dual-axis campaign report dict as Markdown.
``analysis`` is the stored 智能分析 result dict (completed only); when
absent the export is identical to the pre-analysis format.
"""
s = report["summary"] s = report["summary"]
lines: list[str] = [ lines: list[str] = [
f"# 活动周期报告 — {report['name']}", f"# 活动周期报告 — {report['name']}",
@ -216,4 +265,6 @@ def render_campaign_markdown(report: dict[str, Any]) -> str:
f"| {c['scenario_name']} | {c['run_count']} | {_pct(c['pass_rate'])} | " f"| {c['scenario_name']} | {c['run_count']} | {_pct(c['pass_rate'])} | "
f"{_pct(c['availability'])} | {_ms(c['avg_latency_ms'])} |" f"{_pct(c['availability'])} | {_ms(c['avg_latency_ms'])} |"
) )
if analysis:
lines += _render_analysis_lines(analysis, scenario_names or {})
return "\n".join(lines) return "\n".join(lines)

View File

@ -124,7 +124,13 @@ async def get_campaign_report_markdown(campaign_id: str, session: Session = Depe
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 = {s.id: s.name for s in ScenarioRepository(session).list_all()} scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
md = render_campaign_markdown(generate_campaign_report(campaign, runs, scenario_names=scenario_names)) analysis_row = CampaignAnalysisRepository(session).get_by_campaign(campaign_id)
analysis = analysis_row.get_result() if analysis_row and analysis_row.status == "completed" else None
md = render_campaign_markdown(
generate_campaign_report(campaign, runs, scenario_names=scenario_names),
analysis=analysis,
scenario_names=scenario_names,
)
return Response( return Response(
content=md, content=md,
media_type="text/markdown; charset=utf-8", media_type="text/markdown; charset=utf-8",

View File

@ -183,3 +183,30 @@ async def test_rerun_upserts_without_new_row(client, seeded_db, monkeypatch):
async def test_get_analysis_missing_campaign_404(client, seeded_db): async def test_get_analysis_missing_campaign_404(client, seeded_db):
resp = await client.get("/api/campaigns/nope/analysis") resp = await client.get("/api/campaigns/nope/analysis")
assert resp.status_code == 404 assert resp.status_code == 404
async def test_markdown_export_includes_completed_analysis(client, seeded_db):
campaign_id = await _create_campaign(client, seeded_db)
_complete_analysis_row(seeded_db, campaign_id)
resp = await client.get(f"/api/campaigns/{campaign_id}/report/markdown")
assert resp.status_code == 200
assert "## 智能分析" in resp.text
assert "整体达标" in resp.text
assert "表现稳定" in resp.text
async def test_markdown_export_ignores_non_completed_analysis(client, seeded_db):
campaign_id = await _create_campaign(client, seeded_db)
row = CampaignAnalysisDB(campaign_id=campaign_id, status="failed", error="模型超时")
seeded_db.add(row)
seeded_db.commit()
resp = await client.get(f"/api/campaigns/{campaign_id}/report/markdown")
assert resp.status_code == 200
assert "智能分析" not in resp.text
async def test_markdown_export_without_analysis_unchanged(client, seeded_db):
campaign_id = await _create_campaign(client, seeded_db)
resp = await client.get(f"/api/campaigns/{campaign_id}/report/markdown")
assert resp.status_code == 200
assert "智能分析" not in resp.text

View File

@ -0,0 +1,138 @@
"""Integration tests for the runner's auto analysis hook (v0.7 ticket 05).
Only 正式线 campaigns (time_scale == 1) enqueue the analysis task on
COMPLETED; accelerated/cancelled campaigns and missing analysis models all
skip silently. The analysis service itself is spied, not executed.
"""
import pytest
from agenteval.evaluation import campaign_runner
from agenteval.evaluation.campaign_runner import request_cancel, start_campaign
from agenteval.models import (
Campaign,
CampaignPlanEntry,
CampaignStatus,
Case,
CaseType,
ChannelType,
EvalTarget,
PlatformType,
Scenario,
TargetStatus,
)
from agenteval.storage.db import utc_now
from agenteval.storage.repository import CampaignRepository, ScenarioRepository, TargetRepository
from tests.unit.mock_channel import MockChannel
TICK = 0.01
@pytest.fixture()
def seeded_db(db_session, monkeypatch):
from agenteval.channels import factory as factory_module
from agenteval.evaluation import engine as engine_module
from agenteval.storage import db as db_module
from agenteval.storage import repository as repo_module
def _test_get_session():
return db_session
monkeypatch.setattr(db_module, "get_session", _test_get_session)
monkeypatch.setattr(repo_module, "get_session", _test_get_session)
monkeypatch.setattr(engine_module, "get_session", _test_get_session)
monkeypatch.setattr(campaign_runner, "get_session", _test_get_session)
channel = MockChannel(reply_delay=0.0)
monkeypatch.setattr(factory_module.ChannelFactory, "create", lambda target: channel)
TargetRepository(db_session).create(EvalTarget(
id="t-1", name="mock-target",
platform=PlatformType.AI_DIGITAL_EMPLOYEE, channel_type=ChannelType.TUTU_API,
channel_config={"base_url": "http://mock", "token": "x"}, status=TargetStatus.ACTIVE,
))
ScenarioRepository(db_session).create(Scenario(
id="s-1", name="mock-scenario",
cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
))
return db_session
@pytest.fixture()
def analysis_spy(monkeypatch):
"""Spy the analysis seam: resolvable model, recorded enqueue calls."""
calls: list[tuple[str, str]] = []
monkeypatch.setattr(
campaign_runner, "start_campaign_analysis",
lambda cid, *, triggered_by: calls.append((cid, triggered_by)),
)
monkeypatch.setattr(campaign_runner, "resolve_analysis_model", lambda campaign, session: object())
return calls
def _make_campaign(session, **overrides) -> Campaign:
payload = dict(
name="auto-trigger",
target_id="t-1",
window_seconds=1,
time_scale=1.0, # 正式线:1s 窗口真实耗时 ~1s
plan=[CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=1)],
)
payload.update(overrides)
return CampaignRepository(session).create(Campaign(**payload))
async def _await_task(campaign_id, timeout=5.0):
import asyncio
task = campaign_runner.campaign_registry.get(campaign_id)
if task is not None:
await asyncio.wait_for(task, timeout=timeout)
async def test_realtime_completion_auto_enqueues_analysis(seeded_db, analysis_spy):
campaign = _make_campaign(seeded_db)
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
await _await_task(campaign.id)
final = CampaignRepository(seeded_db).get(campaign.id)
assert final.status == CampaignStatus.COMPLETED
assert analysis_spy == [(campaign.id, "auto")]
async def test_accelerated_completion_does_not_enqueue(seeded_db, analysis_spy):
campaign = _make_campaign(seeded_db, time_scale=1000.0) # 加速调试线
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
await _await_task(campaign.id)
final = CampaignRepository(seeded_db).get(campaign.id)
assert final.status == CampaignStatus.COMPLETED
assert analysis_spy == []
async def test_cancelled_campaign_does_not_enqueue(seeded_db, analysis_spy):
import asyncio
campaign = _make_campaign(seeded_db, window_seconds=100)
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
await asyncio.sleep(0.05)
repo = CampaignRepository(seeded_db)
current = repo.get(campaign.id)
current.status = CampaignStatus.CANCELLED
current.completed_at = utc_now()
repo.update(current)
request_cancel(campaign.id)
await _await_task(campaign.id)
assert repo.get(campaign.id).status == CampaignStatus.CANCELLED
assert analysis_spy == []
async def test_missing_analysis_model_skips_silently(seeded_db, monkeypatch, analysis_spy):
monkeypatch.setattr(campaign_runner, "resolve_analysis_model", lambda campaign, session: None)
campaign = _make_campaign(seeded_db)
start_campaign(campaign.id, seeded_db, tick_seconds=TICK)
await _await_task(campaign.id)
final = CampaignRepository(seeded_db).get(campaign.id)
assert final.status == CampaignStatus.COMPLETED # 活动完成流程不受影响
assert analysis_spy == []

View File

@ -185,3 +185,64 @@ def test_render_campaign_markdown_empty_bucket_dashes():
md = render_campaign_markdown(_campaign_report()) md = render_campaign_markdown(_campaign_report())
# bucket 1 has no runs: pass_rate/availability/latency all render as — # bucket 1 has no runs: pass_rate/availability/latency all render as —
assert "| 3600–7200 | 0 | — | — | — |" in md assert "| 3600–7200 | 0 | — | — | — |" in md
def _analysis() -> dict:
"""A hand-built dict matching the stored campaign analysis result shape."""
return {
"overall": "整窗通过率偏低,售后场景拖后腿",
"problems": [
{
"severity": "high",
"title": "售后答非所问",
"description": "多轮对话中反复偏离用户问题",
"scenario_ids": ["s-1"],
"evidence_run_ids": ["run-abc", "run-def"],
},
{
"severity": "low",
"title": "响应偏慢",
"description": "高峰时段时延偏高",
"scenario_ids": ["s-2"],
"evidence_run_ids": [],
},
],
"scenario_narratives": [
{"scenario_id": "s-1", "narrative": "售后场景表现不稳定"},
{"scenario_id": "s-2", "narrative": "售前场景表现稳定"},
],
"suggestions": [
{"priority": 2, "text": "次要建议:扩容"},
{"priority": 1, "text": "首要建议:补充售后知识库"},
],
}
def test_render_campaign_markdown_appends_analysis_sections():
md = render_campaign_markdown(
_campaign_report(),
analysis=_analysis(),
scenario_names={"s-1": "售后场景", "s-2": "售前场景"},
)
assert "## 智能分析" in md
assert "### 总体结论" in md
assert "整窗通过率偏低,售后场景拖后腿" in md
assert "### 问题诊断" in md
assert "**[高] 售后答非所问**(场景:售后场景)" in md
assert "`run-abc`" in md and "`run-def`" in md
assert "**[低] 响应偏慢**(场景:售前场景)" in md
assert "### 分场景叙述" in md
assert "**售后场景**:售后场景表现不稳定" in md
assert "### 改善建议" in md
# 建议按 priority 升序
assert md.index("首要建议") < md.index("次要建议")
def test_render_campaign_markdown_analysis_falls_back_to_id_prefix():
md = render_campaign_markdown(_campaign_report(), analysis=_analysis(), scenario_names={})
assert "**[高] 售后答非所问**(场景:s-1)" in md
def test_render_campaign_markdown_without_analysis_unchanged():
md = render_campaign_markdown(_campaign_report())
assert "智能分析" not in md