Step-Back Prompting - Abstract Before You Search

from typing import List, Dict

STEPBACK_SYSTEM_PROMPT = """You are an expert at transforming specific questions into
broader, more general questions that capture the underlying principles.

Given a specific question, generate a more abstract version that covers the
fundamental concepts or principles needed to answer the specific question.

Return only the abstract question, nothing else."""


def generate_stepback_query(query: str, llm) -> str:
    """Generate an abstract step-back version of the query."""
    response = llm.invoke(
        system=STEPBACK_SYSTEM_PROMPT,
        prompt=f"Specific question: {query}\n\nAbstract question:",
    )
    return response.strip()


def retrieve_both(
    original_query: str,
    stepback_query: str,
    retriever,
    top_k: int = 4,
) -> Dict[str, List[Dict]]:
    """Retrieve context for both the abstract and specific queries."""
    abstract_docs = retriever.search(stepback_query, top_k=top_k)
    specific_docs = retriever.search(original_query, top_k=top_k)
    return {
        "abstract": abstract_docs,
        "specific": specific_docs,
    }


def merge_contexts(
    abstract_docs: List[Dict],
    specific_docs: List[Dict],
) -> str:
    """Merge abstract principles and specific details, deduplicating by source."""
    seen_ids = set()
    merged = []

    for doc in abstract_docs:
        if doc["id"] not in seen_ids:
            merged.append(f"[General principle]\n{doc['content']}")
            seen_ids.add(doc["id"])

    for doc in specific_docs:
        if doc["id"] not in seen_ids:
            merged.append(f"[Specific context]\n{doc['content']}")
            seen_ids.add(doc["id"])

    return "\n\n".join(merged)


GENERATION_PROMPT = """Use the following context to answer the question.
The context includes both general principles and specific information.
Reason from the principles where a direct answer is not available.

Context:
{context}

Question: {question}

Answer:"""


def step_back_rag(query: str, retriever, llm) -> Dict:
    # Step 1: Generate abstract step-back query
    stepback_query = generate_stepback_query(query, llm)

    # Step 2: Retrieve from both queries
    docs = retrieve_both(query, stepback_query, retriever)

    # Step 3: Merge contexts
    context = merge_contexts(docs["abstract"], docs["specific"])

    # Step 4: Generate grounded answer
    prompt = GENERATION_PROMPT.format(context=context, question=query)
    response = llm.invoke(prompt)

    return {
        "answer": response,
        "original_query": query,
        "stepback_query": stepback_query,
        "abstract_sources": [d["source"] for d in docs["abstract"]],
        "specific_sources": [d["source"] for d in docs["specific"]],
    }

query = "What is the temperature of the core of the Sun?"

result = step_back_rag(query, retriever, llm)

print(f"Step-back query: {result['stepback_query']}")
# → "What are the physical properties and internal structure of main-sequence stars?"

print(f"Answer: {result['answer']}")
# → "The solar core reaches approximately 15 million Kelvin. This extreme temperature
#     is required to sustain the proton-proton chain reaction that powers the Sun —
#     a consequence of the balance between thermal pressure and gravitational
#     compression in stellar interiors."