bitflicker64 commented on code in PR #409: URL: https://github.com/apache/hugegraph-ai/pull/409#discussion_r4505777993
########## graphrag/eval/src/hugegraph_llm_eval/utils/text_normalize.py: ########## @@ -0,0 +1,158 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Text normalization utilities for text evaluation. + +Implements the MRQA official evaluation standard from HippoRAG 2 (eval_utils.py) +with Porter stemming and bilingual (EN/ZH) support. + +For Chinese text the normalization additionally: +- converts full-width ASCII characters (letters, digits, punctuation) to half-width +- performs simplified/traditional Chinese conversion when ``opencc`` is available +- removes Chinese punctuation and collapses whitespace +""" + +import re +import string +from typing import cast + +# Chinese punctuation set +_CHINESE_PUNCTUATION = set("，。！？、；：''（）【】《》〈〉…—～·「」『』〔〕") +# English stop words removed during normalization (MRQA standard) +_ARTICLES_PATTERN = re.compile(r"\b(a|an|the)\b") + +# Full-width ASCII block: U+FF01..U+FF5E map to U+0021..U+007E. +_FULLWIDTH_SPACE = " " # full-width space + + +def _to_halfwidth(text: str) -> str: + """Convert full-width ASCII characters to their half-width forms. + + Covers full-width letters, digits, punctuation and the full-width space. + This unifies mixed full/half-width text (e.g. Chinese manuals often contain + full-width numbers and letters) before comparison. + """ + # Full-width ASCII block U+FF01..U+FF5E maps to U+0021..U+007E. + table = {0xFF01 + i: 0x0021 + i for i in range(94)} + table[ord(_FULLWIDTH_SPACE)] = ord(" ") + return text.translate(table) + + +def _simplify_chinese(text: str) -> str: + """Convert traditional Chinese characters to simplified forms if possible. + + Uses ``opencc-python-reimplemented`` / ``opencc`` when installed. If the + library is not available the text is returned unchanged so evaluation + keeps working without extra dependencies. + """ + try: + # opencc-python-reimplemented exposes OpenCC in the same way + from opencc import OpenCC # type: ignore[import-not-found] # ty: ignore[import-not-found] + + converter = OpenCC("t2s") + return cast(str, converter.convert(text)) + except Exception: + return text + + +def normalize_answer(answer: str, language: str = "en") -> str: + """Normalize an answer string for comparison. + + Steps (EN): lowercase → remove punctuation → remove articles + (a/an/the) → collapse whitespace. + Steps (ZH): full-width to half-width → traditional to simplified Chinese + → lowercase → remove punctuation (incl. Chinese) → collapse whitespace. + + Reference: HippoRAG 2 / MRQA official eval_utils.normalize_answer + (standard SQuAD normalization: lowercase, remove punctuation, remove + a/an/the, collapse whitespace). Note: ``and`` is a conjunction, not an + article, and is intentionally NOT removed. + + Args: + answer: Raw answer text. + language: 'en' for English, 'zh' for Chinese. + + Returns: + Normalized string. + """ + if not answer: + return "" + + def _preprocess(text: str) -> str: + # Language-specific preprocessing before shared normalization. + if language == "zh": + text = _to_halfwidth(text) + text = _simplify_chinese(text) + return text + + def _lower(text: str) -> str: + return text.lower() + + def _remove_punc(text: str) -> str: + exclude = set(string.punctuation) | _CHINESE_PUNCTUATION + return "".join(ch for ch in text if ch not in exclude) + + def _remove_articles(text: str) -> str: + if language == "en": + return _ARTICLES_PATTERN.sub(" ", text) + return text + + def _white_space_fix(text: str) -> str: + return " ".join(text.split()) + + return _white_space_fix(_remove_articles(_remove_punc(_lower(_preprocess(answer))))) + + +def tokenize(text: str, language: str = "en", stem: bool = False) -> list[str]: + """Tokenize text into words, optionally with stemming. + + For English: split on whitespace after normalization. + For Chinese: use jieba segmentation. + + Args: + text: Raw text to tokenize. + language: 'en' or 'zh'. + stem: If True, apply Porter stemmer to English tokens (HippoRAG 2 standard). + + Returns: + List of tokens. + """ + if language == "zh": + import jieba # type: ignore[import-untyped] + + return list(jieba.cut(normalize_answer(text, language))) Review Comment: Important: `jieba.cut` keeps the spaces that `normalize_answer` leaves as tokens, so zh ROUGE-L matches whitespace: `_chinese_rouge("苹果 香蕉 橙子", "苹果 葡萄 西瓜")` scores F1 0.6, against 0.3333 for the same words without spaces. Drop whitespace-only tokens before returning. ########## graphrag/eval/src/hugegraph_llm_eval/integrations/graph_extraction/evaluators/semantic.py: ########## @@ -0,0 +1,542 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""LLM semantic entity and triple matching with deterministic scoring.""" + +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from datetime import datetime +from itertools import islice +from typing import Literal + +from pydantic import Field, JsonValue, model_validator + +from hugegraph_llm_eval.contracts.base import StrictModel +from hugegraph_llm_eval.contracts.digests import canonical_digest, canonical_json_bytes +from hugegraph_llm_eval.contracts.domain.evaluation import ( + EvaluationContext, + EvaluationInput, + EvaluationRecordDraft, + EvaluationRecordStatus, + EvaluatorDescriptor, +) +from hugegraph_llm_eval.contracts.domain.graph import GraphCandidate, GraphEdge, GraphVertex +from hugegraph_llm_eval.contracts.domain.judge import JudgeCallRecord +from hugegraph_llm_eval.contracts.errors import FailureCategory, FailureDetail +from hugegraph_llm_eval.contracts.identity import EvaluatorRef, SchemaRef +from hugegraph_llm_eval.evaluation.judges.base import ( + Judge, + JudgeCallError, + JudgeMessage, + JudgeOptions, + JudgeRequest, +) +from hugegraph_llm_eval.evaluation.judges.batching import BatchingOptions, BatchInputError, paired_payload_batches +from hugegraph_llm_eval.evaluation.judges.content import JudgeContent +from hugegraph_llm_eval.evaluation.judges.session import JudgeSession +from hugegraph_llm_eval.integrations.graph_extraction.evaluators.semantic_scoring import ( + MatchPair, + global_optimal_matches, + semantic_findings, + semantic_measurements, +) +from hugegraph_llm_eval.integrations.graph_extraction.materializers import ( + GRAPH_EXTRACTION_INPUT_SCHEMA, + GraphArtifactMaterializer, + GraphMaterializationError, +) + +GraphItem = GraphVertex | GraphEdge +SemanticMatchIssue = Literal[ + "label_mismatch", + "name_mismatch", + "property_mismatch", + "relation_mismatch", + "endpoint_mismatch", + "endpoint_label_mismatch", + "direction_mismatch", + "no_semantic_match", +] + + +class SemanticVerdict(StrictModel): + """One proposed semantic pair, or a reference with no match in this batch.""" + + reference_index: int = Field(ge=0) + candidate_index: int | None = Field(default=None, ge=0) + match_score: float | None = Field(default=None, ge=0, le=100) + match_issues: tuple[SemanticMatchIssue, ...] + reason: str = Field(min_length=1, max_length=4_000) + + @model_validator(mode="after") + def _match_score_presence(self) -> "SemanticVerdict": + """Require a score exactly when the verdict contains a match.""" + if self.candidate_index is None and self.match_score is not None: + raise ValueError("unmatched semantic verdict must not contain a match score") + if self.candidate_index is not None and self.match_score is None: + raise ValueError("matched semantic verdict must contain a match score") + self._validate_match_issues() + return self + + def _validate_match_issues(self) -> None: + """Reject duplicate issue labels and hard conflicts on accepted matches.""" + if len(self.match_issues) != len(set(self.match_issues)): + raise ValueError("semantic verdict match issues must be unique") + if self.candidate_index is None and "no_semantic_match" not in self.match_issues: + raise ValueError("unmatched semantic verdict must contain no_semantic_match") + if self.candidate_index is not None and { + "no_semantic_match", + "name_mismatch", + }.intersection(self.match_issues): + raise ValueError("matched semantic verdict cannot contain unmatched or name mismatch issues") + if self.candidate_index is not None and { + "endpoint_mismatch", + "direction_mismatch", + }.intersection(self.match_issues): + raise ValueError("matched semantic verdict cannot contain endpoint or direction mismatch issues") + + +class SemanticVerdictOutput(StrictModel): + verdicts: tuple[SemanticVerdict, ...] + + @model_validator(mode="after") + def _consistent_pairs(self) -> "SemanticVerdictOutput": + candidates_by_reference: dict[int, set[int | None]] = {} + for item in self.verdicts: + candidates = candidates_by_reference.setdefault(item.reference_index, set()) + if item.candidate_index in candidates: + raise ValueError("semantic verdict candidate/reference pairs must be unique") + candidates.add(item.candidate_index) + if None in candidates and len(candidates) > 1: + raise ValueError("unmatched semantic verdict cannot accompany matches for the same reference") + return self + + +@dataclass(frozen=True, slots=True) +class _SemanticSpec: + kind: Literal["entity", "triple"] + invocation_key: str + evaluator: EvaluatorRef + + +@dataclass(frozen=True, slots=True) +class _SemanticBatchResult: + """Result of one independent candidate-batch judge call.""" + + output: SemanticVerdictOutput | None + judge_calls: tuple[JudgeCallRecord, ...] + failure: FailureDetail | None = None + + +class _SemanticGraphEvaluator: + input_type = EvaluationInput + + def __init__( + self, + materializer: GraphArtifactMaterializer, + spec: _SemanticSpec, + *, + judge: Judge, + prompt: JudgeContent, + rubric: JudgeContent, + clock: Callable[[], datetime], + batching: BatchingOptions | None = None, + matching_strategy: Literal["max_weight", "max_cardinality"] = "max_weight", + ) -> None: + if matching_strategy not in ("max_weight", "max_cardinality"): + raise ValueError("matching_strategy must be 'max_weight' or 'max_cardinality'") + self._batching = batching or BatchingOptions() + self._matching_strategy = matching_strategy + self._materializer = materializer + self._spec = spec + self._judge = judge + self._prompt = prompt + self._rubric = rubric + self._clock = clock + self.descriptor = EvaluatorDescriptor( + evaluator=spec.evaluator, + implementation_digest=canonical_digest( + { + "algorithm": f"graph-semantic-{spec.kind}", + "version": 9, + "matching": "two-sided-window-alternative-pairs-global-one-to-one", + "matching_strategy": matching_strategy, + "batching": self._batching.model_dump(mode="json"), + "judge": judge.descriptor.model_dump(mode="json"), + "prompt": prompt.reference.model_dump(mode="json"), + "rubric": rubric.reference.model_dump(mode="json"), + } + ), + input_schema=GRAPH_EXTRACTION_INPUT_SCHEMA, + output_schema=SchemaRef(namespace="graph", name=f"semantic_{spec.kind}_record", version=1), + ) + + @property + def invocation_key(self) -> str: + return self._spec.invocation_key + + def evaluate(self, evaluation_input: EvaluationInput, context: EvaluationContext) -> EvaluationRecordDraft: + try: + value = self._materializer.extraction(evaluation_input) + except GraphMaterializationError as exc: + return self._failure("graph_extraction_materialization_failed", str(exc), FailureCategory.INVALID_INPUT) + if context.cancellation.is_cancelled: + return self._failure( + "evaluation_cancelled", + context.cancellation.reason or "evaluation cancelled", + FailureCategory.CANCELLED, + ) + candidate = _items(value.candidate, self._spec.kind) + reference = _items(value.reference, self._spec.kind) + if not candidate or not reference: Review Comment: Important: when candidate and reference both have no items, such as `edges: []` on a vertex-only chunk for the triple metric, the record completes with precision, recall and F1 of 0.0, scoring exact agreement as total failure. Return `not_applicable` (or 1.0) when both sides are empty. ########## graphrag/eval/src/hugegraph_llm_eval/integrations/retrieval/evaluators/coverage.py: ########## @@ -0,0 +1,332 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""LLM-judged context entity recall and evidence attribution Evaluators.""" + +from collections.abc import Callable +from datetime import datetime +from typing import Annotated + +from pydantic import Field, model_validator + +from hugegraph_llm_eval.contracts.base import StrictModel +from hugegraph_llm_eval.contracts.domain.artifacts import ArtifactRef +from hugegraph_llm_eval.contracts.domain.evaluation import ( + EvaluationContext, + EvaluationInput, + EvaluationRecordDraft, + EvidenceLink, + Finding, + FindingSeverity, +) +from hugegraph_llm_eval.contracts.errors import FailureCategory +from hugegraph_llm_eval.contracts.identity import ComponentRef +from hugegraph_llm_eval.evaluation.judges.base import Judge, JudgeCallError +from hugegraph_llm_eval.evaluation.judges.batching import BatchingOptions, BatchInputError, payload_batches +from hugegraph_llm_eval.evaluation.judges.content import JudgeContent +from hugegraph_llm_eval.evaluation.judges.session import JudgeSession +from hugegraph_llm_eval.integrations.retrieval.evaluators.judged_common import ( + JudgedRetrievalEvaluator, + indexed, + ratio, + require_unique_indices, + subject, +) +from hugegraph_llm_eval.integrations.retrieval.materializers import RetrievalArtifactMaterializer +from hugegraph_llm_eval.utils.text_preview import text_preview + +_ENTITY_EXTRACTION_INSTRUCTION = ( + "Extract every distinct entity from reference_answer, without deciding whether it is covered. " + 'Return only {"entities": ["entity name"]}; preserve named entities, include unsupported ones, ' + "and do not invent entities. The caller will freeze this inventory before evaluating retrieval." +) +_ENTITY_COVERAGE_INSTRUCTION = ( + "The supplied entities are the fixed reference inventory. Return has_reference_entities and verdicts. " + "For EVERY supplied entity_index return exactly one verdict with the same entity text, covered and reason. " + "Do not add, remove, rename or renumber entities, including those not covered by any retrieved context." +) + + +class ReferenceEntitiesOutput(StrictModel): + entities: tuple[Annotated[str, Field(min_length=1, max_length=4_000)], ...] + + @model_validator(mode="after") + def _unique_entities(self) -> "ReferenceEntitiesOutput": + if len(self.entities) != len(set(self.entities)): + raise ValueError("reference entity inventory must be unique") + return self + + +class ContextEntityVerdict(StrictModel): + """LLM verdict for one answer entity against all retrieved contexts.""" + + entity_index: int = Field(ge=0) + entity: str = Field(min_length=1, max_length=4_000) + covered: bool + reason: str = Field(min_length=1, max_length=4_000) + + +class ContextEntityRecallOutput(StrictModel): + """Semantic coverage judgements against a fixed reference entity inventory.""" + + has_reference_entities: bool + verdicts: tuple[ContextEntityVerdict, ...] + + @model_validator(mode="after") + def _complete_entity_decision(self) -> "ContextEntityRecallOutput": + require_unique_indices(tuple(item.entity_index for item in self.verdicts)) + if self.has_reference_entities != bool(self.verdicts): + raise ValueError("entity verdicts must agree with has_reference_entities") + return self + + +class EvidenceVerdict(StrictModel): + evidence_index: int = Field(ge=0) + attributed: bool + reason: str = Field(min_length=1, max_length=4_000) + + +class EvidenceRecallOutput(StrictModel): + verdicts: tuple[EvidenceVerdict, ...] + + @model_validator(mode="after") + def _unique_indices(self) -> "EvidenceRecallOutput": + require_unique_indices(tuple(item.evidence_index for item in self.verdicts)) + return self + + +class ContextEntityRecallEvaluator(JudgedRetrievalEvaluator): + def __init__( + self, + materializer: RetrievalArtifactMaterializer, + *, + judge: Judge, + prompt: JudgeContent, + rubric: JudgeContent, + clock: Callable[[], datetime], + batching: BatchingOptions | None = None, + ) -> None: + super().__init__( + materializer, + invocation_key="context_entity_recall_llm", + judge=judge, + prompt=prompt, + rubric=rubric, + clock=clock, + batching=batching, + implementation={ + "entity_protocol": "fixed-inventory-v1", + "inventory_prompt": _ENTITY_EXTRACTION_INSTRUCTION, + "coverage_prompt": _ENTITY_COVERAGE_INSTRUCTION, + }, + ) + + def evaluate(self, evaluation_input: EvaluationInput, context: EvaluationContext) -> EvaluationRecordDraft: + value = self.materialize(evaluation_input) + if isinstance(value, EvaluationRecordDraft): + return value + session = self.session(context) + supplied = value.reference.entities is not None + entities = value.reference.entities + try: + if entities is None: + entities = self._reference_entities(value.reference.answer, session) + if not entities: + return self.not_applicable_record("reference entity inventory is empty; recall is undefined", session) + verdicts: list[ContextEntityVerdict] = [] + batches = payload_batches( + tuple(enumerate(entities)), + lambda items: { + "question": value.query.question, + "entities": [{"entity_index": index, "text": text} for index, text in items], + "contexts": indexed(value.candidate.contexts, "context_index"), + }, + self._batching, + ) + for ordinal, (items, payload) in enumerate(batches): + key = "judge_entity_coverage" if ordinal == 0 else f"judge_entity_coverage_{ordinal}" + output = session.call( + key, + self.request( + ContextEntityRecallOutput, + payload, + prompt=self._entity_prompt("coverage", _ENTITY_COVERAGE_INSTRUCTION), + ), + ) + expected = dict(items) + if {item.entity_index for item in output.verdicts} != set(expected) or any( + expected.get(item.entity_index) != item.entity for item in output.verdicts + ): + return self.failure( + "incomplete_context_entity_verdicts", + "entity coverage must preserve every supplied entity and its global index", + FailureCategory.INVALID_PROVIDER_OUTPUT, + calls=session.calls, + ) + verdicts.extend(output.verdicts) + except JudgeCallError as exc: + return self.judge_failure(exc, session) + except BatchInputError as exc: + return self.failure( + "evaluation_payload_too_large", + str(exc), + FailureCategory.RESOURCE_EXHAUSTED, + calls=session.calls, + ) + ordered = sorted(verdicts, key=lambda item: item.entity_index) + covered = sum(item.covered for item in ordered) + artifact = subject(evaluation_input, "retrieval_reference") + source = "supplied" if supplied else "model-extracted" + return self.completed( + ratio("context_entity_recall_llm", round(covered / len(entities), 4)), + _entity_findings(ordered, artifact, supplied=supplied), + f"retrieved contexts cover {covered} of {len(entities)} entities in the fixed {source} inventory", + session, + ) + + def _reference_entities(self, answer: str, session: JudgeSession) -> tuple[str, ...]: + if not answer.strip(): + return () + batches = payload_batches((answer,), lambda items: {"reference_answer": items[0]}, self._batching) + _, payload = next(batches) + output = session.call( + "extract_reference_entities", + self.request( + ReferenceEntitiesOutput, + payload, + prompt=self._entity_prompt("inventory", _ENTITY_EXTRACTION_INSTRUCTION), + ), + ) + return output.entities + + def _entity_prompt(self, task: str, instruction: str) -> JudgeContent: + return JudgeContent( + component=ComponentRef(namespace="retrieval", name=f"entity_{task}", revision="v1"), + text=f"{self._prompt.text}\n\n{instruction}", + ) + + +def _entity_findings( + verdicts: list[ContextEntityVerdict], + artifact: ArtifactRef, + *, + supplied: bool, +) -> tuple[Finding, ...]: + """Render entity-coverage verdicts as item-level findings.""" + return tuple( + Finding( + code="context_entity_recall.covered" if verdict.covered else "context_entity_recall.missing", + severity=FindingSeverity.INFO if verdict.covered else FindingSeverity.ERROR, + message=text_preview(f"{verdict.entity}: {verdict.reason}", 4_000), + evidence=( + EvidenceLink( + artifact=artifact, + locator=( + f"entities[{verdict.entity_index}]" if supplied else f"answer.entity[{verdict.entity_index}]" + ), + description=text_preview(verdict.entity, 2_000), + ), + ), + ) + for verdict in verdicts + ) + + +class EvidenceRecallEvaluator(JudgedRetrievalEvaluator): + def __init__( + self, + materializer: RetrievalArtifactMaterializer, + *, + judge: Judge, + prompt: JudgeContent, + rubric: JudgeContent, + clock: Callable[[], datetime], + batching: BatchingOptions | None = None, + ) -> None: + super().__init__( + materializer, + invocation_key="evidence_recall_llm", + judge=judge, + prompt=prompt, + rubric=rubric, + clock=clock, + batching=batching, + ) + + def evaluate(self, evaluation_input: EvaluationInput, context: EvaluationContext) -> EvaluationRecordDraft: + value = self.materialize(evaluation_input) + if isinstance(value, EvaluationRecordDraft): + return value + session = self.session(context) + evidence = value.reference.evidence + artifact = subject(evaluation_input, "retrieval_reference") + if not evidence: Review Comment: Important: a sample without reference evidence gets `evidence_recall_llm` 1.0 with no judge call, whatever was retrieved, and `retrieval_input` defaults `reference_evidence` to `()`, so such samples inflate averages. Return `not_applicable_record` here, as context entity recall does for an empty inventory. ########## graphrag/eval/src/hugegraph_llm_eval/integrations/judges/openai_compatible.py: ########## @@ -0,0 +1,197 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""OpenAI-compatible structured-output adapter with auditable attempts.""" + +import json +import math +import time +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime +from typing import Protocol, cast + +from pydantic import Field, JsonValue, field_validator + +from hugegraph_llm_eval.contracts.base import StrictModel +from hugegraph_llm_eval.contracts.domain.judge import ProviderUsage +from hugegraph_llm_eval.contracts.errors import FailureCategory, FailureDetail +from hugegraph_llm_eval.evaluation.judges.base import ( + JudgeDescriptor, + JudgeRequest, + OutputT, +) +from hugegraph_llm_eval.integrations.judges.attempts import ( + AttemptFailureError, + AttemptLoopJudge, + AttemptOutcome, + failure, +) +from hugegraph_llm_eval.integrations.judges.openai_response import ( + OpenAIResponse, + OpenAIResponseUsage, + classify_exception, + exception_usage, + provider_outcome, +) + + +class OpenAIParseTransport(Protocol): + def parse(self, **kwargs: object) -> object: ... + + +@dataclass(frozen=True, slots=True) +class OpenAICompatiblePricing: + input_usd_per_million: float | None + output_usd_per_million: float | None + + def __post_init__(self) -> None: + values = (self.input_usd_per_million, self.output_usd_per_million) + if any(value is not None and (not math.isfinite(value) or value < 0) for value in values): + msg = "judge token pricing must be finite and non-negative" + raise ValueError(msg) + + +class OpenAICompatibleSettings(StrictModel): + temperature: float = Field(default=0, ge=0, le=2) + seed: int | None = 42 + max_tokens: int = Field(default=131_072, ge=1) + include_json_schema_instruction: bool = True + extra_body: dict[str, JsonValue] | None = None + + @field_validator("temperature") + @classmethod + def _validate_temperature(cls, value: float) -> float: + if not math.isfinite(value): + msg = "temperature must be finite" + raise ValueError(msg) + return value + + +def _configuration_payload( + settings: OpenAICompatibleSettings, + pricing: OpenAICompatiblePricing, + transport: JsonValue = None, +) -> JsonValue: + payload: dict[str, JsonValue] = { + "generation": settings.model_dump(mode="json"), + "pricing": { + "input_usd_per_million": pricing.input_usd_per_million, + "output_usd_per_million": pricing.output_usd_per_million, + }, + } + if transport is not None: + payload["transport"] = transport + return payload + + +def _structured_messages( + request: JudgeRequest[OutputT], + *, + include_schema_instruction: bool, +) -> list[dict[str, object]]: + messages = [message.model_dump(mode="json") for message in request.messages] + if not include_schema_instruction: + return messages + schema = json.dumps( + request.response_type.model_json_schema(), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + instruction = { + "role": "system", + "content": f"Return only valid JSON matching this JSON Schema: {schema}", + } + return [instruction, *messages] + + +class OpenAICompatibleJudge(AttemptLoopJudge): + """Adapts one chat-completions parse transport to the Judge port.""" + + def __init__( + self, + *, + transport: OpenAIParseTransport, + descriptor: JudgeDescriptor, + pricing: OpenAICompatiblePricing, + settings: OpenAICompatibleSettings, + clock: Callable[[], datetime], + transport_configuration: JsonValue = None, + monotonic: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep, + ) -> None: + super().__init__(descriptor=descriptor, clock=clock, monotonic=monotonic, sleep=sleep) + self._transport = transport + self._pricing = pricing + self._settings = settings + expected = JudgeDescriptor.create( + provider=descriptor.provider, + model=descriptor.model, + configuration=_configuration_payload(settings, pricing, transport_configuration), + ) + if descriptor.configuration_digest != expected.configuration_digest: + msg = "JudgeDescriptor configuration digest does not match OpenAI-compatible settings" + raise ValueError(msg) + + def _classify(self, error: Exception) -> FailureDetail: + return classify_exception(error) + + def _billed_usage(self, error: Exception) -> ProviderUsage | None: + return exception_usage(error, self._usage) + + def _attempt(self, request: JudgeRequest[OutputT], timeout: float | None) -> AttemptOutcome: + kwargs: dict[str, object] = { + "model": self.descriptor.model, + "messages": _structured_messages( + request, + include_schema_instruction=self._settings.include_json_schema_instruction, + ), + "response_format": request.response_type, + "temperature": self._settings.temperature, + "max_tokens": self._settings.max_tokens, Review Comment: Important: every request sends `max_tokens` and `temperature=0`, which OpenAI rejects with a 400 for reasoning models (o-series, `gpt-5`), classified as permanent `provider_request_rejected`, and `openai_judge` offers no way to change either. Let callers send `max_completion_tokens` instead and omit `temperature` and `seed`. ########## graphrag/eval/src/hugegraph_llm_eval/prompt.py: ########## @@ -0,0 +1,355 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# Prompt text stays on one line so the rendered instruction remains auditable. +# ruff: noqa: E501 + +"""Stable prompt selection and semantic-judge protocol adaptation.""" + +from typing import Final, cast + +from pydantic import JsonValue + +from hugegraph_llm_eval.contracts.digests import canonical_digest +from hugegraph_llm_eval.prompt_templates.answer import ( + _CORRECTNESS_CLASSIFY_PROMPT_ZH, + _NLI_STATEMENT_PROMPT_ZH, + _STATEMENT_DECOMPOSE_PROMPT_ZH, + CORRECTNESS_CLASSIFY_PROMPT, + NLI_STATEMENT_PROMPT, + STATEMENT_DECOMPOSE_PROMPT, +) +from hugegraph_llm_eval.prompt_templates.context import ( + _CLAIM_DECOMPOSE_PROMPT_ZH, + _CONTEXT_PRECISION_PROMPT_ZH, + _CONTEXT_RELEVANCE_PROMPT_ZH, + _ENTITY_EXTRACT_PROMPT_ZH, + _EVIDENCE_RECALL_PROMPT_ZH, + CLAIM_DECOMPOSE_PROMPT, + CONTEXT_PRECISION_PROMPT, + CONTEXT_RELEVANCE_PROMPT, + ENTITY_EXTRACT_PROMPT, + EVIDENCE_RECALL_PROMPT, +) +from hugegraph_llm_eval.prompt_templates.graph import ( + _ANSWER_RELEVANCE_PROMPT_ZH, + _ENTITY_SEMANTIC_MATCH_PROMPT_ZH, + _TRIPLE_SEMANTIC_MATCH_PROMPT_ZH, + ANSWER_RELEVANCE_PROMPT, + ENTITY_SEMANTIC_MATCH_PROMPT, + TRIPLE_SEMANTIC_MATCH_PROMPT, +) + +_PROMPT_REGISTRY: Final[dict[str, dict[str, str]]] = { + "STATEMENT_DECOMPOSE_PROMPT": {"en": STATEMENT_DECOMPOSE_PROMPT, "zh": _STATEMENT_DECOMPOSE_PROMPT_ZH}, + "NLI_STATEMENT_PROMPT": {"en": NLI_STATEMENT_PROMPT, "zh": _NLI_STATEMENT_PROMPT_ZH}, + "CORRECTNESS_CLASSIFY_PROMPT": {"en": CORRECTNESS_CLASSIFY_PROMPT, "zh": _CORRECTNESS_CLASSIFY_PROMPT_ZH}, + "CONTEXT_PRECISION_PROMPT": {"en": CONTEXT_PRECISION_PROMPT, "zh": _CONTEXT_PRECISION_PROMPT_ZH}, + "CONTEXT_RELEVANCE_PROMPT": {"en": CONTEXT_RELEVANCE_PROMPT, "zh": _CONTEXT_RELEVANCE_PROMPT_ZH}, + "ENTITY_EXTRACT_PROMPT": {"en": ENTITY_EXTRACT_PROMPT, "zh": _ENTITY_EXTRACT_PROMPT_ZH}, + "EVIDENCE_RECALL_PROMPT": {"en": EVIDENCE_RECALL_PROMPT, "zh": _EVIDENCE_RECALL_PROMPT_ZH}, + "CLAIM_DECOMPOSE_PROMPT": {"en": CLAIM_DECOMPOSE_PROMPT, "zh": _CLAIM_DECOMPOSE_PROMPT_ZH}, + "ENTITY_SEMANTIC_MATCH_PROMPT": { + "en": ENTITY_SEMANTIC_MATCH_PROMPT, + "zh": _ENTITY_SEMANTIC_MATCH_PROMPT_ZH, + }, + "TRIPLE_SEMANTIC_MATCH_PROMPT": { + "en": TRIPLE_SEMANTIC_MATCH_PROMPT, + "zh": _TRIPLE_SEMANTIC_MATCH_PROMPT_ZH, + }, + "ANSWER_RELEVANCE_PROMPT": {"en": ANSWER_RELEVANCE_PROMPT, "zh": _ANSWER_RELEVANCE_PROMPT_ZH}, +} + + +def get_prompt(name: str, language: str = "zh") -> str: + """Return the requested historical prompt, defaulting to Chinese.""" + variants = _PROMPT_REGISTRY.get(name) + if variants is None: + raise KeyError(f"Unknown LLM-Judge prompt: {name}") + return variants.get(language, variants["en"]) + + +def answer_judge_prompt(kind: str, language: str = "zh") -> str: + """Adapt the historical answer-evaluation prompt to the current typed protocol.""" + prompt_name = { + "decompose": "STATEMENT_DECOMPOSE_PROMPT", + "verify": "NLI_STATEMENT_PROMPT", + "correctness": "CORRECTNESS_CLASSIFY_PROMPT", + }.get(kind) + if prompt_name is None: + raise ValueError(f"Unsupported answer judge prompt kind: {kind}") + historical = get_prompt(prompt_name, language) + protocol = { + "decompose": """\ +当前运行时协议覆盖上面的历史输出格式：下一条 JSON 中的 question 和 text 是待拆分内容。 +只返回 {\"claims\": [\"原子主张 1\", \"原子主张 2\"]}，不得返回 statements 或其他键。""", + "verify": """\ +当前运行时协议覆盖上面的历史输出格式：下一条 JSON 中包含 context，以及带零基 +statement_index 的 statements。必须为每个 statement_index 恰好返回一项 verdict： +{\"verdicts\": [{\"statement_index\": 0, \"entailed\": true, \"reason\": \"上下文支持该主张。\"}]}。 +不得返回 statement 或 Yes/No verdict 字段，也不得遗漏任一索引。""", + "correctness": """\ +当前运行时协议覆盖上面的历史输出格式：下一条 JSON 中的 candidate 与 reference 都带零基 +statement_index。返回 {\"matches\": [{\"candidate_index\": 0, \"reference_index\": 1, \"reason\": \"语义等价。\"}], +\"false_positives\": [{\"statement_index\": 2, \"reason\": \"标准答案不支持。\"}], +\"false_negatives\": [{\"statement_index\": 3, \"reason\": \"候选答案遗漏。\"}]}。 +每个 candidate index 必须恰好出现在 matches 或 false_positives，每个 reference index 必须恰好出现在 +matches 或 false_negatives；不得返回 tp、fp、fn 或其他键。""", + }[kind] + return f"{historical}\n\n---\n\n{protocol}" Review Comment: Minor: the historical templates are appended unrendered, so the opt-in default prompts send literal `{question}`, `{context}` and `{statements}` slots and few-shot JSON written as `{{...}}`; `retrieval_judge_prompt` at line 152 does the same. Render the templates (unescape braces, drop the placeholder tail) before appending the protocol. ########## graphrag/eval/src/hugegraph_llm_eval/utils/text_normalize.py: ########## @@ -0,0 +1,158 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Text normalization utilities for text evaluation. + +Implements the MRQA official evaluation standard from HippoRAG 2 (eval_utils.py) +with Porter stemming and bilingual (EN/ZH) support. + +For Chinese text the normalization additionally: +- converts full-width ASCII characters (letters, digits, punctuation) to half-width +- performs simplified/traditional Chinese conversion when ``opencc`` is available +- removes Chinese punctuation and collapses whitespace +""" + +import re +import string +from typing import cast + +# Chinese punctuation set +_CHINESE_PUNCTUATION = set("，。！？、；：''（）【】《》〈〉…—～·「」『』〔〕") Review Comment: Important: the two quote characters in this set are ASCII `'`, so Chinese quotes `“ ” ‘ ’` survive normalization and become zh ROUGE-L tokens: `_chinese_rouge("“北京”是首都", "北京是首都")` scores F1 0.75 for identical content. Add `“”‘’` to the set. ########## graphrag/eval/src/hugegraph_llm_eval/integrations/judges/codex_app_server.py: ########## @@ -0,0 +1,471 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""Judge backed by ``codex app-server``, so evaluation can run on a ChatGPT/Codex subscription.""" + +import os +import queue +import shutil +import tempfile +import threading +import time +import weakref +from collections.abc import Callable, Mapping +from contextlib import suppress +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from types import TracebackType +from typing import Literal, Self + +from pydantic import JsonValue, ValidationError + +from hugegraph_llm_eval.contracts.domain.judge import ProviderUsage +from hugegraph_llm_eval.contracts.errors import FailureCategory, FailureDetail +from hugegraph_llm_eval.evaluation.judges.base import JudgeDescriptor, JudgeRequest, OutputT +from hugegraph_llm_eval.integrations.judges.attempts import ( + AttemptFailureError, + AttemptLoopJudge, + AttemptOutcome, + failure, +) +from hugegraph_llm_eval.integrations.judges.cli_common import ( + cli_version, + split_messages, + strict_json_schema, + token_count, +) +from hugegraph_llm_eval.integrations.judges.codex_connection import ( + AppServerConnection, + AppServerProcess, + ConnectionClosedError, + RpcError, + spawn_app_server, +) +from hugegraph_llm_eval.utils.clock import utc_now + +PROVIDER = "codex-app-server" +_STARTUP_TIMEOUT_SECONDS = 60.0 +_TRANSIENT_KINDS = frozenset({"serverOverloaded", "internalServerError", "flexUnavailable"}) +_TRANSIENT_PREFIXES = ("httpConnectionFailed", "responseStream") +_DEFAULT_INSTRUCTIONS = "You are an evaluation judge. Do not use tools. Answer only with the requested JSON." +_TOOL_FEATURES = ("plugins", "apps", "computer_use", "browser_use", "shell_tool", "unified_exec", "multi_agent") +_DEFAULT_CONFIG: Mapping[str, JsonValue] = { + "include_environment_context": False, + "include_permissions_instructions": False, + "include_apps_instructions": False, + "project_doc_max_bytes": 0, + "web_search": "disabled", +} +_SUBSCRIPTION_ENV_NAMES = frozenset( + { + "OPENAI_API_KEY", + "OPENAI_BASE_URL", + "OPENAI_ORG_ID", + "OPENAI_PROJECT_ID", + "OPENAI_ORGANIZATION", + "OPENAI_PROJECT", + "CODEX_API_KEY", + "CODEX_ACCESS_TOKEN", + "OPENAI_FEDERATION_RULE_ID", + "OPENAI_IDENTITY_TOKEN_FILE", + "OPENAI_WORKLOAD_IDENTITY_CONTEXT", + } +) + + +class _ConnectionHolder: + """Owns the shared connection; its ``close`` also serves as the finalizer.""" + + def __init__(self) -> None: + self.connection: AppServerConnection | None = None + + def close(self) -> None: + connection, self.connection = self.connection, None + if connection is not None: + connection.close() + + +class CodexAppServerJudge(AttemptLoopJudge): + """Runs Judge calls as ephemeral threads on one shared ``codex app-server`` process.""" + + def __init__( + self, + *, + model: str, + effort: str | None = None, + executable: str = "codex", + config: Mapping[str, JsonValue] | None = None, + auth_mode: Literal["inherit", "subscription"] = "inherit", + disable_user_mcp_servers: bool = True, + call_timeout_seconds: float = 600.0, + version: str | None = None, + clock: Callable[[], datetime] = utc_now, + spawn: Callable[..., AppServerProcess] = spawn_app_server, + monotonic: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep, + ) -> None: + if call_timeout_seconds <= 0: + msg = "call_timeout_seconds must be positive" + raise ValueError(msg) + if auth_mode not in ("inherit", "subscription"): + msg = "auth_mode must be 'inherit' or 'subscription'" + raise ValueError(msg) + provided = dict(config or {}) + if "mcp_servers" in provided: + raise ValueError("config.mcp_servers cannot be set for an evaluation judge") + features = provided.pop("features", {}) + if provided: + keys = ", ".join(sorted(provided)) + raise ValueError(f"unsupported raw Codex config keys: {keys}; use typed options") + if not isinstance(features, Mapping) or not all(isinstance(name, str) for name in features): + raise ValueError("config.features must be a string-keyed mapping") + unknown_features = sorted(set(features) - set(_TOOL_FEATURES)) + if unknown_features: + raise ValueError(f"unsupported Codex feature flags: {', '.join(unknown_features)}") + unsafe_features = [name for name, enabled in features.items() if enabled is not False] + if unsafe_features: + raise ValueError(f"Codex judge cannot enable tool capabilities: {', '.join(sorted(unsafe_features))}") + if not disable_user_mcp_servers: + msg = "disable_user_mcp_servers cannot be false for an evaluation judge" + raise ValueError(msg) + merged: dict[str, JsonValue] = { + **_DEFAULT_CONFIG, + "features": {**dict.fromkeys(_TOOL_FEATURES, False), **features}, + } + configuration: JsonValue = { + "cli_version": version if version is not None else cli_version(executable), + "effort": effort, + "auth_mode": auth_mode, + "route": "openai-chatgpt-account" if auth_mode == "subscription" else "inherited", + "config": merged, + "disable_user_mcp_servers": True, + "sandbox": "read-only", + "sampling": "provider-controlled", + } + super().__init__( + descriptor=JudgeDescriptor.create(provider=PROVIDER, model=model, configuration=configuration), + clock=clock, + monotonic=monotonic, + sleep=sleep, + ) + self._executable = executable + self._effort = effort + self._config = merged + self._auth_mode = auth_mode + self._call_timeout = call_timeout_seconds + self._spawn = spawn + self._mcp_overrides: dict[str, JsonValue] = {} + self._lock = threading.Lock() + self._holder = _ConnectionHolder() + self._workdir = tempfile.mkdtemp(prefix="hugegraph-llm-eval-codex-") + self._isolated_codex_home = Path(self._workdir) / "codex-home" + self._isolated_codex_home.mkdir(mode=0o700) + weakref.finalize(self, shutil.rmtree, self._workdir, ignore_errors=True) + weakref.finalize(self, self._holder.close) + + def close(self) -> None: + with self._lock: + self._holder.close() + + def __enter__(self) -> Self: + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + traceback: TracebackType | None, + ) -> None: + self.close() + + def _connection(self, deadline: float) -> AppServerConnection: + if not self._lock.acquire(timeout=self._left(deadline)): + raise TimeoutError("connection deadline reached") + try: + connection = self._holder.connection + if connection is None or not connection.alive: + self._holder.close() + startup_deadline = min(deadline, self._monotonic() + _STARTUP_TIMEOUT_SECONDS) + connection = self._start(startup_deadline) + self._holder.connection = connection + return connection + finally: + self._lock.release() + + def _start(self, deadline: float) -> AppServerConnection: + process = self._spawn([self._executable, "app-server"], cwd=self._workdir, env=self._environment()) + connection = AppServerConnection(process) + try: + connection.request( + "initialize", {"clientInfo": {"name": "hugegraph-llm-eval", "version": "1"}}, self._left(deadline) + ) + connection.notify("initialized") + if self._auth_mode == "subscription": + self._require_chatgpt_account(connection, self._left(deadline)) + self._mcp_overrides = self._verified_isolation_overrides(connection, self._left(deadline)) + except BaseException: + connection.close() + raise + return connection + + def _require_chatgpt_account(self, connection: AppServerConnection, timeout: float) -> None: + account = connection.request("account/read", {"refreshToken": False}, timeout).get("account") + if not isinstance(account, dict): + raise AttemptFailureError( + failure(FailureCategory.PROVIDER_UNAVAILABLE, "provider_not_authenticated", False) + ) + if account.get("type") != "chatgpt": + raise AttemptFailureError( + failure(FailureCategory.PROVIDER_UNAVAILABLE, "provider_account_not_chatgpt", False) + ) + + def _environment(self) -> dict[str, str]: + """Use a private Codex home so user hooks and config cannot enter an eval run.""" + env = dict(os.environ) + source_home = Path(env.get("CODEX_HOME", Path.home() / ".codex")).expanduser() + auth_file = source_home / "auth.json" + if auth_file.is_file(): + target_auth = self._isolated_codex_home / "auth.json" + shutil.copyfile(auth_file, target_auth) Review Comment: Important: Codex refresh tokens are single-use, so a token refresh inside this private `CODEX_HOME` leaves the user's `auth.json` holding a used token and forces a new `codex` login, and a restarted app-server recopies the stale file. Keep auth in the user's `CODEX_HOME` and isolate through config overrides, or copy a refreshed `auth.json` back before cleanup. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
