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.
This commit is contained in:
sinohqb
parent
3e485bfbe6
commit
2285a25009
187
backend/agenteval/exploration/judge.py
Normal file
187
backend/agenteval/exploration/judge.py
Normal file
@ -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
|
||||
@ -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")
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
|
||||
239
tests/unit/test_exploration_judge.py
Normal file
239
tests/unit/test_exploration_judge.py
Normal file
@ -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
|
||||
Loading…
Reference in New Issue
Block a user