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.
308 lines
12 KiB
Python
308 lines
12 KiB
Python
"""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))
|