f3a528611e
架构保养第二轮候选 1：分析 / 周期对比 / judge 复核三条 LLM 任务链 收进各自的模块级 TaskRegistry（强引用防 GC、按 id 幂等、shutdown 统一收敛），删除 judge 的 _BACKGROUND_TASKS 私货，start_* 不再返回 无人消费的 Task。启动清理块补两笔 orphan 清扫：滞留的 generating 分析与对比行标记为 failed，与僵尸运行清扫同构。新增 7 个单测。
195 lines
7.4 KiB
Python
195 lines
7.4 KiB
Python
"""Judge sampling review for exploration sessions (judge 岗位抽样复核).
|
||
|
||
会话关闭后，平台对该会话对话抽样（默认 ≤3 段，控 token），经 judge 岗位模型
|
||
独立复核，产出质量维度结论（态度、专业性、幻觉）落入会话的 ``judge_review``。
|
||
复核是异步后台执行：失败落错误不阻塞会话状态，也不影响体验记录这条第一手
|
||
证据线；未配置模型时静默跳过。LLM 调用沿 v0.7 分析的 ``ChatClient`` 接缝
|
||
注入，测试用假客户端覆盖。
|
||
"""
|
||
|
||
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.task_registry import TaskRegistry
|
||
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()
|
||
|
||
|
||
judge_registry = TaskRegistry()
|
||
|
||
|
||
def start_judge_review(exploration_session_id: str) -> None:
|
||
"""以后台任务启动抽样复核（结果经会话 judge_review 观测）。
|
||
|
||
registry 持强引用防 GC，shutdown 时统一收敛；同 id 在跑时幂等不重复派生。
|
||
"""
|
||
judge_registry.launch(
|
||
exploration_session_id,
|
||
lambda _cancel: execute_judge_review(exploration_session_id),
|
||
)
|