Eval Grading Improvements Tutorial¶

# BUGGY VERSION (from before PR #2674)
if any(pattern in answer.lower() for pattern in incorrect_patterns):
    score = 0.0  # Wrong! Penalizes even when correct answer is present

# FIXED VERSION (from PR #2674)
def _deterministic_grade(answer: str, rubric: GradingRubric) -> float:
    answer_lower = answer.lower()

    # Check if correct keywords are present
    has_correct = any(
        keyword.lower() in answer_lower
        for keyword in rubric.correct_keywords
    )

    # Only penalize incorrect patterns if correct answer is MISSING
    if not has_correct:
        for pattern in rubric.incorrect_patterns:
            if pattern.lower() in answer_lower:
                return 0.0

    # Grade based on keyword matches
    matches = sum(
        1 for keyword in rubric.correct_keywords
        if keyword.lower() in answer_lower
    )
    return matches / len(rubric.correct_keywords)

def _entity_linked_retrieval(self, question: str) -> list[Fact]:
    """
    Retrieves all facts containing structured entity IDs found in the question.

    Structured ID patterns:
    - INC-YYYY-NNN (incidents)
    - CVE-YYYY-NNNNN (vulnerabilities)
    - PROJ-NNN (projects)
    - SRV-NNN (servers)
    """
    import re

    # Extract entity IDs from question
    entity_patterns = [
        r'INC-\d{4}-\d{3}',    # Incident IDs
        r'CVE-\d{4}-\d{4,}',   # CVE IDs
        r'PROJ-\d{3}',         # Project IDs
        r'SRV-\d{3}',          # Server IDs
    ]

    entities = []
    for pattern in entity_patterns:
        entities.extend(re.findall(pattern, question, re.IGNORECASE))

    if not entities:
        return []

    # Search for facts containing any of these entity IDs
    all_facts = []
    for entity_id in entities:
        # Get facts from memory where fact text contains the entity ID
        facts = self.memory.search(
            query=entity_id,
            filters=None,  # Search ALL contexts
            limit=10
        )
        all_facts.extend(facts)

    return self._deduplicate_facts(all_facts)

def _multi_entity_retrieval(self, question: str) -> list[Fact]:
    """
    Detects questions with multiple entities and retrieves facts for each independently.

    Useful for multi-hop reasoning questions like:
    - "How did X affect Y?"
    - "What's the relationship between A and B?"
    - "Compare X and Y"
    """
    # Detect named entities or key phrases
    entities = self._extract_entities(question)

    if len(entities) < 2:
        return []

    # Retrieve facts for EACH entity independently
    all_facts = []
    for entity in entities:
        facts = self.memory.search(
            query=entity,
            limit=5
        )
        all_facts.extend(facts)

    return self._deduplicate_facts(all_facts)

def _extract_entities(self, question: str) -> list[str]:
    """
    Extract named entities and key noun phrases from question.

    Uses simple heuristics:
    - Capitalized phrases (e.g., "Alpine Lodge")
    - Quoted terms (e.g., "Snowfall incident")
    - Common entity patterns
    """
    import re

    entities = []

    # Extract quoted terms
    entities.extend(re.findall(r'"([^"]+)"', question))

    # Extract capitalized phrases (2-4 words)
    entities.extend(
        re.findall(r'\b([A-Z][a-z]+(?: [A-Z][a-z]+){1,3})\b', question)
    )

    # Remove duplicates, keep unique entities
    return list(set(entities))

def answer_question(self, question: str) -> str:
    # 1. Detect question intent
    intent = self._detect_intent(question)

    # 2. Try entity-linked retrieval first (structured IDs)
    facts = self._entity_linked_retrieval(question)

    # 3. If insufficient, try multi-entity retrieval
    if len(facts) < 3:
        facts.extend(self._multi_entity_retrieval(question))

    # 4. Fall back to standard retrieval
    if len(facts) < 3:
        facts.extend(self._standard_retrieval(question))

    # 5. Synthesize answer from retrieved facts
    answer = self._synthesize_with_llm(question, facts)

    return answer

# DON'T: Apply incorrect patterns unconditionally
if "outdated_info" in answer:
    return 0.0

# DO: Check for correct answer first
if not has_correct_answer and "outdated_info" in answer:
    return 0.0

def _deduplicate_facts(self, facts: list[Fact]) -> list[Fact]:
    seen = set()
    unique = []
    for fact in facts:
        # Use fact ID or content hash as deduplication key
        key = fact.id or hash(fact.content)
        if key not in seen:
            seen.add(key)
            unique.append(fact)
    return unique

def answer_question(self, question: str) -> str:
    retrieval_method = None

    facts = self._entity_linked_retrieval(question)
    if facts:
        retrieval_method = "entity_linked"
    else:
        facts = self._multi_entity_retrieval(question)
        if facts:
            retrieval_method = "multi_entity"
        else:
            facts = self._standard_retrieval(question)
            retrieval_method = "standard"

    logger.info(f"Used {retrieval_method} retrieval: {len(facts)} facts")
    return self._synthesize_with_llm(question, facts)