From 15c542d92c6a77828015a0167b794160f8b29304 Mon Sep 17 00:00:00 2001 From: sinohqb Date: Mon, 3 Aug 2026 02:06:29 +0800 Subject: [PATCH] feat(analysis): two-phase campaign analysis agent with storage and API Add the analysis role's execution path: a two-phase orchestration (per-scenario diagnosis gathered in parallel, then a synthesis pass) that reads the existing campaign report aggregation plus capped failure samples, validates the LLM's JSON against the report schema, and strips fabricated run/scenario references before persisting. Results upsert one row per campaign (generating/completed/failed) with the model config snapshot; GET/POST /api/campaigns/{id}/analysis expose the state machine, guarding non-terminal campaigns and missing analysis models with 400s. --- backend/agenteval/evaluation/analysis.py | 307 ++++++++++++++++++ backend/agenteval/storage/db.py | 22 ++ .../storage/model_config_repository.py | 4 + backend/agenteval/storage/repository.py | 39 +++ backend/agenteval/web/routers/campaigns.py | 37 ++- .../c8f5e4b13d26_add_campaign_analyses.py | 38 +++ tests/conftest.py | 1 + .../integration/test_campaign_analysis_api.py | 185 +++++++++++ .../unit/test_campaign_analyses_migration.py | 31 ++ tests/unit/test_campaign_analysis.py | 265 +++++++++++++++ 10 files changed, 928 insertions(+), 1 deletion(-) create mode 100644 backend/agenteval/evaluation/analysis.py create mode 100644 migrations/versions/c8f5e4b13d26_add_campaign_analyses.py create mode 100644 tests/integration/test_campaign_analysis_api.py create mode 100644 tests/unit/test_campaign_analyses_migration.py create mode 100644 tests/unit/test_campaign_analysis.py diff --git a/backend/agenteval/evaluation/analysis.py b/backend/agenteval/evaluation/analysis.py new file mode 100644 index 0000000..79c036d --- /dev/null +++ b/backend/agenteval/evaluation/analysis.py @@ -0,0 +1,307 @@ +"""Campaign intelligence analysis — the two-phase analysis agent (分析岗位). + +对终态活动的聚合结果做活动级、跨场景的叙述性研判（CONTEXT.md「分析岗位」）。 +输入完全复用 ``generate_campaign_report`` 的既有聚合口径（ADR-0002/0004，不重算 +数字），外加每场景少量代表性失败对话样例；阶段一按场景并行诊断，阶段二综合 +研判产出结构化报告。LLM 调用经由 ``chat_client`` 注入，测试用假客户端替换。 +""" + +import asyncio +import json +import logging +from typing import Any, Awaitable, Callable, Optional + +from sqlmodel import Session + +from agenteval.evaluation.report import generate_campaign_report +from agenteval.model_gateway import ModelGateway +from agenteval.models import Campaign, ModelCapability, RunStatus +from agenteval.services.model_configs import ( + ModelConfigError, + ModelConfigService, + ModelRuntimeConfig, +) +from agenteval.storage.db import get_session +from agenteval.storage.model_config_repository import ModelConfigRepository +from agenteval.storage.repository import ( + CampaignAnalysisRepository, + CampaignRepository, + RunRepository, + ScenarioRepository, +) +from agenteval.utils.llm import extract_reply_text, parse_json_from_llm_text + +_logger = logging.getLogger("agenteval") + +# LLM 客户端协议：接收 chat 消息列表，返回文本内容。生产实现走 ModelGateway， +# 测试注入假客户端（同 MockChannel 先例）。 +ChatClient = Callable[[list[dict[str, str]]], Awaitable[str]] + +MAX_SAMPLES_PER_SCENARIO = 3 +SAMPLE_TEXT_LIMIT = 200 +_VALID_SEVERITIES = {"high", "medium", "low"} + + +class AnalysisError(RuntimeError): + """分析生成失败（数据缺失或模型输出无法解析），可重试。""" + + +def resolve_analysis_model(campaign: Campaign, session: Session) -> Optional[ModelRuntimeConfig]: + """解析该活动应使用的分析模型：活动覆盖 ?? 全局分析默认；解析不到返回 None。""" + config_id = campaign.analysis_model_config_id + if config_id is None: + default = ModelConfigRepository(session).get_analysis_default() + config_id = default.id if default else None + if config_id is None: + return None + try: + return ModelConfigService(session).resolve(config_id, expected_capability=ModelCapability.CHAT) + except ModelConfigError: + return None + + +def collect_failure_samples( + campaign_id: str, + session: Session, + *, + per_scenario: int = MAX_SAMPLES_PER_SCENARIO, + text_limit: int = SAMPLE_TEXT_LIMIT, +) -> dict[str, list[dict[str, str]]]: + """每场景最多 ``per_scenario`` 条代表性失败对话（用户消息/回复/判定理由，截断）。""" + run_repo = RunRepository(session) + samples: dict[str, list[dict[str, str]]] = {} + for run in run_repo.list_by_campaign(campaign_id): + if run.status != RunStatus.COMPLETED: + continue + failed = [r for r in run_repo.get_results(run.id) if not r.passed] + if not failed: + continue + turns = {t.id: t for t in run_repo.get_turns(run.id)} + bucket = samples.setdefault(run.scenario_id, []) + for result in failed: + if len(bucket) >= per_scenario: + break + turn = turns.get(result.turn_id) + user = extract_reply_text(turn.get_sent_message().get("msgBody")) if turn else "" + reply = extract_reply_text(turn.get_reply().get("msgBody")) if turn and turn.get_reply() else "" + bucket.append({ + "run_id": run.id or "", + "user": user[:text_limit], + "reply": reply[:text_limit], + "reason": (result.reason or "")[:text_limit], + }) + return {sid: items for sid, items in samples.items() if items} + + +def _parse_stage(content: str, label: str) -> dict[str, Any]: + try: + parsed = parse_json_from_llm_text(content) + except Exception as exc: + raise AnalysisError(f"{label}输出解析失败: {exc}") from exc + if not isinstance(parsed, dict): + raise AnalysisError(f"{label}输出不是 JSON 对象") + return parsed + + +async def _analyze_scenario( + entry: dict[str, Any], + samples: list[dict[str, str]], + chat_client: ChatClient, +) -> dict[str, Any]: + """阶段一：单个场景的诊断（叙述 + 问题点草稿）。""" + system_prompt = ( + "你是智能客服质量评估平台的分析专家，负责对一次评估活动中某个场景的表现做诊断。" + "只输出一个 JSON 对象：" + '{"narrative": "该场景的叙述性表现分析（2-4 句）", ' + '"problems": [{"severity": "high|medium|low", "title": "...", "description": "...", ' + '"evidence_run_ids": ["来自输入数据的真实 run_id"]}]}' + "；没有问题时 problems 为空数组。全部使用中文。" + ) + user_prompt = json.dumps( + { + "场景": {"id": entry["scenario_id"], "名称": entry.get("scenario_name", "")}, + "聚合指标": { + "执行次数": entry.get("run_count"), + "通过率": entry.get("pass_rate"), + "可用性": entry.get("availability"), + "平均时延ms": entry.get("avg_latency_ms"), + }, + "代表性失败对话": samples, + }, + ensure_ascii=False, + ) + parsed = _parse_stage( + await chat_client([ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_prompt}, + ]), + f"场景「{entry.get('scenario_name', entry['scenario_id'])}」阶段一", + ) + narrative = parsed.get("narrative") + if not isinstance(narrative, str) or not narrative.strip(): + raise AnalysisError(f"场景「{entry.get('scenario_name', entry['scenario_id'])}」阶段一缺少 narrative") + return { + "scenario_id": entry["scenario_id"], + "narrative": narrative, + "problems": parsed.get("problems") if isinstance(parsed.get("problems"), list) else [], + } + + +async def _synthesize( + campaign: Campaign, + report: dict[str, Any], + stage1: list[dict[str, Any]], + chat_client: ChatClient, +) -> dict[str, Any]: + """阶段二：汇总各场景产出，产总体结论 + 跨场景问题 + 优先级建议。""" + system_prompt = ( + "你是智能客服质量评估平台的首席分析专家，负责对整个评估活动做综合研判。" + "只输出一个 JSON 对象：" + '{"overall": "总体结论（一段话）", ' + '"problems": [{"severity": "high|medium|low", "title": "...", "description": "...", ' + '"scenario_ids": ["涉及场景 id"], "evidence_run_ids": ["来自输入数据的真实 run_id"]}], ' + '"suggestions": [{"priority": 1, "text": "可执行的改善建议"}]}' + "；问题按严重度从高到低排列，建议按优先级排列。全部使用中文。" + ) + user_prompt = json.dumps( + { + "活动": { + "名称": campaign.name, + "窗口秒数": campaign.window_seconds, + "总体指标": report.get("summary", {}), + }, + "各场景诊断": stage1, + }, + ensure_ascii=False, + ) + parsed = _parse_stage( + await chat_client([ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_prompt}, + ]), + "阶段二综合研判", + ) + overall = parsed.get("overall") + if not isinstance(overall, str) or not overall.strip(): + raise AnalysisError("阶段二综合研判缺少 overall") + return parsed + + +async def analyze_campaign( + *, + campaign: Campaign, + report: dict[str, Any], + failure_samples: dict[str, list[dict[str, str]]], + valid_run_ids: set[str], + chat_client: ChatClient, +) -> dict[str, Any]: + """两阶段编排：阶段一按场景并行诊断，阶段二综合研判。 + + 输出遵循结构化报告 schema（overall/problems/scenario_narratives/suggestions）。 + 模型虚构的 run_id / scenario_id 在返回前按白名单剔除；任何解析失败抛 + ``AnalysisError``（由调用方落 failed 状态）。 + """ + capability = report.get("capability_summary") or [] + if not capability: + raise AnalysisError("活动没有可分析的场景数据") + + stage1 = await asyncio.gather(*[ + _analyze_scenario(entry, failure_samples.get(entry["scenario_id"], []), chat_client) + for entry in capability + ]) + stage2 = await _synthesize(campaign, report, list(stage1), chat_client) + + valid_scenario_ids = {entry["scenario_id"] for entry in capability} + problems = [] + for p in stage2.get("problems") or []: + if not isinstance(p, dict): + continue + severity = p.get("severity") + problems.append({ + "severity": severity if severity in _VALID_SEVERITIES else "medium", + "title": str(p.get("title", "")), + "description": str(p.get("description", "")), + "scenario_ids": [s for s in p.get("scenario_ids") or [] if s in valid_scenario_ids], + "evidence_run_ids": [r for r in p.get("evidence_run_ids") or [] if r in valid_run_ids], + }) + suggestions = [ + {"priority": int(s.get("priority", i + 1)), "text": str(s.get("text", ""))} + for i, s in enumerate(stage2.get("suggestions") or []) + if isinstance(s, dict) + ] + return { + "overall": stage2["overall"], + "problems": problems, + "scenario_narratives": [ + {"scenario_id": s["scenario_id"], "narrative": s["narrative"]} for s in stage1 + ], + "suggestions": suggestions, + } + + +def _gateway_chat_client(runtime: ModelRuntimeConfig) -> ChatClient: + gateway = ModelGateway(timeout=180.0) + + async def _chat(messages: list[dict[str, str]]) -> str: + return await gateway.chat(runtime, messages, temperature=0.2) + + return _chat + + +async def execute_campaign_analysis( + campaign_id: str, + *, + triggered_by: str, + chat_client: Optional[ChatClient] = None, +) -> None: + """后台执行体：generating → completed/failed 状态机（upsert，每活动一行）。 + + 与 Runs 同款后台任务约定：自持 Session、try/finally 关闭、失败落 error。 + """ + session = get_session() + try: + analyses = CampaignAnalysisRepository(session) + campaign = CampaignRepository(session).get(campaign_id) + if not campaign: + return + runtime = resolve_analysis_model(campaign, session) + if runtime is None: + analyses.upsert( + campaign_id, status="failed", triggered_by=triggered_by, + error="未配置分析模型：请在模型配置中心将某个 chat 配置设为「分析默认」", + ) + return + analyses.upsert( + campaign_id, status="generating", + model_config_id=runtime.id, triggered_by=triggered_by, + ) + try: + client = chat_client or _gateway_chat_client(runtime) + 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) + result = await analyze_campaign( + campaign=campaign, + report=report, + failure_samples=collect_failure_samples(campaign_id, session), + valid_run_ids={r.id for r in runs if r.id}, + chat_client=client, + ) + except Exception as exc: + _logger.warning("活动 %s 智能分析失败: %s", campaign_id, exc) + analyses.upsert( + campaign_id, status="failed", model_config_id=runtime.id, + error=str(exc)[:500], triggered_by=triggered_by, + ) + return + analyses.upsert( + campaign_id, status="completed", result=result, + model_config_id=runtime.id, triggered_by=triggered_by, + ) + finally: + session.close() + + +def start_campaign_analysis(campaign_id: str, *, triggered_by: str) -> asyncio.Task: + """以后台任务启动分析生成（fire-and-forget；状态经 campaign_analyses 表观测）。""" + return asyncio.create_task(execute_campaign_analysis(campaign_id, triggered_by=triggered_by)) diff --git a/backend/agenteval/storage/db.py b/backend/agenteval/storage/db.py index 76e5409..c92eaa5 100644 --- a/backend/agenteval/storage/db.py +++ b/backend/agenteval/storage/db.py @@ -202,6 +202,28 @@ class CampaignDB(SQLModel, table=True): self.summary = _json_dumps(summary) +class CampaignAnalysisDB(SQLModel, table=True): + """One row per campaign holding its intelligent analysis (智能分析) state.""" + + __tablename__ = "campaign_analyses" + + id: Optional[str] = Field(default_factory=new_uuid, primary_key=True) + campaign_id: str = Field(unique=True, index=True) + status: str = "generating" + result: Optional[str] = None + model_config_id: Optional[str] = None + error: Optional[str] = None + triggered_by: str = "manual" + created_at: Optional[datetime] = Field(default_factory=utc_now) + updated_at: Optional[datetime] = Field(default_factory=utc_now) + + def get_result(self) -> Optional[dict[str, Any]]: + return _json_loads(self.result) if self.result else None + + def set_result(self, result: dict[str, Any]) -> None: + self.result = _json_dumps(result) + + class EvalRunDB(SQLModel, table=True): """Database table for evaluation runs.""" diff --git a/backend/agenteval/storage/model_config_repository.py b/backend/agenteval/storage/model_config_repository.py index 3b0f26d..11b7e91 100644 --- a/backend/agenteval/storage/model_config_repository.py +++ b/backend/agenteval/storage/model_config_repository.py @@ -28,6 +28,10 @@ class ModelConfigRepository: def get_by_name(self, name: str) -> ModelConfigDB | None: return self.session.exec(select(ModelConfigDB).where(ModelConfigDB.name == name)).first() + def get_analysis_default(self) -> ModelConfigDB | None: + statement = select(ModelConfigDB).where(ModelConfigDB.is_analysis_default.is_(True)) + return self.session.exec(statement).first() + def create(self, config: ModelConfigDB) -> ModelConfigDB: if config.is_default: self.clear_default(config.capability) diff --git a/backend/agenteval/storage/repository.py b/backend/agenteval/storage/repository.py index f3ccfb3..7b3ee3a 100644 --- a/backend/agenteval/storage/repository.py +++ b/backend/agenteval/storage/repository.py @@ -7,6 +7,7 @@ from sqlmodel import Session, select from agenteval.models import Campaign, Case, EvalResult, EvalRun, EvalTarget, Scenario from agenteval.services.model_configs import ModelConfigService from agenteval.storage.db import ( + CampaignAnalysisDB, CampaignDB, EvalResultDB, EvalRunDB, @@ -394,6 +395,44 @@ class CampaignRepository(BaseRepository[Campaign, CampaignDB]): return self._from_db(existing) +class CampaignAnalysisRepository: + """Repository for campaign analysis rows (one per campaign, upserted).""" + + def __init__(self, session: Optional[Session] = None): + self.session = session or get_session() + + def get_by_campaign(self, campaign_id: str) -> Optional[CampaignAnalysisDB]: + statement = select(CampaignAnalysisDB).where(CampaignAnalysisDB.campaign_id == campaign_id) + return self.session.exec(statement).first() + + def upsert( + self, + campaign_id: str, + *, + status: str, + result: Optional[dict] = None, + model_config_id: Optional[str] = None, + error: Optional[str] = None, + triggered_by: str = "manual", + ) -> CampaignAnalysisDB: + row = self.get_by_campaign(campaign_id) + if row is None: + row = CampaignAnalysisDB(campaign_id=campaign_id) + row.status = status + if result is not None: + row.set_result(result) + else: + row.result = None + row.model_config_id = model_config_id + row.error = error + row.triggered_by = triggered_by + row.updated_at = utc_now() + self.session.add(row) + self.session.commit() + self.session.refresh(row) + return row + + class ResultRepository: """Repository for evaluation results.""" diff --git a/backend/agenteval/web/routers/campaigns.py b/backend/agenteval/web/routers/campaigns.py index 1fdd390..88ecfe7 100644 --- a/backend/agenteval/web/routers/campaigns.py +++ b/backend/agenteval/web/routers/campaigns.py @@ -11,6 +11,7 @@ from fastapi import APIRouter, Depends, HTTPException, Response from pydantic import BaseModel, Field from sqlmodel import Session +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.report import ( build_campaign_timeline, @@ -19,9 +20,10 @@ from agenteval.evaluation.report import ( ) from agenteval.evaluation.report_render import render_campaign_markdown from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus -from agenteval.storage.db import utc_now +from agenteval.storage.db import iso_utc, utc_now from agenteval.storage.model_config_repository import ModelConfigRepository from agenteval.storage.repository import ( + CampaignAnalysisRepository, CampaignRepository, RunRepository, ScenarioRepository, @@ -140,6 +142,39 @@ async def get_campaign_timeline(campaign_id: str, session: Session = Depends(get return {"entries": build_campaign_timeline(campaign, runs, scenario_names=scenario_names)} +@router.get("/{campaign_id}/analysis") +async def get_campaign_analysis(campaign_id: str, session: Session = Depends(get_db)) -> dict: + if not CampaignRepository(session).get(campaign_id): + raise HTTPException(status_code=404, detail="campaign not found") + row = CampaignAnalysisRepository(session).get_by_campaign(campaign_id) + if row is None: + return {"status": "none"} + return { + "status": row.status, + "result": row.get_result(), + "error": row.error, + "model_config_id": row.model_config_id, + "triggered_by": row.triggered_by, + "updated_at": iso_utc(row.updated_at), + } + + +@router.post("/{campaign_id}/analysis") +async def trigger_campaign_analysis(campaign_id: str, session: Session = Depends(get_db)) -> dict: + campaign = CampaignRepository(session).get(campaign_id) + if not campaign: + raise HTTPException(status_code=404, detail="campaign not found") + if campaign.status in (CampaignStatus.PLANNED, CampaignStatus.RUNNING): + raise HTTPException(status_code=400, detail="活动完成后才能生成智能分析") + if resolve_analysis_model(campaign, session) is None: + raise HTTPException( + status_code=400, + detail="未配置分析模型：请在模型配置中心将某个 chat 配置设为「分析默认」，或为该活动指定分析模型", + ) + start_campaign_analysis(campaign_id, triggered_by="manual") + return {"status": "generating"} + + @router.get("/{campaign_id}") async def get_campaign(campaign_id: str, session: Session = Depends(get_db)) -> dict: campaign = CampaignRepository(session).get(campaign_id) diff --git a/migrations/versions/c8f5e4b13d26_add_campaign_analyses.py b/migrations/versions/c8f5e4b13d26_add_campaign_analyses.py new file mode 100644 index 0000000..2a1dc4e --- /dev/null +++ b/migrations/versions/c8f5e4b13d26_add_campaign_analyses.py @@ -0,0 +1,38 @@ +"""add campaign_analyses table + +Revision ID: c8f5e4b13d26 +Revises: b7e4d3a92c15 +Create Date: 2026-08-03 +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +import sqlmodel # noqa: F401 +from alembic import op + +revision: str = "c8f5e4b13d26" +down_revision: Union[str, Sequence[str], None] = "b7e4d3a92c15" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "campaign_analyses", + sa.Column("id", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column("campaign_id", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column("status", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column("result", sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column("model_config_id", sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column("error", sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column("triggered_by", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column("created_at", sa.DateTime(), nullable=True), + sa.Column("updated_at", sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("campaign_id", name="uq_campaign_analyses_campaign_id"), + ) + + +def downgrade() -> None: + op.drop_table("campaign_analyses") diff --git a/tests/conftest.py b/tests/conftest.py index 53bb094..203c886 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -30,6 +30,7 @@ def db_session(tmp_db_path: Path) -> Session: """ # Import DB models so their table=True declarations register in metadata. from agenteval.storage.db import ( # noqa: F401 + CampaignAnalysisDB, CampaignDB, EvalResultDB, EvalRunDB, diff --git a/tests/integration/test_campaign_analysis_api.py b/tests/integration/test_campaign_analysis_api.py new file mode 100644 index 0000000..43aff15 --- /dev/null +++ b/tests/integration/test_campaign_analysis_api.py @@ -0,0 +1,185 @@ +"""Integration tests for /api/campaigns/{id}/analysis (v0.7 ticket 03).""" + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlmodel import select + +from agenteval.models import ( + CampaignStatus, Case, CaseType, ChannelType, EvalTarget, PlatformType, + Scenario, TargetStatus, +) +from agenteval.storage.db import CampaignAnalysisDB, ModelConfigDB +from agenteval.storage.model_config_repository import ModelConfigRepository +from agenteval.storage.repository import CampaignRepository, ScenarioRepository, TargetRepository +from agenteval.web.app import app + + +@pytest.fixture() +def seeded_db(db_session, monkeypatch): + from agenteval.storage import db as db_module + from agenteval.storage import repository as repo_module + from agenteval.web import app as app_module + from agenteval.web.routers import campaigns as campaigns_module + + monkeypatch.setattr(app_module, "init_db", lambda: None) + monkeypatch.setattr(campaigns_module, "start_campaign", lambda *a, **k: None) + + monkeypatch.setattr(db_module, "get_session", lambda: db_session) + monkeypatch.setattr(repo_module, "get_session", lambda: db_session) + + from agenteval.web.deps import get_db + + def _test_get_db(): + try: + yield db_session + finally: + pass + + app.dependency_overrides[get_db] = _test_get_db + + 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"])], + )) + + yield db_session + + app.dependency_overrides.clear() + + +@pytest.fixture() +async def client(): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as c: + yield c + + +def _payload() -> dict: + return { + "name": "24h-cycle", + "target_id": "t-1", + "window_seconds": 86400, + "time_scale": 1.0, + "plan": [{"scenario_id": "s-1", "offset_seconds": 0, "count": 1}], + } + + +async def _create_campaign(client, session, status: CampaignStatus = CampaignStatus.COMPLETED) -> str: + campaign_id = (await client.post("/api/campaigns", json=_payload())).json()["id"] + repo = CampaignRepository(session) + campaign = repo.get(campaign_id) + campaign.status = status + repo.update(campaign) + return campaign_id + + +def _seed_analysis_default(session) -> None: + ModelConfigRepository(session).create(ModelConfigDB( + id="mc-1", name="analysis-cfg", provider="openai_compatible", capability="chat", + endpoint_url="https://models.example.com/v1/chat/completions", model_name="m", + is_analysis_default=True, + )) + + +def _complete_analysis_row(session, campaign_id: str) -> None: + from agenteval.storage.db import utc_now + + row = CampaignAnalysisDB( + campaign_id=campaign_id, status="completed", model_config_id="mc-1", + triggered_by="manual", updated_at=utc_now(), + ) + row.set_result({ + "overall": "整体达标", + "problems": [], + "scenario_narratives": [{"scenario_id": "s-1", "narrative": "表现稳定"}], + "suggestions": [{"priority": 1, "text": "保持"}], + }) + existing = session.exec( + select(CampaignAnalysisDB).where(CampaignAnalysisDB.campaign_id == campaign_id) + ).first() + if existing: + session.delete(existing) + session.commit() + session.add(row) + session.commit() + + +async def test_get_analysis_empty_state(client, seeded_db): + campaign_id = await _create_campaign(client, seeded_db) + resp = await client.get(f"/api/campaigns/{campaign_id}/analysis") + assert resp.status_code == 200 + assert resp.json() == {"status": "none"} + + +async def test_post_analysis_rejects_non_terminal_campaign(client, seeded_db): + campaign_id = await _create_campaign(client, seeded_db, status=CampaignStatus.RUNNING) + resp = await client.post(f"/api/campaigns/{campaign_id}/analysis") + assert resp.status_code == 400 + + +async def test_post_analysis_rejects_planned_campaign(client, seeded_db): + campaign_id = await _create_campaign(client, seeded_db, status=CampaignStatus.PLANNED) + resp = await client.post(f"/api/campaigns/{campaign_id}/analysis") + assert resp.status_code == 400 + + +async def test_post_analysis_without_model_guides_configuration(client, seeded_db): + campaign_id = await _create_campaign(client, seeded_db) + resp = await client.post(f"/api/campaigns/{campaign_id}/analysis") + assert resp.status_code == 400 + assert "分析" in resp.json()["detail"] + + +async def test_post_then_get_completed_analysis(client, seeded_db, monkeypatch): + from agenteval.web.routers import campaigns as campaigns_module + + _seed_analysis_default(seeded_db) + campaign_id = await _create_campaign(client, seeded_db) + + # 假后台任务：同步写入 completed 行（真任务的单测覆盖在 test_campaign_analysis.py） + monkeypatch.setattr( + campaigns_module, + "start_campaign_analysis", + lambda cid, *, triggered_by: _complete_analysis_row(seeded_db, cid), + ) + + resp = await client.post(f"/api/campaigns/{campaign_id}/analysis") + assert resp.status_code == 200 + + got = (await client.get(f"/api/campaigns/{campaign_id}/analysis")).json() + assert got["status"] == "completed" + assert got["model_config_id"] == "mc-1" + assert got["result"]["overall"] == "整体达标" + assert got["result"]["scenario_narratives"][0]["narrative"] == "表现稳定" + + +async def test_rerun_upserts_without_new_row(client, seeded_db, monkeypatch): + from agenteval.web.routers import campaigns as campaigns_module + + _seed_analysis_default(seeded_db) + campaign_id = await _create_campaign(client, seeded_db) + monkeypatch.setattr( + campaigns_module, + "start_campaign_analysis", + lambda cid, *, triggered_by: _complete_analysis_row(seeded_db, cid), + ) + + await client.post(f"/api/campaigns/{campaign_id}/analysis") + await client.post(f"/api/campaigns/{campaign_id}/analysis") + + rows = seeded_db.exec( + select(CampaignAnalysisDB).where(CampaignAnalysisDB.campaign_id == campaign_id) + ).all() + assert len(rows) == 1 + + +async def test_get_analysis_missing_campaign_404(client, seeded_db): + resp = await client.get("/api/campaigns/nope/analysis") + assert resp.status_code == 404 diff --git a/tests/unit/test_campaign_analyses_migration.py b/tests/unit/test_campaign_analyses_migration.py new file mode 100644 index 0000000..75e533e --- /dev/null +++ b/tests/unit/test_campaign_analyses_migration.py @@ -0,0 +1,31 @@ +"""Verify the campaign_analyses migration creates the table with a unique campaign_id.""" + +import importlib + +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations + + +def test_campaign_analyses_migration_creates_table(tmp_path, monkeypatch): + engine = sa.create_engine(f"sqlite:///{tmp_path / 'analyses.db'}") + + with engine.begin() as connection: + operations = Operations(MigrationContext.configure(connection)) + migration = importlib.import_module( + "migrations.versions.c8f5e4b13d26_add_campaign_analyses" + ) + monkeypatch.setattr(migration, "op", operations) + migration.upgrade() + + inspector = sa.inspect(connection) + assert "campaign_analyses" in inspector.get_table_names() + + columns = {c["name"] for c in inspector.get_columns("campaign_analyses")} + assert { + "id", "campaign_id", "status", "result", + "model_config_id", "error", "triggered_by", "created_at", "updated_at", + } <= columns + + uniques = inspector.get_unique_constraints("campaign_analyses") + assert any(uc["column_names"] == ["campaign_id"] for uc in uniques) diff --git a/tests/unit/test_campaign_analysis.py b/tests/unit/test_campaign_analysis.py new file mode 100644 index 0000000..4a2ea65 --- /dev/null +++ b/tests/unit/test_campaign_analysis.py @@ -0,0 +1,265 @@ +"""Two-phase campaign analysis orchestration (v0.7 ticket 03).""" + +import json + +import pytest +from sqlmodel import select + +from agenteval.evaluation.analysis import ( + AnalysisError, + analyze_campaign, + collect_failure_samples, + execute_campaign_analysis, + resolve_analysis_model, +) +from agenteval.models import Campaign, CampaignPlanEntry, EvalRun, RunStatus +from agenteval.storage.db import CampaignAnalysisDB, EvalResultDB, ModelConfigDB, TurnDB +from agenteval.storage.model_config_repository import ModelConfigRepository +from agenteval.storage.repository import CampaignRepository, RunRepository + + +class FakeChatClient: + """Queued-response fake for the analysis LLM seam.""" + + def __init__(self, *responses): + self._responses = list(responses) + self.calls: list[list[dict]] = [] + + async def __call__(self, messages: list[dict]) -> str: + self.calls.append(messages) + if not self._responses: + raise AssertionError("unexpected extra LLM call") + item = self._responses.pop(0) + if isinstance(item, Exception): + raise item + return item + + +def _campaign(**overrides) -> Campaign: + data = { + "id": "camp-1", + "name": "24h 正式线", + "target_id": "t-1", + "window_seconds": 86400, + "time_scale": 1.0, + "plan": [CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=2)], + } + data.update(overrides) + return Campaign(**data) + + +def _report() -> dict: + return { + "campaign_id": "camp-1", + "name": "24h 正式线", + "summary": { + "total_runs": 3, + "completed_runs": 3, + "overall_pass_rate": 0.5, + "overall_availability": 1.0, + "avg_latency_ms": 800, + }, + "time_trend": [], + "capability_summary": [ + {"scenario_id": "s-1", "scenario_name": "售前咨询", "run_count": 2, + "pass_rate": 0.5, "availability": 1.0, "avg_latency_ms": 700}, + {"scenario_id": "s-2", "scenario_name": "售后工单", "run_count": 1, + "pass_rate": 0.0, "availability": 1.0, "avg_latency_ms": 900}, + ], + } + + +STAGE1_A = json.dumps({ + "narrative": "售前场景半数对话答非所问", + "problems": [{"severity": "high", "title": "答非所问", "description": "...", "evidence_run_ids": ["run-1"]}], +}) +STAGE1_B = json.dumps({"narrative": "售后场景全部超时", "problems": []}) +STAGE2 = json.dumps({ + "overall": "活动整体不达标，售前售后均有阻断性问题", + "problems": [ + {"severity": "high", "title": "售前答非所问", "description": "意图识别漂移", + "scenario_ids": ["s-1", "ghost-scenario"], "evidence_run_ids": ["run-1", "fake-run"]}, + {"severity": "nonsense", "title": "售后超时", "description": "上游缓慢", + "scenario_ids": ["s-2"], "evidence_run_ids": ["run-2"]}, + ], + "suggestions": [{"priority": 1, "text": "补充售前意图语料"}, {"priority": 2, "text": "排查工单接口"}], +}) + + +async def test_two_phase_orchestration_assembles_result(): + client = FakeChatClient(STAGE1_A, STAGE1_B, STAGE2) + result = await analyze_campaign( + campaign=_campaign(), + report=_report(), + failure_samples={"s-2": [{"run_id": "run-2", "user": "u", "reply": "r", "reason": "超时"}]}, + valid_run_ids={"run-1", "run-2"}, + chat_client=client, + ) + + # 阶段一每场景一次调用，阶段二一次汇总调用 + assert len(client.calls) == 3 + assert "售前咨询" in str(client.calls[0]) and "售后工单" in str(client.calls[1]) + stage2_prompt = str(client.calls[2]) + assert "售前场景半数对话答非所问" in stage2_prompt and "售后场景全部超时" in stage2_prompt + + assert result["overall"] == "活动整体不达标，售前售后均有阻断性问题" + assert result["scenario_narratives"] == [ + {"scenario_id": "s-1", "narrative": "售前场景半数对话答非所问"}, + {"scenario_id": "s-2", "narrative": "售后场景全部超时"}, + ] + assert result["suggestions"] == [ + {"priority": 1, "text": "补充售前意图语料"}, + {"priority": 2, "text": "排查工单接口"}, + ] + + +async def test_fabricated_evidence_is_dropped_before_persisting(): + client = FakeChatClient(STAGE1_A, STAGE1_B, STAGE2) + result = await analyze_campaign( + campaign=_campaign(), + report=_report(), + failure_samples={}, + valid_run_ids={"run-1", "run-2"}, + chat_client=client, + ) + + problem = result["problems"][0] + assert problem["evidence_run_ids"] == ["run-1"] # fake-run 剔除 + assert problem["scenario_ids"] == ["s-1"] # ghost-scenario 剔除 + assert result["problems"][1]["severity"] == "medium" # 非法 severity 归一 + + +async def test_unparseable_llm_output_fails_analysis(): + client = FakeChatClient("这不是 JSON") + with pytest.raises(AnalysisError): + await analyze_campaign( + campaign=_campaign(), + report=_report(), + failure_samples={}, + valid_run_ids=set(), + chat_client=client, + ) + + +async def test_scenario_without_failure_samples_still_gets_narrative(): + client = FakeChatClient(STAGE1_A, STAGE1_B, STAGE2) + result = await analyze_campaign( + campaign=_campaign(), + report=_report(), + failure_samples={}, + valid_run_ids=set(), + chat_client=client, + ) + assert len(result["scenario_narratives"]) == 2 + + +# ── 分析模型解析 ───────────────────────────────────────────────────────── + +def _seed_config(session, config_id: str, *, analysis_default: bool = False, enabled: bool = True) -> None: + ModelConfigRepository(session).create(ModelConfigDB( + id=config_id, name=f"cfg-{config_id}", provider="openai_compatible", capability="chat", + endpoint_url="https://models.example.com/v1/chat/completions", model_name="m", + enabled=enabled, is_analysis_default=analysis_default, + )) + + +def test_resolve_returns_none_when_nothing_configured(db_session): + assert resolve_analysis_model(_campaign(), db_session) is None + + +def test_resolve_prefers_campaign_override(db_session): + _seed_config(db_session, "mc-default", analysis_default=True) + _seed_config(db_session, "mc-override") + runtime = resolve_analysis_model(_campaign(analysis_model_config_id="mc-override"), db_session) + assert runtime is not None and runtime.id == "mc-override" + + +def test_resolve_falls_back_to_global_default(db_session): + _seed_config(db_session, "mc-default", analysis_default=True) + runtime = resolve_analysis_model(_campaign(), db_session) + assert runtime is not None and runtime.id == "mc-default" + + +def test_resolve_returns_none_when_override_missing(db_session): + assert resolve_analysis_model(_campaign(analysis_model_config_id="ghost"), db_session) is None + + +# ── 失败样例选取 ───────────────────────────────────────────────────────── + +def _seed_failed_run(session, run_id: str, scenario_id: str, failed_turns: int) -> None: + RunRepository(session).create(EvalRun( + id=run_id, target_id="t-1", scenario_id=scenario_id, + campaign_id="camp-1", status=RunStatus.COMPLETED, + )) + for i in range(failed_turns): + turn = TurnDB(id=f"{run_id}-turn-{i}", run_id=run_id, case_id=f"c{i}", round_index=0) + turn.set_sent_message({"msgBody": {"content": f"用户消息-{run_id}-{i}"}}) + turn.set_reply({"msgBody": {"content": "答" * 300}}) + session.add(turn) + session.add(EvalResultDB( + run_id=run_id, case_id=f"c{i}", turn_id=turn.id, + rule_type="llm_score", passed=False, reason=f"不合格-{i}", + )) + session.commit() + + +def test_collect_failure_samples_caps_and_truncates(db_session): + _seed_failed_run(db_session, "run-1", "s-1", failed_turns=4) + samples = collect_failure_samples("camp-1", db_session) + + assert set(samples) == {"s-1"} + assert len(samples["s-1"]) == 3 # 每场景最多 3 条 + item = samples["s-1"][0] + assert item["run_id"] == "run-1" + assert item["user"].startswith("用户消息-run-1") + assert len(item["reply"]) <= 200 # 截断 + assert item["reason"] + + +def test_collect_failure_samples_skips_passing_runs(db_session): + RunRepository(db_session).create(EvalRun( + id="run-ok", target_id="t-1", scenario_id="s-1", + campaign_id="camp-1", status=RunStatus.COMPLETED, + )) + assert collect_failure_samples("camp-1", db_session) == {} + + +# ── 后台执行状态机 ─────────────────────────────────────────────────────── + +async def test_execute_writes_completed_row_with_model_snapshot(db_session, monkeypatch): + from agenteval.evaluation import analysis as analysis_module + + monkeypatch.setattr(analysis_module, "get_session", lambda: db_session) + _seed_config(db_session, "mc-default", analysis_default=True) + CampaignRepository(db_session).create(_campaign()) + _seed_failed_run(db_session, "run-1", "s-1", failed_turns=1) + _seed_failed_run(db_session, "run-2", "s-2", failed_turns=1) + + client = FakeChatClient(STAGE1_A, STAGE1_B, STAGE2) + await execute_campaign_analysis("camp-1", triggered_by="manual", chat_client=client) + + row = db_session.exec( + select(CampaignAnalysisDB).where(CampaignAnalysisDB.campaign_id == "camp-1") + ).one() + assert row.status == "completed" + assert row.model_config_id == "mc-default" + assert row.triggered_by == "manual" + assert row.get_result()["overall"].startswith("活动整体不达标") + + +async def test_execute_records_failure(db_session, monkeypatch): + from agenteval.evaluation import analysis as analysis_module + + monkeypatch.setattr(analysis_module, "get_session", lambda: db_session) + _seed_config(db_session, "mc-default", analysis_default=True) + CampaignRepository(db_session).create(_campaign()) + + client = FakeChatClient("garbage") + await execute_campaign_analysis("camp-1", triggered_by="auto", chat_client=client) + + row = db_session.exec( + select(CampaignAnalysisDB).where(CampaignAnalysisDB.campaign_id == "camp-1") + ).one() + assert row.status == "failed" + assert row.error + assert row.triggered_by == "auto"