Building Trust in AI Responses: Citations, Confidence Scores, and Hallucination Prevention

def build_grounded_prompt(
    query: str,
    context_chunks: list[str]
) -> str:
    """Build a prompt that grounds the LLM in retrieved context."""
    
    context_block = "\n---\n".join(context_chunks)
    
    prompt = f"""You are a product knowledge assistant. Your role is to answer 
questions about our products accurately and honestly.
 
CRITICAL RULES:
1. ONLY use information from the provided context.
2. If the context doesn't contain enough information to answer, say so explicitly.
3. Do NOT invent, infer, or assume information not in the context.
4. If there's any uncertainty, express it clearly.
5. Always cite which source your answer comes from.
 
CONTEXT:
{context_block}
 
QUESTION: {query}
 
ANSWER (be honest about what you know and don't know):"""
    
    return prompt

def calculate_confidence_from_retrieval(
    retrieved_chunks: list[dict],  # Each has "score" (similarity) and "metadata"
    top_k: int = 3
) -> float:
    """
    Confidence based on retrieval quality.
    
    Heuristics:
    - High similarity (> 0.8) = high confidence
    - Multiple relevant chunks = high confidence
    - Low similarity (< 0.6) = low confidence
    - Chunks from primary sources (datasheets) > secondary (forums)
    """
    
    top_chunks = retrieved_chunks[:top_k]
    
    if not top_chunks:
        return 0.0
    
    # Average similarity of top chunks
    avg_similarity = sum(c["score"] for c in top_chunks) / len(top_chunks)
    
    # Penalize low similarity
    if avg_similarity < 0.5:
        return 0.3
    if avg_similarity < 0.65:
        return 0.6
    if avg_similarity < 0.8:
        return 0.8
    
    # Bonus for multiple chunks from primary sources
    primary_sources = sum(
        1 for c in top_chunks 
        if c["metadata"].get("source_type") == "datasheet"
    )
    if primary_sources >= 2:
        return 0.95
    
    return 0.85

def generate_answer_with_citations(
    query: str,
    context_chunks: list[dict],
    confidence_score: float
) -> dict:
    """Generate an answer with citations."""
    
    sources_with_context = []
    for i, chunk in enumerate(context_chunks[:5], start=1):
        sources_with_context.append(
            f"[Source {i}] {chunk['metadata'].get('title', 'Unknown')}: "
            f"{chunk['content']}"
        )
    
    context_block = "\n\n".join(sources_with_context)
    
    prompt = f"""Answer using provided sources. 
Always cite [Source N] when using information.
 
{context_block}
 
Question: {query}
 
Answer (with citations):"""
    
    answer_text = call_llm(prompt)
    
    return {
        "answer": answer_text,
        "sources": [
            {
                "title": context_chunks[i]["metadata"].get("title"),
                "url": context_chunks[i]["metadata"].get("url"),
                "type": context_chunks[i]["metadata"].get("source_type"),
            }
            for i in range(min(3, len(context_chunks)))
        ],
        "confidence": confidence_score,
    }

class SensitiveClaimsValidator:
    """Detect and validate high-stakes claims before surfacing them."""
    
    SENSITIVE_PATTERNS = {
        "certification": r"(NSF|FDA|CE|ATEX|ISO)\s*\d+",
        "safety": r"(safe|hazard|risk|danger|toxic|flammable)",
        "legal": r"(complian|regulat|legal|warrant|liability)",
        "specification": r"(maximum|minimum|rated|specified).*?(psi|bar|°|volt|amp)",
    }
    
    def validate_answer(self, answer: str, source_chunks: list[dict]) -> dict:
        """Check if sensitive claims are actually in the sources."""
        
        issues = []
        
        for claim_type, pattern in self.SENSITIVE_PATTERNS.items():
            matches = re.findall(pattern, answer, re.IGNORECASE)
            for match in matches:
                # Check if this exact claim appears in sources
                if not self._claim_in_sources(match, source_chunks):
                    issues.append({
                        "type": claim_type,
                        "claim": match,
                        "severity": "high" if claim_type in ["certification", "safety"] else "medium",
                    })
        
        return {
            "safe_to_surface": len([i for i in issues if i["severity"] == "high"]) == 0,
            "issues": issues,
        }
    
    def _claim_in_sources(self, claim: str, chunks: list[dict]) -> bool:
        """Check if claim appears verbatim in sources."""
        return any(claim in chunk["content"] for chunk in chunks)
 
# Usage
validator = SensitiveClaimsValidator()
result = validator.validate_answer(answer_text, source_chunks)
 
if not result["safe_to_surface"]:
    # Escalate to human review or flag for manual verification
    return {
        "answer": answer_text,
        "status": "requires_review",
        "issues": result["issues"],
    }

def should_escalate(
    query: str,
    confidence_score: float,
    has_sensitive_claims: bool,
    retrieval_quality: dict
) -> bool:
    """Determine if a question should escalate to human."""
    
    # Escalate on low confidence
    if confidence_score < 0.5:
        return True
    
    # Escalate on unresolved sensitive claims
    if has_sensitive_claims:
        return True
    
    # Escalate on ambiguous/unclear queries
    if retrieval_quality["top_match_score"] < 0.4:
        return True
    
    # Escalate on custom requests
    if "custom" in query.lower() or "specific" in query.lower():
        return True
    
    return False
 
# Usage
if should_escalate(query, confidence, has_claims, retrieval_quality):
    return {
        "answer": "This question requires specialist attention. Connecting you with our team...",
        "escalate_to": "sales_team",
        "context": {
            "query": query,
            "confidence": confidence,
            "reason": "high-stakes or custom request",
        }
    }

def log_interaction(
    query: str,
    answer: str,
    confidence: float,
    sources: list[str],
    user_feedback: int = None  # -1 (bad), 0 (neutral), 1 (good)
) -> None:
    """Log interaction for analysis and improvement."""
    
    db.interactions.insert_one({
        "query": query,
        "answer": answer,
        "confidence": confidence,
        "sources_used": sources,
        "user_feedback": user_feedback,
        "timestamp": datetime.now(),
        "feedback_status": "neutral" if user_feedback is None else (
            "positive" if user_feedback == 1 else "negative"
        ),
    })
    
    # Alert on negative feedback with high confidence (hallucination signal)
    if user_feedback == -1 and confidence > 0.8:
        alert_team({
            "type": "potential_hallucination",
            "query": query,
            "answer": answer,
            "confidence": confidence,
        })
 
# Analytical query: find high-confidence answers that got negative feedback
def find_hallucinations():
    return db.interactions.find({
        "confidence": {"$gt": 0.8},
        "user_feedback": -1,
    })