Attributed Question Answering (AQA)

class AqaInput:
    prompt: str
    source_passages: List[str]

class AqaOutput:
    answer: str
    attributed_passages: List[str]
    answerable_probability: float

class GenAIAqa:
    def __init__(
        self,
        *,
        answer_style: int = 1,
        safety_settings: List[SafetySetting] = [],
        temperature: Optional[float] = None
    )

def invoke(
    self,
    input: AqaInput,
    config: Optional[RunnableConfig] = None,
    **kwargs: Any
) -> AqaOutput

from langchain_google_genai import GenAIAqa, AqaInput

# Initialize AQA service
aqa = GenAIAqa()

# Prepare source passages
passages = [
    "Machine learning is a subset of artificial intelligence that enables computers to learn and improve from experience without being explicitly programmed.",
    "Deep learning is a specialized form of machine learning that uses neural networks with multiple layers to model and understand complex patterns.",
    "Natural language processing (NLP) is a branch of AI that helps computers understand, interpret, and manipulate human language."
]

# Create input with question and passages
input_data = AqaInput(
    prompt="What is machine learning and how does it relate to AI?",
    source_passages=passages
)

# Generate attributed answer
result = aqa.invoke(input_data)

print(f"Answer: {result.answer}")
print(f"Confidence: {result.answerable_probability:.2f}")
print("Sources used:")
for i, passage in enumerate(result.attributed_passages, 1):
    print(f"{i}. {passage}")

# Simulate document content
document_sections = [
    "Python was created by Guido van Rossum and first released in 1991. It emphasizes code readability with its notable use of significant whitespace.",
    "Python supports multiple programming paradigms, including procedural, object-oriented, and functional programming.",
    "Python's design philosophy emphasizes code readability and a syntax that allows programmers to express concepts in fewer lines of code.",
    "Python has a large standard library, which is often cited as one of its greatest strengths, providing tools for many tasks."
]

# Ask specific questions about the document
questions = [
    "Who created Python and when?",
    "What programming paradigms does Python support?",
    "What are Python's main strengths?"
]

aqa = GenAIAqa()

for question in questions:
    input_data = AqaInput(
        prompt=question,
        source_passages=document_sections
    )
    
    result = aqa.invoke(input_data)
    
    print(f"\nQ: {question}")
    print(f"A: {result.answer}")
    print(f"Confidence: {result.answerable_probability:.2f}")

# Research passages about climate change
research_passages = [
    "Climate change refers to long-term shifts in global temperatures and weather patterns, primarily caused by human activities since the 1800s.",
    "The greenhouse effect occurs when certain gases in Earth's atmosphere trap heat from the sun, warming the planet.",
    "Carbon dioxide levels have increased by over 40% since pre-industrial times, primarily due to fossil fuel burning.",
    "Renewable energy sources like solar, wind, and hydroelectric power produce electricity without releasing greenhouse gases.",
    "Climate adaptation strategies include building sea walls, developing drought-resistant crops, and creating early warning systems."
]

# Research questions
research_questions = [
    "What causes climate change?",
    "How do renewable energy sources help with climate change?",
    "What are some climate adaptation strategies?"
]

aqa = GenAIAqa()

research_results = {}
for question in research_questions:
    input_data = AqaInput(
        prompt=question,
        source_passages=research_passages
    )
    
    result = aqa.invoke(input_data)
    research_results[question] = result

# Generate research report
print("=== Climate Change Research Report ===\n")
for question, result in research_results.items():
    print(f"Question: {question}")
    print(f"Answer: {result.answer}")
    print(f"Source Attribution: {len(result.attributed_passages)} passages used")
    print(f"Confidence: {result.answerable_probability:.1%}")
    print("-" * 50)

from langchain_google_genai import GoogleVectorStore

# Assume we have a populated vector store
vector_store = GoogleVectorStore(corpus_id="knowledge-base")

def aqa_with_retrieval(question: str, num_passages: int = 5):
    """Combine vector search with AQA for grounded answers."""
    
    # Retrieve relevant passages
    docs = vector_store.similarity_search(question, k=num_passages)
    passages = [doc.page_content for doc in docs]
    
    # Generate attributed answer
    aqa = GenAIAqa()
    input_data = AqaInput(
        prompt=question,
        source_passages=passages
    )
    
    result = aqa.invoke(input_data)
    
    return result, docs  # Return both answer and source documents

# Use the combined approach
question = "How do neural networks work?"
aqa_result, source_docs = aqa_with_retrieval(question)

print(f"Question: {question}")
print(f"Answer: {aqa_result.answer}")
print(f"Attribution confidence: {aqa_result.answerable_probability:.2f}")

print("\nSource documents:")
for i, doc in enumerate(source_docs, 1):
    print(f"{i}. {doc.page_content[:100]}...")

import google.ai.generativelanguage as genai

# Abstractive answering (default)
abstractive_aqa = GenAIAqa(
    answer_style=genai.GenerateAnswerRequest.AnswerStyle.ABSTRACTIVE
)

# Extractive answering (if available)
extractive_aqa = GenAIAqa(
    answer_style=genai.GenerateAnswerRequest.AnswerStyle.EXTRACTIVE
)

passages = [
    "Photosynthesis is the process by which plants convert sunlight into energy.",
    "During photosynthesis, plants absorb carbon dioxide from the air and water from the soil.",
    "The chlorophyll in plant leaves captures solar energy to power the photosynthesis reaction."
]

question = "How do plants convert sunlight into energy?"

# Compare answer styles
for name, aqa_instance in [("Abstractive", abstractive_aqa), ("Extractive", extractive_aqa)]:
    try:
        input_data = AqaInput(prompt=question, source_passages=passages)
        result = aqa_instance.invoke(input_data)
        
        print(f"\n{name} Answer:")
        print(f"Response: {result.answer}")
        print(f"Attribution: {len(result.attributed_passages)} passages")
    except Exception as e:
        print(f"{name} style not available: {e}")

# Conservative answers (lower temperature)
conservative_aqa = GenAIAqa(temperature=0.1)

# Creative answers (higher temperature)  
creative_aqa = GenAIAqa(temperature=0.9)

passages = [
    "Artificial intelligence can be applied to healthcare for diagnostic assistance.",
    "AI helps doctors analyze medical images like X-rays and MRIs more accurately.",
    "Machine learning algorithms can predict patient outcomes based on medical data."
]

question = "How is AI used in healthcare?"

print("Conservative Answer:")
result1 = conservative_aqa.invoke(AqaInput(prompt=question, source_passages=passages))
print(result1.answer)

print("\nCreative Answer:")
result2 = creative_aqa.invoke(AqaInput(prompt=question, source_passages=passages))
print(result2.answer)

from langchain_core.runnables import RunnableLambda

def prepare_aqa_input(data):
    """Helper function to prepare AQA input from chain data."""
    return AqaInput(
        prompt=data["question"],
        source_passages=data["passages"]
    )

def extract_answer(aqa_output):
    """Extract just the answer from AQA output."""
    return aqa_output.answer

# Create a chain that processes questions with AQA
aqa = GenAIAqa()

aqa_chain = (
    RunnableLambda(prepare_aqa_input)
    | aqa
    | RunnableLambda(extract_answer)
)

# Use the chain
chain_input = {
    "question": "What is quantum computing?",
    "passages": [
        "Quantum computing uses quantum mechanical phenomena like superposition and entanglement.",
        "Unlike classical bits, quantum bits (qubits) can exist in multiple states simultaneously.",
        "Quantum computers could solve certain problems exponentially faster than classical computers."
    ]
}

answer = aqa_chain.invoke(chain_input)
print(f"Chain Answer: {answer}")

def assess_aqa_quality(question: str, passages: List[str], expected_keywords: List[str]):
    """Assess the quality of AQA responses."""
    
    aqa = GenAIAqa()
    input_data = AqaInput(prompt=question, source_passages=passages)
    result = aqa.invoke(input_data)
    
    # Check confidence
    confidence_score = result.answerable_probability
    
    # Check if answer contains expected keywords
    answer_lower = result.answer.lower()
    keyword_matches = sum(1 for keyword in expected_keywords 
                         if keyword.lower() in answer_lower)
    keyword_coverage = keyword_matches / len(expected_keywords)
    
    # Check attribution quality
    attribution_ratio = len(result.attributed_passages) / len(passages)
    
    return {
        "answer": result.answer,
        "confidence": confidence_score,
        "keyword_coverage": keyword_coverage,
        "attribution_ratio": attribution_ratio,
        "quality_score": (confidence_score + keyword_coverage) / 2
    }

# Test quality assessment
passages = [
    "Machine learning algorithms learn patterns from data to make predictions.",
    "Supervised learning uses labeled examples to train models.",
    "Unsupervised learning finds hidden patterns in unlabeled data."
]

assessment = assess_aqa_quality(
    question="What is machine learning?",
    passages=passages,
    expected_keywords=["algorithms", "data", "patterns", "predictions"]
)

print(f"Answer: {assessment['answer']}")
print(f"Confidence: {assessment['confidence']:.2f}")
print(f"Keyword Coverage: {assessment['keyword_coverage']:.2f}")
print(f"Overall Quality: {assessment['quality_score']:.2f}")