Source Attribution and Citation in RAG

def build_attributed_prompt(query, retrieved_chunks):
    """Build a prompt that instructs the LLM to cite sources inline."""
    context_block = ""
    for i, chunk in enumerate(retrieved_chunks):
        source_id = f"[{i+1}]"
        context_block += (
            f"Source {source_id}: {chunk['title']}\n"
            f" URL: {chunk['url']}\n"
            f" Content: {chunk['text']}\n\n"
            )
        system_prompt = """You are a research assistant. Answer the user's question
        using ONLY the provided sources. Follow these citation rules strictly:
        1. Every factual claim must have an inline citation like [1], [2], etc.
        2. If multiple sources support a claim, cite all of them: [1][3].
        3. If no source supports a claim, do not make it. Say "I could not find
        information about this in the provided sources."
        4. At the end, list all cited sources with their titles and URLs.
        5. Never cite a source for a claim it does not actually support."""
        return system_prompt, f"Sources:\n{context_block}\nQuestion: {query}"
        # Example usage
        chunks = [
            {"title": "Returns Policy", "url": "/docs/returns", "text": "Items may be returned within 30 days..."},
            {"title": "Warranty Guide", "url": "/docs/warranty", "text": "Electronics carry a 2-year warranty..."},
            ]
        system, user_msg = build_attributed_prompt("What is the return window?", chunks)
        # LLM output: "Items can be returned within 30 days of purchase [1].
        # Electronics also carry a 2-year warranty [2].
        #
        # Sources:
        # [1] Returns Policy - /docs/returns
        # [2] Warranty Guide - /docs/warranty"

from pydantic import BaseModel
from openai import OpenAI
class Citation(BaseModel):
    source_id: int
    quote: str # Exact quote from the source that supports the claim
class AnswerStatement(BaseModel):
    claim: str
    citations: list[Citation]
class AttributedAnswer(BaseModel):
    statements: list[AnswerStatement]
    unsupported_aspects: list[str] # Parts of the question with no source support
    client = OpenAI()
    def generate_attributed_answer(query, chunks):
        """Generate a structured answer with per-claim citations."""
        context = "\n".join(
            f"[Source {i+1}]: {c['text']}" for i, c in enumerate(chunks)
            )
        response = client.beta.chat.completions.parse(
            model="gpt-4o",
            messages=[
            {"role": "system", "content": (
            "Answer the question using only the provided sources. "
            "For each statement, include the source ID and an exact "
            "quote from that source as evidence. List any aspects of "
            "the question that the sources do not address."
            )},
            {"role": "user", "content": f"Sources:\n{context}\n\nQuestion: {query}"}
            ],
            response_format=AttributedAnswer,
            )
        return response.choices[0].message.parsed

from transformers import pipeline
nli = pipeline("text-classification", model="cross-encoder/nli-deberta-v3-large")
def verify_citations(statements, source_texts):
    """Verify that each citation's source actually supports its claim."""
    results = []
    for stmt in statements:
        claim = stmt.claim
        for cit in stmt.citations:
            source = source_texts[cit.source_id - 1]
            # NLI: does the source entail the claim?
            result = nli(f"{source}", f"{claim}")
            label = result[0]["label"]
            score = result[0]["score"]
            results.append({
                "claim": claim,
                "source_id": cit.source_id,
                "nli_label": label,
                "nli_score": score,
                "verified": label == "ENTAILMENT" and score > 0.7
                })
            return results
            # Flag unverified citations for human review or removal

from rapidfuzz import fuzz
def verify_quote(quote, source_text, threshold=85):
    """Check if a citation quote actually appears in the source."""
    # Try exact substring match first
    if quote.lower() in source_text.lower():
        return {"match": "exact", "score": 100}
        # Fall back to fuzzy matching for minor variations
        # Slide a window of quote-length across the source
        best_score = 0
        quote_len = len(quote)
        for i in range(len(source_text) - quote_len + 1):
            window = source_text[i:i + quote_len]
            score = fuzz.ratio(quote.lower(), window.lower())
            best_score = max(best_score, score)
            if best_score >= threshold:
                return {"match": "fuzzy", "score": best_score}
                return {"match": "none", "score": best_score}