feat(classifier): implement Tier 3 LLM fallback adapter and boundary disambiguation test suite
This commit is contained in:
+90
-6
@@ -1,38 +1,122 @@
|
||||
"""Optional LLM fallback adapter (Tier 3).
|
||||
|
||||
Disabled by default. Provides fallback interface for boundary disambiguation
|
||||
without requiring OpenAI/Anthropic/Gemini API keys for core POC execution.
|
||||
without requiring external API keys for core POC execution.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from typing import Callable, Optional
|
||||
|
||||
from src.adapters.base import BaseNLPAdapter
|
||||
from src.models import ClassificationResult, ECPSnapshot
|
||||
from src.models import ClassificationResult, DecisionCategory, ECPSnapshot
|
||||
|
||||
|
||||
class LLMFallbackAdapter(BaseNLPAdapter):
|
||||
"""Optional adapter for LLM fallback boundary disambiguation."""
|
||||
|
||||
def __init__(self, model_name: str = "gpt-4o-mini", api_key: str | None = None) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
model_name: str = "gpt-4o-mini",
|
||||
api_key: str | None = None,
|
||||
provider_fn: Optional[Callable[[str], str]] = None,
|
||||
) -> None:
|
||||
self.model_name = model_name
|
||||
self.api_key = api_key or os.environ.get("OPENAI_API_KEY")
|
||||
self.provider_fn = provider_fn
|
||||
|
||||
def is_available(self) -> bool:
|
||||
return bool(self.api_key)
|
||||
"""Returns True if an API key or custom provider function is configured."""
|
||||
return bool(self.api_key or self.provider_fn)
|
||||
|
||||
def evaluate_similarity(self, text: str, terms: list[str]) -> float:
|
||||
return 0.0
|
||||
|
||||
def build_prompt(
|
||||
self, ecp: ECPSnapshot, content_md: str, initial_result: ClassificationResult
|
||||
) -> str:
|
||||
"""Constructs a structured disambiguation prompt for the LLM."""
|
||||
return (
|
||||
f"You are an NLP Entity Inherence Evaluator.\n"
|
||||
f"Target Entity: {ecp.target_name} (Aliases: {', '.join(ecp.aliases)})\n"
|
||||
f"Domain: {ecp.domain}\n"
|
||||
f"Initial Tier-1 Decision: {initial_result.decision.value} (Confidence: {initial_result.confidence})\n\n"
|
||||
f"Document Content:\n```markdown\n{content_md[:2000]}\n```\n\n"
|
||||
f"Evaluate if the document is substantively inherent to the target entity.\n"
|
||||
f'Respond with JSON: {{"decision": "DIRECT_INHERENT"|"CONTEXTUAL_INHERENT"|"TANGENTIAL"|"NOT_RELATED", '
|
||||
f'"confidence": 0.0-1.0, "rationale": "explanation"}}'
|
||||
)
|
||||
|
||||
def disambiguate(
|
||||
self,
|
||||
ecp: ECPSnapshot,
|
||||
content_md: str,
|
||||
initial_result: ClassificationResult,
|
||||
) -> ClassificationResult | None:
|
||||
# If API key is not configured or case is already clear, skip
|
||||
"""
|
||||
Executes LLM fallback for ambiguous boundary cases.
|
||||
Returns a refined ClassificationResult or None if skipped.
|
||||
"""
|
||||
if not self.is_available():
|
||||
return None
|
||||
# In POC, Tier 1 is definitive; LLM fallback stub is available for extension
|
||||
|
||||
prompt = self.build_prompt(ecp, content_md, initial_result)
|
||||
|
||||
# If custom provider function is provided (e.g. for testing or custom runtime)
|
||||
if self.provider_fn is not None:
|
||||
raw_response = self.provider_fn(prompt)
|
||||
return self._parse_llm_response(raw_response, initial_result)
|
||||
|
||||
# Stub default for POC when only API key string is present without active SDK
|
||||
return None
|
||||
|
||||
def _parse_llm_response(
|
||||
self,
|
||||
raw_response: str,
|
||||
initial_result: ClassificationResult,
|
||||
) -> ClassificationResult | None:
|
||||
"""Parses and validates structured JSON response from LLM."""
|
||||
try:
|
||||
# Extract JSON block if surrounded by markdown code fences
|
||||
clean_str = raw_response.strip()
|
||||
if clean_str.startswith("```json"):
|
||||
clean_str = clean_str[7:]
|
||||
if clean_str.startswith("```"):
|
||||
clean_str = clean_str[3:]
|
||||
if clean_str.endswith("```"):
|
||||
clean_str = clean_str[:-3]
|
||||
clean_str = clean_str.strip()
|
||||
|
||||
parsed = json.loads(clean_str)
|
||||
if not isinstance(parsed, dict):
|
||||
return None
|
||||
|
||||
raw_decision = parsed.get("decision")
|
||||
if not raw_decision:
|
||||
return None
|
||||
|
||||
decision = DecisionCategory(raw_decision)
|
||||
confidence = float(parsed.get("confidence", 0.90))
|
||||
confidence = max(0.0, min(1.0, confidence))
|
||||
rationale = str(parsed.get("rationale", "LLM boundary disambiguation."))
|
||||
is_inherent = decision in (
|
||||
DecisionCategory.DIRECT_INHERENT,
|
||||
DecisionCategory.CONTEXTUAL_INHERENT,
|
||||
)
|
||||
|
||||
return ClassificationResult(
|
||||
decision=decision,
|
||||
is_inherent=is_inherent,
|
||||
confidence=round(confidence, 4),
|
||||
detected_language=initial_result.detected_language,
|
||||
matched_anchors=initial_result.matched_anchors,
|
||||
negative_matches=initial_result.negative_matches,
|
||||
graph_matches=initial_result.graph_matches,
|
||||
evidence=initial_result.evidence,
|
||||
rationale=f"[Tier 3 LLM] {rationale}",
|
||||
warnings=initial_result.warnings + ["[Tier 3 LLM Override applied]"],
|
||||
)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
+30
-7
@@ -3,7 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
from typing import Any, Optional
|
||||
|
||||
from src.language import detect_language, normalize_text
|
||||
from src.models import (
|
||||
@@ -39,20 +39,25 @@ def count_phrase_occurrences(phrase: str, normalized_text: str) -> int:
|
||||
|
||||
|
||||
class InherenceClassifier:
|
||||
"""Tier 1 Deterministic NLP Entity Inherence Classifier."""
|
||||
"""Tier 1 Deterministic NLP Entity Inherence Classifier with optional Tier 2 / Tier 3 adapters."""
|
||||
|
||||
def __init__(self, enable_embeddings: bool = False, enable_llm: bool = False) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
enable_embeddings: bool = False,
|
||||
enable_llm: bool = False,
|
||||
llm_adapter: Optional[Any] = None,
|
||||
) -> None:
|
||||
self.enable_embeddings = enable_embeddings
|
||||
self.enable_llm = enable_llm
|
||||
self.enable_llm = enable_llm or (llm_adapter is not None)
|
||||
self._embeddings_adapter = None
|
||||
self._llm_adapter = None
|
||||
self._llm_adapter = llm_adapter
|
||||
|
||||
if enable_embeddings:
|
||||
from src.adapters.embeddings import LocalEmbeddingsAdapter
|
||||
|
||||
self._embeddings_adapter = LocalEmbeddingsAdapter()
|
||||
|
||||
if enable_llm:
|
||||
if self.enable_llm and self._llm_adapter is None:
|
||||
from src.adapters.llm import LLMFallbackAdapter
|
||||
|
||||
self._llm_adapter = LLMFallbackAdapter()
|
||||
@@ -224,7 +229,7 @@ class InherenceClassifier:
|
||||
if not evidence and has_negative_match:
|
||||
evidence = extract_evidence_snippets(content_md, matched_negative_anchors)
|
||||
|
||||
return ClassificationResult(
|
||||
tier1_result = ClassificationResult(
|
||||
decision=decision,
|
||||
is_inherent=is_inherent,
|
||||
confidence=round(confidence, 4),
|
||||
@@ -236,3 +241,21 @@ class InherenceClassifier:
|
||||
rationale=rationale,
|
||||
warnings=warnings,
|
||||
)
|
||||
|
||||
# Tier 3 (LLM Fallback): Disambiguation for ambiguous boundary cases when enabled
|
||||
if self.enable_llm and self._llm_adapter and self._llm_adapter.is_available():
|
||||
# Invoke LLM only for low-confidence or boundary/tangential decisions
|
||||
if (
|
||||
tier1_result.confidence < 0.60
|
||||
or tier1_result.decision == DecisionCategory.TANGENTIAL
|
||||
):
|
||||
try:
|
||||
llm_result = self._llm_adapter.disambiguate(ecp, content_md, tier1_result)
|
||||
if llm_result is not None:
|
||||
return llm_result
|
||||
except Exception as e:
|
||||
tier1_result.warnings.append(
|
||||
f"LLM fallback failed, retained Tier 1 decision: {e}"
|
||||
)
|
||||
|
||||
return tier1_result
|
||||
|
||||
Reference in New Issue
Block a user