15c542d92c
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.
266 lines
10 KiB
Python
266 lines
10 KiB
Python
"""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"
|