"""Julgamento por rubrica ancorada: júri de 3 juízes, divergência e adjudicação.""" from __future__ import annotations import json import re TEMPLATE = { "pt": ( "Você é um avaliador especialista. Pontue a resposta abaixo em relação a cada " "critério da rubrica, em escala de 0 a 5, onde 0 = critério ausente ou incorreto, " "3 = parcialmente atendido, 5 = plenamente atendido com precisão técnica.\n" "Regras: (1) pontue cada critério isoladamente; (2) cite o trecho da resposta que " "sustenta a nota; (3) não pontue por extensão, estilo ou fluência; (4) não reescreva " "a resposta; (5) não conceda ponto por informação correta fora do critério.\n\n" "PERGUNTA:\n{pergunta}\n\nRUBRICA:\n{rubrica}\n\nRESPOSTA A AVALIAR:\n{resposta}\n\n" "Responda em JSON: {{\"notas\": [n1, n2, ...], \"justificativas\": [\"...\", ...]}}" ), "en": ( "You are an expert grader. Score the answer below against each rubric criterion on " "a 0-5 scale, where 0 = absent or incorrect, 3 = partially met, 5 = fully met with " "technical precision.\n" "Rules: (1) score each criterion separately; (2) quote the passage supporting the " "score; (3) do not score length, style or fluency; (4) do not rewrite the answer; " "(5) do not award points for correct information outside the criterion.\n\n" "QUESTION:\n{question}\n\nRUBRIC:\n{rubric}\n\nANSWER TO GRADE:\n{answer}\n\n" "Reply in JSON: {{\"scores\": [n1, n2, ...], \"justifications\": [\"...\", ...]}}" ), "zh": ( "你是一名专家评审。请对照评分标准的每一项,对下面的回答按 0-5 分打分:" "0 = 缺失或错误,3 = 部分满足,5 = 完全满足且技术准确。\n" "规则:(1) 逐项独立打分;(2) 引用支持该分数的原文;(3) 不因篇幅、风格或流畅度加分;" "(4) 不得改写回答;(5) 不对标准之外的正确信息加分。\n\n" "题目:\n{question}\n\n评分标准:\n{rubric}\n\n待评回答:\n{answer}\n\n" "请以 JSON 返回:{{\"scores\": [n1, n2, ...], \"justifications\": [\"...\", ...]}}" ), } TEMPLATE_ADJ = { "pt": ( "Três avaliadores divergiram sobre a mesma resposta. Reavalie cada critério em " "escala 0 a 5 e justifique por que a sua nota é a mais defensável.\n\n" "PERGUNTA:\n{pergunta}\n\nRUBRICA:\n{rubrica}\n\nRESPOSTA:\n{resposta}\n\n" "NOTAS ANTERIORES: {notas}\n\n" "JSON: {{\"notas\": [...], \"justificativas\": [...]}}" ), "en": ( "Three graders disagreed on the same answer. Regrade each criterion on a 0-5 scale " "and justify why your score is the most defensible.\n\n" "QUESTION:\n{question}\n\nRUBRIC:\n{rubric}\n\nANSWER:\n{answer}\n\n" "PRIOR SCORES: {notas}\n\n" "JSON: {{\"scores\": [...], \"justifications\": [...]}}" ), "zh": ( "三位评审对同一回答分歧较大。请按 0-5 分重新评分,并说明你的分数为何最合理。\n\n" "题目:\n{question}\n\n评分标准:\n{rubric}\n\n回答:\n{answer}\n\n" "此前分数:{notas}\n\n" "JSON:{{\"scores\": [...], \"justifications\": [...]}}" ), } def _rubrica_texto(item: dict) -> str: return "\n".join(f"{i + 1}. {c}" for i, c in enumerate(item.get("rubric", []))) def _fmt(tpl: str, item: dict, resposta: str, notas: object = None) -> str: out = tpl for chave, valor in ( ("{pergunta}", item["q"]), ("{question}", item["q"]), ("{rubrica}", _rubrica_texto(item)), ("{resposta}", resposta), ("{answer}", resposta), ("{notas}", json.dumps(notas if notas is not None else [])), ): out = out.replace(chave, str(valor)) return out def prompt_rubrica(item: dict, resposta: str, lang: str) -> str: return _fmt(TEMPLATE[lang], item, resposta) def prompt_adjudicacao(item: dict, resposta: str, lang: str, notas: object) -> str: return _fmt(TEMPLATE_ADJ[lang], item, resposta, notas) def extrair_notas(texto: str) -> list[float] | None: """Extrai a lista de notas de uma resposta de juiz (JSON ou lista solta).""" if not texto: return None m = re.search(r"\{.*\}", texto, re.S) for candidato in ([m.group(0)] if m else []) + [texto]: try: dados = json.loads(candidato) except json.JSONDecodeError: continue if isinstance(dados, dict): for chave in ("notas", "scores"): if isinstance(dados.get(chave), list): vals = [float(v) for v in dados[chave] if isinstance(v, (int, float))] if vals: return [max(0.0, min(5.0, v)) for v in vals] soltos = re.findall(r"\b([0-5](?:\.\d)?)\b", texto) if soltos: return [max(0.0, min(5.0, float(v))) for v in soltos[:8]] return None def _notas_juiz(juiz, item: dict, resposta: str, lang: str, prompt: str) -> list[float]: n_crit = len(item.get("rubric", [])) if hasattr(juiz, "pontuar"): # MockJudge return juiz.pontuar(item, resposta, lang)["notas"] texto = juiz.completa(prompt, "", lang)["text"] notas = extrair_notas(texto) if notas is None: return [0.0] * n_crit notas = (notas + [0.0] * n_crit)[:n_crit] return [max(0.0, min(5.0, v)) for v in notas] def pontuar_item(item: dict, resposta: str, lang: str, juizes: list, adjudicador, limite: float) -> dict: """Aplica o júri de 3 juízes e, se a divergência superar o limite, o adjudicador.""" n_crit = len(item.get("rubric", [])) if n_crit == 0: return {"notas_por_juiz": [], "medias": [], "dispersao": 0.0, "adjudicado": False} notas_por_juiz = [ _notas_juiz(j, item, resposta, lang, prompt_rubrica(item, resposta, lang)) for j in juizes ] medias = [ round(sum(n[j] for n in notas_por_juiz) / len(notas_por_juiz), 3) for j in range(n_crit) ] dispersao = max( (max(n[j] for n in notas_por_juiz) - min(n[j] for n in notas_por_juiz)) for j in range(n_crit) ) adjudicado = False if dispersao > limite and adjudicador is not None: notas = _notas_juiz( adjudicador, item, resposta, lang, prompt_adjudicacao(item, resposta, lang, notas_por_juiz), ) medias = [round(v, 3) for v in notas] adjudicado = True return { "notas_por_juiz": notas_por_juiz, "medias": medias, "dispersao": dispersao, "adjudicado": adjudicado, }