From 2285a25009108a9e6012f1d735d2a78be3e2a7ee Mon Sep 17 00:00:00 2001 From: sinohqb Date: Mon, 3 Aug 2026 18:59:40 +0800 Subject: [PATCH] feat(exploration): judge sampling review after session close After an exploration session closes, the platform samples up to 3 conversation rounds and runs an independent judge-role review through the v0.7 ChatClient seam, persisting quality-dimension conclusions (attitude, professionalism, hallucination) into the session's judge_review. The review runs as a background task: failures are recorded without touching session state or the first-hand experience record, and a missing model config skips silently. --- backend/agenteval/exploration/judge.py | 187 +++++++++++++++ backend/agenteval/web/routers/exploration.py | 5 +- tests/integration/test_exploration_api.py | 19 ++ tests/unit/test_exploration_judge.py | 239 +++++++++++++++++++ 4 files changed, 449 insertions(+), 1 deletion(-) create mode 100644 backend/agenteval/exploration/judge.py create mode 100644 tests/unit/test_exploration_judge.py diff --git a/backend/agenteval/exploration/judge.py b/backend/agenteval/exploration/judge.py new file mode 100644 index 0000000..5008c25 --- /dev/null +++ b/backend/agenteval/exploration/judge.py @@ -0,0 +1,187 @@ +"""Judge sampling review for exploration sessions (judge 岗位抽样复核). + +会话关闭后，平台对该会话对话抽样（默认 ≤3 段，控 token），经 judge 岗位模型 +独立复核，产出质量维度结论（态度、专业性、幻觉）落入会话的 ``judge_review``。 +复核是异步后台执行：失败落错误不阻塞会话状态，也不影响体验记录这条第一手 +证据线；未配置模型时静默跳过。LLM 调用沿 v0.7 分析的 ``ChatClient`` 接缝 +注入，测试用假客户端覆盖。 +""" + +import asyncio +import json +import logging +from typing import Any, Optional + +from sqlmodel import Session + +from agenteval.evaluation.analysis import ( + ChatClient, + gateway_chat_client, + resolve_analysis_model, +) +from agenteval.exploration.models import ExplorationMessage, ExplorationSession +from agenteval.models import Campaign +from agenteval.services.model_configs import ModelRuntimeConfig +from agenteval.storage.db import get_session, iso_utc, utc_now +from agenteval.storage.repository import ( + CampaignRepository, + ExplorationMessageRepository, + ExplorationSessionRepository, +) +from agenteval.utils.llm import parse_json_from_llm_text + +_logger = logging.getLogger("agenteval") + +MAX_JUDGE_SAMPLES = 3 +SAMPLE_TEXT_LIMIT = 500 +VALID_DIMENSIONS = ("attitude", "professionalism", "hallucination") +VALID_RATINGS = ("good", "acceptable", "poor") + +JUDGE_SYSTEM_PROMPT = """你是评测平台的 judge 岗位模型，负责独立复核一段「虚拟用户」与被评对象的探索对话。 +只依据给定对话抽样判断，不要臆测抽样之外的内容。输出必须是合法 JSON，且仅包含以下结构： +{ + "dimensions": [ + {"dimension": "attitude | professionalism | hallucination", "rating": "good | acceptable | poor", "comment": "一句话依据"} + ], + "summary": "一句话总体结论" +} +维度说明：attitude=服务态度；professionalism=专业性（流程与答复正确性）； +hallucination=幻觉（编造事实、政策或能力），无幻觉时 rating 为 good。 +三个维度必须各出现一次。""" + + +class JudgeReviewError(RuntimeError): + """judge 复核失败（模型输出无法解析或调用失败），不影响会话状态。""" + + +def resolve_judge_model(campaign: Campaign, session: Session) -> Optional[ModelRuntimeConfig]: + """解析 judge 岗位模型：沿用活动分析模型口径（活动覆盖 ?? 全局分析默认）。""" + return resolve_analysis_model(campaign, session) + + +def sample_round_indexes(rounds: list[int], limit: int = MAX_JUDGE_SAMPLES) -> list[int]: + """超限时均匀取样（含首尾），保证抽样可复现。""" + if len(rounds) <= limit: + return list(rounds) + span = len(rounds) - 1 + return [rounds[round(i * span / (limit - 1))] for i in range(limit)] + + +def build_judge_messages( + session_obj: ExplorationSession, samples: list[ExplorationMessage] +) -> list[dict[str, str]]: + transcript = [] + for message in samples: + speaker = "虚拟用户" if message.role == "user" else "被评对象" + content = str(message.content)[:SAMPLE_TEXT_LIMIT] + transcript.append(f"[第 {message.round_index} 轮] {speaker}: {content}") + payload = { + "persona": session_obj.persona, + "goal": session_obj.goal, + "transcript": transcript, + } + return [ + {"role": "system", "content": JUDGE_SYSTEM_PROMPT}, + {"role": "user", "content": json.dumps(payload, ensure_ascii=False)}, + ] + + +def normalize_judge_review(raw: Any) -> dict[str, Any]: + """白名单归一：未知维度丢弃，非法档位归 acceptable（沿 v0.7 白名单经验）。""" + dims = raw.get("dimensions") if isinstance(raw, dict) else None + dims = dims if isinstance(dims, list) else [] + normalized = [] + for item in dims: + if not isinstance(item, dict): + continue + name = item.get("dimension") + if name not in VALID_DIMENSIONS: + continue + rating = item.get("rating") + normalized.append( + { + "dimension": name, + "rating": rating if rating in VALID_RATINGS else "acceptable", + "comment": str(item.get("comment") or "")[:SAMPLE_TEXT_LIMIT], + } + ) + summary = str(raw.get("summary") or "")[:SAMPLE_TEXT_LIMIT] if isinstance(raw, dict) else "" + return {"dimensions": normalized, "summary": summary} + + +async def judge_conversation( + session_obj: ExplorationSession, + samples: list[ExplorationMessage], + *, + chat_client: ChatClient, +) -> dict[str, Any]: + text = await chat_client(build_judge_messages(session_obj, samples)) + try: + raw = parse_json_from_llm_text(text) + except Exception as exc: + raise JudgeReviewError(f"judge 输出解析失败: {exc}") from exc + if not isinstance(raw, dict): + raise JudgeReviewError("judge 输出解析失败: 输出不是 JSON 对象") + review = normalize_judge_review(raw) + review["sampled_rounds"] = sorted({m.round_index for m in samples}) + return review + + +async def execute_judge_review( + exploration_session_id: str, + *, + chat_client: Optional[ChatClient] = None, +) -> None: + """后台执行体：抽样 → judge 复核 → judge_review 落库。 + + 与 Runs 同款后台任务约定：自持 Session、try/finally 关闭、失败落 error + 不阻塞会话状态；未配置模型时静默跳过。 + """ + session = get_session() + try: + repo = ExplorationSessionRepository(session) + session_obj = repo.get(exploration_session_id) + if not session_obj: + return + campaign = CampaignRepository(session).get(session_obj.campaign_id) + runtime = resolve_judge_model(campaign, session) if campaign else None + if runtime is None: + _logger.info("探索会话 %s 跳过 judge 复核：未配置模型", exploration_session_id) + return + + messages = ExplorationMessageRepository(session).list_by_session(session_obj.id) + round_indexes = sorted({m.round_index for m in messages}) + sampled = set(sample_round_indexes(round_indexes)) + samples = [m for m in messages if m.round_index in sampled] + if not samples: + return + + client = chat_client or gateway_chat_client(runtime) + reviewed_at = iso_utc(utc_now()) + try: + review = await judge_conversation(session_obj, samples, chat_client=client) + review.update(status="completed", model_config_id=runtime.id, reviewed_at=reviewed_at) + except Exception as exc: + _logger.warning("探索会话 %s judge 复核失败: %s", exploration_session_id, exc) + review = { + "status": "failed", + "model_config_id": runtime.id, + "reviewed_at": reviewed_at, + "error": str(exc)[:SAMPLE_TEXT_LIMIT], + } + session_obj.judge_review = review + repo.update(session_obj) + finally: + session.close() + + +_BACKGROUND_TASKS: set[asyncio.Task] = set() + + +def start_judge_review(exploration_session_id: str) -> asyncio.Task: + """以后台任务启动抽样复核（fire-and-forget；结果经会话 judge_review 观测）。""" + task = asyncio.create_task(execute_judge_review(exploration_session_id)) + # 强引用防止未持有引用的任务被 GC（asyncio 已知坑） + _BACKGROUND_TASKS.add(task) + task.add_done_callback(_BACKGROUND_TASKS.discard) + return task diff --git a/backend/agenteval/web/routers/exploration.py b/backend/agenteval/web/routers/exploration.py index e781c84..e975a86 100644 --- a/backend/agenteval/web/routers/exploration.py +++ b/backend/agenteval/web/routers/exploration.py @@ -20,6 +20,7 @@ from sqlmodel import Session from agenteval.channels.factory import ChannelFactory from agenteval.config import get_settings from agenteval.evaluation.report import generate_campaign_report +from agenteval.exploration.judge import start_judge_review from agenteval.exploration.models import ( ExplorationBudget, ExplorationMessage, @@ -290,4 +291,6 @@ async def close_session( session_obj.experience = normalize_experience(request.experience) session_obj.status = ExplorationSessionStatus.COMPLETED session_obj.closed_at = utc_now() - return repo.update(session_obj).model_dump(mode="json") + updated = repo.update(session_obj) + start_judge_review(session_obj.id) + return updated.model_dump(mode="json") diff --git a/tests/integration/test_exploration_api.py b/tests/integration/test_exploration_api.py index 9ba20fd..08632e5 100644 --- a/tests/integration/test_exploration_api.py +++ b/tests/integration/test_exploration_api.py @@ -54,6 +54,11 @@ def seeded_db(db_session, monkeypatch): app.dependency_overrides[get_db] = _test_get_db + # 隔离后台 judge 复核任务：API 测试不真正派发后台任务 + from agenteval.web.routers import exploration as exploration_router + + monkeypatch.setattr(exploration_router, "start_judge_review", lambda session_id: None) + target = EvalTarget( id="t-1", name="mock-target", @@ -330,6 +335,20 @@ async def test_close_twice_rejected(seeded_db, mock_channel, client): assert resp.status_code == 409 +async def test_close_triggers_judge_review(seeded_db, mock_channel, client, monkeypatch): + from agenteval.web.routers import exploration as exploration_router + + started: list[str] = [] + monkeypatch.setattr(exploration_router, "start_judge_review", started.append) + + session_id = (await _create_session(client)).json()["id"] + resp = await client.post( + f"/api/exploration/sessions/{session_id}/close", json={"experience": {"goal_achieved": True}} + ) + assert resp.status_code == 200, resp.text + assert started == [session_id] + + # ---------------------------------------------------------------- migration diff --git a/tests/unit/test_exploration_judge.py b/tests/unit/test_exploration_judge.py new file mode 100644 index 0000000..4ab7afb --- /dev/null +++ b/tests/unit/test_exploration_judge.py @@ -0,0 +1,239 @@ +"""Judge sampling review for exploration sessions (v0.9 票据 04). + +会话关闭后，平台抽样对话（≤3 段）经 judge 岗位模型独立复核，质量维度结论 +结构化落入会话的 judge_review。复核是异步后台执行：失败落错误不阻塞会话 +状态；无模型配置时静默跳过。LLM 调用经可注入 ChatClient 接缝（沿 v0.7 +分析 seam），测试用假客户端覆盖（先例：test_campaign_analysis.py）。 +""" + +import json + +import pytest +from agenteval.exploration.judge import ( + MAX_JUDGE_SAMPLES, + execute_judge_review, + normalize_judge_review, + sample_round_indexes, +) +from agenteval.exploration.models import ( + ExplorationMessage, + ExplorationSession, + ExplorationSessionStatus, +) +from agenteval.models import Campaign, CampaignPlanEntry +from agenteval.storage.db import ModelConfigDB +from agenteval.storage.model_config_repository import ModelConfigRepository +from agenteval.storage.repository import ( + CampaignRepository, + ExplorationMessageRepository, + ExplorationSessionRepository, +) + + +class FakeChatClient: + """Queued-response fake for the judge 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 + + +REVIEW_JSON = json.dumps( + { + "dimensions": [ + {"dimension": "attitude", "rating": "good", "comment": "态度友好"}, + {"dimension": "professionalism", "rating": "acceptable", "comment": "流程基本正确"}, + {"dimension": "hallucination", "rating": "poor", "comment": "编造了不存在的政策"}, + ], + "summary": "服务态度好但存在幻觉", + }, + ensure_ascii=False, +) + + +def _campaign() -> Campaign: + return Campaign( + 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=1)], + ) + + +def _seed_config(db_session, config_id: str = "mc-default") -> None: + ModelConfigRepository(db_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=True, + is_analysis_default=True, + ) + ) + + +def _seed_session_with_messages(db_session, rounds: int = 5) -> str: + CampaignRepository(db_session).create(_campaign()) + repo = ExplorationSessionRepository(db_session) + session_obj = repo.create( + ExplorationSession( + campaign_id="camp-1", + target_id="t-1", + persona={"name": "急性子用户"}, + goal="查询账单并缴费", + ) + ) + session_obj.status = ExplorationSessionStatus.COMPLETED + repo.update(session_obj) + + message_repo = ExplorationMessageRepository(db_session) + for i in range(1, rounds + 1): + message_repo.save_message( + ExplorationMessage(session_id=session_obj.id, round_index=i, role="user", content=f"用户消息-{i}") + ) + message_repo.save_message( + ExplorationMessage(session_id=session_obj.id, round_index=i, role="assistant", content=f"回复内容-{i}") + ) + return session_obj.id + + +@pytest.fixture() +def judge_env(db_session, monkeypatch): + from agenteval.exploration import judge as judge_module + + monkeypatch.setattr(judge_module, "get_session", lambda: db_session) + return db_session + + +# ── 抽样 ────────────────────────────────────────────────────────────── + + +def test_sample_round_indexes_returns_all_within_cap(): + assert sample_round_indexes([1, 2]) == [1, 2] + assert sample_round_indexes([1, 2, 3]) == [1, 2, 3] + + +def test_sample_round_indexes_spreads_evenly_over_cap(): + assert sample_round_indexes([1, 2, 3, 4, 5]) == [1, 3, 5] + assert len(sample_round_indexes(list(range(1, 11)))) == MAX_JUDGE_SAMPLES + + +# ── 归一化 ──────────────────────────────────────────────────────────── + + +def test_normalize_judge_review_applies_whitelists(): + raw = { + "dimensions": [ + {"dimension": "attitude", "rating": "good", "comment": "ok"}, + {"dimension": "神秘维度", "rating": "good", "comment": "drop me"}, + {"dimension": "hallucination", "rating": "离谱", "comment": "bad rating"}, + ], + "summary": "结论", + } + review = normalize_judge_review(raw) + dims = {d["dimension"]: d for d in review["dimensions"]} + assert set(dims) == {"attitude", "hallucination"} + assert dims["attitude"]["rating"] == "good" + assert dims["hallucination"]["rating"] == "acceptable" # 非法档位归一 + assert review["summary"] == "结论" + + +def test_normalize_judge_review_rejects_non_list_dimensions(): + review = normalize_judge_review({"dimensions": "不是列表", "summary": 123}) + assert review["dimensions"] == [] + assert review["summary"] == "123" + + +# ── 后台执行编排 ────────────────────────────────────────────────────── + + +async def test_execute_persists_structured_review(judge_env): + session_id = _seed_session_with_messages(judge_env, rounds=5) + _seed_config(judge_env) + + client = FakeChatClient(REVIEW_JSON) + await execute_judge_review(session_id, chat_client=client) + + session_obj = ExplorationSessionRepository(judge_env).get(session_id) + assert session_obj.status == ExplorationSessionStatus.COMPLETED + review = session_obj.judge_review + assert review["status"] == "completed" + assert review["model_config_id"] == "mc-default" + assert review["sampled_rounds"] == [1, 3, 5] + assert len(review["dimensions"]) == 3 + assert review["summary"] == "服务态度好但存在幻觉" + + +async def test_execute_prompt_carries_only_sampled_rounds(judge_env): + session_id = _seed_session_with_messages(judge_env, rounds=5) + _seed_config(judge_env) + + client = FakeChatClient(REVIEW_JSON) + await execute_judge_review(session_id, chat_client=client) + + payload = json.dumps(client.calls[0], ensure_ascii=False) + assert "用户消息-1" in payload and "用户消息-3" in payload and "用户消息-5" in payload + assert "用户消息-2" not in payload and "用户消息-4" not in payload + + +async def test_execute_records_error_on_unparseable_output(judge_env): + session_id = _seed_session_with_messages(judge_env) + _seed_config(judge_env) + + await execute_judge_review(session_id, chat_client=FakeChatClient("这不是 JSON")) + + session_obj = ExplorationSessionRepository(judge_env).get(session_id) + assert session_obj.status == ExplorationSessionStatus.COMPLETED # 复核失败不阻塞会话 + review = session_obj.judge_review + assert review["status"] == "failed" + assert review["error"] + assert review["model_config_id"] == "mc-default" + + +async def test_execute_records_error_on_client_exception(judge_env): + session_id = _seed_session_with_messages(judge_env) + _seed_config(judge_env) + + await execute_judge_review(session_id, chat_client=FakeChatClient(RuntimeError("模型网关超时"))) + + review = ExplorationSessionRepository(judge_env).get(session_id).judge_review + assert review["status"] == "failed" + assert "模型网关超时" in review["error"] + + +async def test_execute_skips_silently_without_model_config(judge_env): + session_id = _seed_session_with_messages(judge_env) + + client = FakeChatClient(REVIEW_JSON) + await execute_judge_review(session_id, chat_client=client) + + assert client.calls == [] + assert ExplorationSessionRepository(judge_env).get(session_id).judge_review is None + + +async def test_execute_skips_session_without_messages(judge_env): + CampaignRepository(judge_env).create(_campaign()) + session_obj = ExplorationSessionRepository(judge_env).create( + ExplorationSession(campaign_id="camp-1", target_id="t-1", persona={}, goal="g") + ) + _seed_config(judge_env) + + client = FakeChatClient(REVIEW_JSON) + await execute_judge_review(session_obj.id, chat_client=client) + + assert client.calls == [] + assert ExplorationSessionRepository(judge_env).get(session_obj.id).judge_review is None