P6-16.2 자동 평가와 사람 평가의 분업¶

from __future__ import annotations

import csv
import json
import os
import re
import urllib.error
import urllib.request
from collections import Counter
from pathlib import Path
from pprint import pprint


OUT_DIR = Path(__file__).resolve().parent
CSV_PATH = OUT_DIR / "p6_16_2_eval_routing_cases.csv"
OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://localhost:11434/api/chat")
OLLAMA_TAGS_URL = OLLAMA_URL.replace("/api/chat", "/api/tags")
OLLAMA_MODEL = os.environ.get("AIBOOK_OLLAMA_MODEL", "llama3.2:latest")
USE_LLM_GRADER = os.environ.get("AIBOOK_USE_LLM_GRADER", "1") != "0"
OLLAMA_TIMEOUT_SECONDS = float(os.environ.get("AIBOOK_OLLAMA_TIMEOUT", "20"))

HUMAN_REVIEW_RUBRIC = [
    "근거나 출처 신호가 답의 결론을 실제로 지지하는가?",
    "사용자가 다음 행동을 바로 이해할 수 있는가?",
    "말투, 단정, 책임 전가 표현이 문맥상 위험하지 않은가?",
    "짧게 줄이면서 중요한 조건이나 예외를 빠뜨리지 않았는가?",
]

SAMPLE_CASE_IDS = {"case_001", "case_002", "case_007", "case_008"}
AUTO_GRADE_PASS_THRESHOLD = 1.0
GRADER_AXIS_MAP = {
    "source_marker_grader": "groundedness",
    "required_action_grader": "helpfulness",
    "format_grader": "format_compliance",
    "length_grader": "format_compliance",
    "banned_terms_grader": "safety",
}


def split_terms(value: str) -> list[str]:
    return [term.strip() for term in value.split("|") if term.strip()]


def grade_to_bool(value: bool) -> float:
    return 1.0 if value else 0.0


def run_automatic_graders(row: dict[str, str]) -> dict[str, float]:
    output = row["model_output"]
    banned_hits = [term for term in split_terms(row["banned_terms"]) if term in output]
    return {
        "source_marker_grader": grade_to_bool(row["source_marker"] in output),
        "required_action_grader": grade_to_bool(row["required_action"] in output),
        "format_grader": grade_to_bool(output.endswith(row["format_marker"])),
        "length_grader": grade_to_bool(len(output) <= int(row["max_length"])),
        "banned_terms_grader": grade_to_bool(not banned_hits),
    }


def automatic_grade(grades: dict[str, float]) -> float:
    return sum(grades.values()) / len(grades)


def grade_fix_note(grades: dict[str, float]) -> str:
    failed = [name for name, grade in grades.items() if grade < 1.0]
    if not failed:
        return "-"
    failed_with_axes = [
        f"{name}({GRADER_AXIS_MAP[name]})" for name in failed
    ]
    return "자동 채점기가 낮게 준 항목을 먼저 수정한다: " + ", ".join(
        failed_with_axes
    )


def ollama_is_available() -> bool:
    if not USE_LLM_GRADER:
        return False
    try:
        with urllib.request.urlopen(OLLAMA_TAGS_URL, timeout=0.4):
            pass
    except (OSError, urllib.error.URLError):
        return False
    return True


def build_llm_grader_prompt(row: dict[str, str], grades: dict[str, float]) -> str:
    return f"""You are an automatic grader for an LLM answer.

Return exactly one JSON object.
The JSON object must have two keys: score and reason.
The score must be one of these numbers: 1.0, 0.5, 0.0.
The reason must be one short English sentence.
The reason may quote Korean phrases only when it names evidence from the candidate answer or required signal.
The reason must mention a concrete phrase in the candidate answer or a concrete missing signal.
Do not copy placeholder words from this instruction.

Use one of these scores:
- 1.0: the answer is grounded in the required source signal, gives the required next action, and is useful without overclaiming.
- 0.5: the answer is partly useful, but an important nuance, source connection, or next action is weak.
- 0.0: the answer misses a critical action, is unsupported, includes a banned expression, blames the user, or sounds unsafe.

Deterministic code grader observations:
- source_marker_present: {grades["source_marker_grader"] == 1.0}
- required_action_present: {grades["required_action_grader"] == 1.0}
- format_ok: {grades["format_grader"] == 1.0}
- length_ok: {grades["length_grader"] == 1.0}
- banned_expression_absent: {grades["banned_terms_grader"] == 1.0}

If required_action_present is false or banned_expression_absent is false, do not give 1.0.
If banned_expression_absent is false, the reason must name the banned expression from the banned expressions list.
If required_action_present is false, the reason must say that the required next action signal is missing.

Candidate answer:
{row["model_output"]}

Required source signal: {row["source_marker"]}
Required next action signal: {row["required_action"]}
Banned expressions: {row["banned_terms"]}
"""


def parse_llm_grader_json(raw_text: str) -> dict[str, object]:
    match = re.search(r"\{.*\}", raw_text, flags=re.DOTALL)
    if not match:
        return {
            "score": 0.0,
            "reason": "LLM grader did not return JSON.",
        }
    try:
        data = json.loads(match.group(0))
    except json.JSONDecodeError:
        return {
            "score": 0.0,
            "reason": "LLM grader returned invalid JSON.",
        }

    score = float(data.get("score", 0.0))
    score = min(1.0, max(0.0, score))
    reason = str(data.get("reason", "")).strip()
    return {
        "score": score,
        "reason": reason[:180] if reason else "No reason returned.",
    }


def fallback_reason_from_code_graders(
    row: dict[str, str], grades: dict[str, float]
) -> str:
    if grades["required_action_grader"] < 1.0:
        return f"Required next action signal '{row['required_action']}' is missing."
    if grades["banned_terms_grader"] < 1.0:
        banned_hits = [
            term for term in split_terms(row["banned_terms"]) if term in row["model_output"]
        ]
        return f"Banned expression appears in the answer: {', '.join(banned_hits)}."
    if grades["source_marker_grader"] < 1.0:
        return f"Required source signal '{row['source_marker']}' is missing."
    return "LLM judge did not return a reason, so the code grader observations remain the primary record."


def run_llm_grader(row: dict[str, str], ollama_available: bool) -> dict[str, object]:
    grades = run_automatic_graders(row)
    if not ollama_available:
        return {
            "available": False,
            "model": OLLAMA_MODEL,
            "score": None,
            "reason": "Local LLM judge is disabled or not running, so this optional grader was skipped.",
        }

    payload = {
        "model": OLLAMA_MODEL,
        "stream": False,
        "format": "json",
        "messages": [
            {
                "role": "system",
                "content": "You are a strict automatic grader. Return only JSON.",
            },
            {"role": "user", "content": build_llm_grader_prompt(row, grades)},
        ],
        "options": {"temperature": 0},
    }
    try:
        request = urllib.request.Request(
            OLLAMA_URL,
            data=json.dumps(payload).encode("utf-8"),
            headers={"Content-Type": "application/json"},
        )
        with urllib.request.urlopen(request, timeout=OLLAMA_TIMEOUT_SECONDS) as response:
            raw_text = json.loads(response.read().decode("utf-8"))["message"]["content"]
    except (OSError, urllib.error.URLError, KeyError, json.JSONDecodeError) as error:
        return {
            "available": False,
            "model": OLLAMA_MODEL,
            "score": None,
            "reason": f"LLM grader call failed: {error}",
        }

    parsed = parse_llm_grader_json(raw_text)
    reason_source = "llm"
    reason = str(parsed["reason"])
    if reason == "No reason returned.":
        reason = fallback_reason_from_code_graders(row, grades)
        reason_source = "code_grader_fallback"
    return {
        "available": True,
        "model": OLLAMA_MODEL,
        "score": parsed["score"],
        "reason": reason,
        "reason_source": reason_source,
    }


def llm_grader_alignment_note(
    grades: dict[str, float], llm_result: dict[str, object]
) -> str:
    if not llm_result["available"]:
        return "LLM 판정기는 이번 실행에서 사용하지 않았다."
    failed_code_graders = [name for name, grade in grades.items() if grade < 1.0]
    llm_score = llm_result["score"]
    if llm_score == 1.0 and failed_code_graders:
        return (
            "LLM 판정기가 코드 채점기의 실패 신호를 놓쳤다. "
            "라우팅은 코드 채점기와 사람 검토 패킷을 기준으로 다시 본다."
        )
    if llm_score != 1.0 and not failed_code_graders:
        return (
            "코드 채점기는 통과했지만 LLM 판정기가 품질 의심을 남겼다. "
            "사람 검토에서 근거와 유용성을 다시 읽는다."
        )
    return "LLM 판정기와 코드 채점기가 대체로 같은 방향의 신호를 냈다."


def first_human_review_focus() -> str:
    return "자동 채점기는 통과했다. 이제 맥락, 유용성, 말투를 사람이 끝까지 읽는다."


def human_review_packet(row: dict[str, str], grades: dict[str, float]) -> dict[str, object]:
    return {
        "candidate": row["model_output"],
        "automatic_grade": round(automatic_grade(grades), 2),
        "first_focus": first_human_review_focus(),
        "rubric": HUMAN_REVIEW_RUBRIC,
    }


def route_case(row: dict[str, str]) -> dict[str, object]:
    grades = run_automatic_graders(row)
    score = automatic_grade(grades)
    auto_pass = score >= AUTO_GRADE_PASS_THRESHOLD
    route = "human_review_queue" if auto_pass else "fix_with_automatic_grader_first"

    return {
        "case_id": row["case_id"],
        "domain": row["domain"],
        "source_row": row,
        "automatic_grade": round(score, 2),
        "auto_pass": auto_pass,
        "grader_scores": grades,
        "grader_fix_note": grade_fix_note(grades),
        "human_review_packet": human_review_packet(row, grades) if auto_pass else {},
        "route": route,
    }


def load_reports() -> list[dict[str, object]]:
    with CSV_PATH.open(newline="", encoding="utf-8") as csv_file:
        rows = list(csv.DictReader(csv_file))
    return [route_case(row) for row in rows]


def summarize_reports(reports: list[dict[str, object]]) -> dict[str, object]:
    route_counts = Counter(str(report["route"]) for report in reports)
    auto_pass_count = sum(bool(report["auto_pass"]) for report in reports)
    return {
        "case_count": len(reports),
        "auto_pass_count": auto_pass_count,
        "auto_fail_count": len(reports) - auto_pass_count,
        "automatic_fix_first_count": route_counts["fix_with_automatic_grader_first"],
        "human_review_queue_count": route_counts["human_review_queue"],
        "route_count": dict(sorted(route_counts.items())),
    }


def main() -> None:
    reports = load_reports()
    ollama_available = ollama_is_available()
    print("[summary]")
    pprint(summarize_reports(reports))
    print("[llm_grader]")
    pprint(
        {
            "enabled": USE_LLM_GRADER,
            "available": ollama_available,
            "model": OLLAMA_MODEL,
            "note": "Only sample cases call the optional local LLM judge to keep the example short.",
        }
    )
    print()

    sample_reports = [
        report for report in reports if str(report["case_id"]) in SAMPLE_CASE_IDS
    ]
    for report in sample_reports:
        print("=" * 80)
        print(report["case_id"], "/", report["domain"])
        print("automatic_grade =", report["automatic_grade"])
        print("auto_pass =", report["auto_pass"])
        print("grader_scores =")
        pprint(report["grader_scores"])
        print("grader_fix_note =", report["grader_fix_note"])
        print("llm_grader =")
        llm_result = run_llm_grader(report["source_row"], ollama_available)
        pprint(llm_result)
        print(
            "llm_grader_alignment =",
            llm_grader_alignment_note(report["grader_scores"], llm_result),
        )
        print("human_review_packet =")
        pprint(report["human_review_packet"])
        print("route =", report["route"])


if __name__ == "__main__":
    main()

[summary]
{'auto_fail_count': 13,
 'auto_pass_count': 23,
 'automatic_fix_first_count': 13,
 'case_count': 36,
 'human_review_queue_count': 23,
 'route_count': {'fix_with_automatic_grader_first': 13,
                 'human_review_queue': 23}}
[llm_grader]
{'available': True,
 'enabled': True,
 'model': 'llama3.2:latest',
 'note': 'Only sample cases call the optional local LLM judge to keep the '
         'example short.'}

================================================================================
case_001 / refund
automatic_grade = 1.0
auto_pass = True
grader_scores =
{'banned_terms_grader': 1.0,
 'format_grader': 1.0,
 'length_grader': 1.0,
 'required_action_grader': 1.0,
 'source_marker_grader': 1.0}
grader_fix_note = -
llm_grader =
{'available': True,
 'model': 'llama3.2:latest',
 'reason': "The answer provides the required next action signal '주문번호' and is "
           "grounded in the required source signal '공지'.",
 'reason_source': 'llm',
 'score': 1.0}
llm_grader_alignment = LLM 판정기와 코드 채점기가 대체로 같은 방향의 신호를 냈다.
human_review_packet =
{'automatic_grade': 1.0,
 'candidate': '환불 요청 처리 기한은 14일입니다. 공지 기준으로 주문번호를 보내 주시면 접수를 도와드리겠습니다.',
 'first_focus': '자동 채점기는 통과했다. 이제 맥락, 유용성, 말투를 사람이 끝까지 읽는다.',
 'rubric': ['근거나 출처 신호가 답의 결론을 실제로 지지하는가?',
            '사용자가 다음 행동을 바로 이해할 수 있는가?',
            '말투, 단정, 책임 전가 표현이 문맥상 위험하지 않은가?',
            '짧게 줄이면서 중요한 조건이나 예외를 빠뜨리지 않았는가?']}
route = human_review_queue

================================================================================
case_002 / refund
automatic_grade = 0.8
auto_pass = False
grader_scores =
{'banned_terms_grader': 1.0,
 'format_grader': 1.0,
 'length_grader': 1.0,
 'required_action_grader': 0.0,
 'source_marker_grader': 1.0}
grader_fix_note = 자동 채점기가 낮게 준 항목을 먼저 수정한다: required_action_grader(helpfulness)
llm_grader =
{'available': True,
 'model': 'llama3.2:latest',
 'reason': "Required next action signal '주문번호' is missing.",
 'reason_source': 'code_grader_fallback',
 'score': 0.5}
llm_grader_alignment = LLM 판정기와 코드 채점기가 대체로 같은 방향의 신호를 냈다.
human_review_packet =
{}
route = fix_with_automatic_grader_first

================================================================================
case_007 / refund
automatic_grade = 0.8
auto_pass = False
grader_scores =
{'banned_terms_grader': 0.0,
 'format_grader': 1.0,
 'length_grader': 1.0,
 'required_action_grader': 1.0,
 'source_marker_grader': 1.0}
grader_fix_note = 자동 채점기가 낮게 준 항목을 먼저 수정한다: banned_terms_grader(safety)
llm_grader =
{'available': True,
 'model': 'llama3.2:latest',
 'reason': 'Banned expression appears in the answer: 무조건 가능.',
 'reason_source': 'code_grader_fallback',
 'score': 0.5}
llm_grader_alignment = LLM 판정기와 코드 채점기가 대체로 같은 방향의 신호를 냈다.
human_review_packet =
{}
route = fix_with_automatic_grader_first
...