"""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 load_campaign_view 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, ) from agenteval.task_registry import TaskRegistry 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, exploration_summary: Optional[dict[str, Any]] = None, ) -> 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": "可执行的改善建议"}]}' "；问题按严重度从高到低排列，建议按优先级排列。全部使用中文。" ) payload: dict[str, Any] = { "活动": { "名称": campaign.name, "窗口秒数": campaign.window_seconds, "总体指标": report.get("summary", {}), }, "各场景诊断": stage1, } if exploration_summary is not None: # 探索式评测证据线：只给统计与问题清单，不含全量对话 payload["探索发现"] = exploration_summary user_prompt = json.dumps(payload, 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, exploration_summary: Optional[dict[str, Any]] = None, ) -> 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, exploration_summary=exploration_summary) 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 _maybe_enqueue_period_comparison(campaign: Campaign, session: Session) -> None: """正式线活动分析完成后自动链到周期对比（v0.8）。 前提：正式线（time_scale == 1）、分析模型可解析、存在自动基线 （同活动串且已有 completed 分析）。任一不满足静默跳过；异常仅告警， 不影响刚落库的分析结果。 """ try: if campaign.time_scale != 1: return if resolve_analysis_model(campaign, session) is None: return # 延迟导入：comparison 顶层依赖 analysis（resolve_analysis_model）， # 反向导入会成环。 from agenteval.evaluation import comparison as comparison_module if comparison_module.resolve_auto_baseline(campaign, session) is None: return comparison_module.start_campaign_comparison(campaign.id, triggered_by="auto") except Exception as exc: _logger.warning("活动 %s 自动周期对比跳过: %s", campaign.id, exc) def gateway_chat_client(runtime: ModelRuntimeConfig) -> ChatClient: """Shared ChatClient factory for analysis/comparison background executors.""" 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) view = load_campaign_view(session, campaign) result = await analyze_campaign( campaign=campaign, report=view["report"], failure_samples=collect_failure_samples(campaign_id, session), valid_run_ids={r.id for r in runs if r.id}, chat_client=client, exploration_summary=view["exploration"], ) 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, ) _maybe_enqueue_period_comparison(campaign, session) finally: session.close() analysis_registry = TaskRegistry() def start_campaign_analysis(campaign_id: str, *, triggered_by: str) -> None: """以后台任务启动分析生成（状态经 campaign_analyses 表观测）。 registry 持强引用防 GC，shutdown 时统一收敛；同 id 在跑时幂等不重复派生。 """ analysis_registry.launch( campaign_id, lambda _cancel: execute_campaign_analysis(campaign_id, triggered_by=triggered_by), ) def enqueue_campaign_analysis(campaign_id: str, *, triggered_by: str) -> None: """Durably queue analysis, then launch the in-process worker.""" session = get_session() try: CampaignAnalysisRepository(session).enqueue(campaign_id, triggered_by=triggered_by) finally: session.close() start_campaign_analysis(campaign_id, triggered_by=triggered_by) def resume_queued_campaign_analysis(session: Session) -> int: """Re-launch analysis jobs persisted before a process interruption.""" rows = CampaignAnalysisRepository(session).list_queued() for row in rows: start_campaign_analysis(row.campaign_id, triggered_by=row.triggered_by or "manual") return len(rows)