diff --git a/backend/agenteval/evaluation/campaign_runner.py b/backend/agenteval/evaluation/campaign_runner.py index 46f02f2..3534fd2 100644 --- a/backend/agenteval/evaluation/campaign_runner.py +++ b/backend/agenteval/evaluation/campaign_runner.py @@ -20,6 +20,7 @@ from typing import Optional from sqlmodel import Session +from agenteval.evaluation.analysis import resolve_analysis_model, start_campaign_analysis from agenteval.evaluation.campaign_scheduler import ( TickAction, clock_offset, @@ -183,6 +184,23 @@ async def advance_campaign( 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 ────────────────────────────────────────────────── @@ -232,6 +250,7 @@ async def run_campaign_loop( current.status = CampaignStatus.COMPLETED current.completed_at = utc_now() repo.update(current) + _auto_start_analysis(current, session) return try: diff --git a/backend/agenteval/evaluation/report_render.py b/backend/agenteval/evaluation/report_render.py index 1dbd6f4..12ebac9 100644 --- a/backend/agenteval/evaluation/report_render.py +++ b/backend/agenteval/evaluation/report_render.py @@ -173,8 +173,57 @@ def render_markdown(report: dict[str, Any]) -> str: return "\n".join(lines) -def render_campaign_markdown(report: dict[str, Any]) -> str: - """Render a dual-axis campaign report dict as Markdown.""" +_SEVERITY_LABELS = {"high": "高", "medium": "中", "low": "低"} + + +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"] lines: list[str] = [ 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"{_pct(c['availability'])} | {_ms(c['avg_latency_ms'])} |" ) + if analysis: + lines += _render_analysis_lines(analysis, scenario_names or {}) return "\n".join(lines) diff --git a/backend/agenteval/web/routers/campaigns.py b/backend/agenteval/web/routers/campaigns.py index 88ecfe7..ebde8dc 100644 --- a/backend/agenteval/web/routers/campaigns.py +++ b/backend/agenteval/web/routers/campaigns.py @@ -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") runs = RunRepository(session).list_by_campaign(campaign_id) 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( content=md, media_type="text/markdown; charset=utf-8", diff --git a/tests/integration/test_campaign_analysis_api.py b/tests/integration/test_campaign_analysis_api.py index 43aff15..f54ea59 100644 --- a/tests/integration/test_campaign_analysis_api.py +++ b/tests/integration/test_campaign_analysis_api.py @@ -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): resp = await client.get("/api/campaigns/nope/analysis") 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 diff --git a/tests/integration/test_campaign_analysis_auto_trigger.py b/tests/integration/test_campaign_analysis_auto_trigger.py new file mode 100644 index 0000000..9fa5add --- /dev/null +++ b/tests/integration/test_campaign_analysis_auto_trigger.py @@ -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 == [] diff --git a/tests/unit/test_report_render.py b/tests/unit/test_report_render.py index 817f0f1..19f245c 100644 --- a/tests/unit/test_report_render.py +++ b/tests/unit/test_report_render.py @@ -185,3 +185,64 @@ def test_render_campaign_markdown_empty_bucket_dashes(): md = render_campaign_markdown(_campaign_report()) # bucket 1 has no runs: pass_rate/availability/latency all render as — 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