Retrieval-Augmented
Generation Mastery

from openai import OpenAI
import chromadb

client = OpenAI()
chroma = chromadb.Client()
collection = chroma.get_or_create_collection("docs")

def embed(text: str) -> list[float]:
    res = client.embeddings.create(model="text-embedding-3-small", input=text)
    return res.data[0].embedding

def index_documents(docs: list[str]):
    """Embed and store documents in the vector store."""
    embeddings = [embed(doc) for doc in docs]
    collection.add(
        documents=docs,
        embeddings=embeddings,
        ids=[f"doc_{i}" for i in range(len(docs))]
    )

def rag_query(question: str, k: int = 3) -> str:
    # 1️⃣ Embed the user query
    q_emb = embed(question)

    # 2️⃣ Retrieve top-k similar chunks
    results = collection.query(query_embeddings=[q_emb], n_results=k)
    context = "\n\n".join(results["documents"][0])

    # 3️⃣ Augment prompt with retrieved context
    prompt = f"""Use ONLY the context below to answer the question.

Context:
{context}

Question: {question}"""

    # 4️⃣ Generate answer
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# Usage
index_documents(["RAG grounds LLMs in real documents...", "Retrieval happens via vector search..."])
answer = rag_query("What is RAG?")
print(answer)

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# Semantic chunker splits where embedding similarity drops sharply
splitter = SemanticChunker(
    embeddings=OpenAIEmbeddings(model="text-embedding-3-small"),
    breakpoint_threshold_type="percentile",   # or "standard_deviation"
    breakpoint_threshold_amount=85,            # split at 85th percentile jumps
)

docs = splitter.create_documents([long_text])
print(f"Created {len(docs)} semantically coherent chunks")

# ── Parent-Child pattern ──
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain.text_splitter import RecursiveCharacterTextSplitter

parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
child_splitter  = RecursiveCharacterTextSplitter(chunk_size=400)

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=InMemoryStore(),
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)
# Search returns small child chunks → fetches big parent for LLM context

# MRL: truncate embedding dimensions without retraining
# text-embedding-3-* supports this natively

from openai import OpenAI

client = OpenAI()

def embed_mrl(text: str, dims: int = 256) -> list[float]:
    """
    Smaller dims = cheaper storage + faster search, small accuracy drop.
    dims=256  → 6× smaller, ~2% accuracy loss
    dims=1536 → balanced (default small model)
    dims=3072 → maximum quality (large model)
    """
    res = client.embeddings.create(
        model="text-embedding-3-large",
        input=text,
        dimensions=dims   # ← MRL truncation
    )
    return res.data[0].embedding

# Hybrid retrieval: combine dense + sparse (BM25)
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

bm25 = BM25Retriever.from_documents(docs)
bm25.k = 4

dense = vectorstore.as_retriever(search_kwargs={"k": 4})

# 60% dense, 40% BM25 — best of both worlds
hybrid = EnsembleRetriever(
    retrievers=[bm25, dense],
    weights=[0.4, 0.6]
)

from qdrant_client import QdrantClient, models

client = QdrantClient(url="http://localhost:6333")

# Create collection with HNSW index
client.create_collection(
    collection_name="docs",
    vectors_config=models.VectorParams(size=1536, distance=models.Distance.COSINE),
    hnsw_config=models.HnswConfigDiff(m=16, ef_construct=100)  # tune for recall/speed
)

# Upsert with metadata payload
client.upsert(
    collection_name="docs",
    points=[
        models.PointStruct(
            id=1,
            vector=embed("Climate change impacts 2024"),
            payload={"source": "ipcc.pdf", "year": 2024, "category": "climate"}
        )
    ]
)

# Filtered semantic search — only docs from 2024+, climate category
results = client.search(
    collection_name="docs",
    query_vector=embed("global warming effects"),
    query_filter=models.Filter(
        must=[
            models.FieldCondition(key="year", range=models.Range(gte=2024)),
            models.FieldCondition(key="category", match=models.MatchValue(value="climate"))
        ]
    ),
    limit=5
)

from llama_index.core import KnowledgeGraphIndex, SimpleDirectoryReader
from llama_index.core.graph_stores import SimpleGraphStore
from llama_index.llms.openai import OpenAI as LlamaOpenAI

# Load documents
documents = SimpleDirectoryReader("./data").load_data()

# Build KG Index — LLM extracts SPO triples automatically
kg_index = KnowledgeGraphIndex.from_documents(
    documents,
    llm=LlamaOpenAI(model="gpt-4o"),
    max_triplets_per_chunk=10,
    include_embeddings=True,   # hybrid: graph + vector
    graph_store=SimpleGraphStore(),
)

# Query with graph traversal
query_engine = kg_index.as_query_engine(
    include_text=True,
    response_mode="tree_summarize",
    embedding_mode="hybrid",   # keyword + vector on graph
    similarity_top_k=5,
)

# Multi-hop: "What did the CEO of the company that built Claude found before Anthropic?"
response = query_engine.query("What companies did Dario Amodei found?")
print(response)

# Visualize the graph
kg_index.get_networkx_graph()  # → export to Gephi / pyvis

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool

@tool
def search_docs(query: str) -> str:
    """Search internal knowledge base for relevant documents."""
    results = vectorstore.similarity_search(query, k=3)
    return "\n\n".join(r.page_content for r in results)

@tool
def web_search(query: str) -> str:
    """Search the web for current information not in the knowledge base."""
    results = tavily.invoke(query)
    return str(results[:2])

@tool
def execute_python(code: str) -> str:
    """Execute Python code for calculations and data analysis."""
    import subprocess
    result = subprocess.run(["python3", "-c", code], capture_output=True, text=True, timeout=10)
    return result.stdout or result.stderr

# Build ReAct agent — it autonomously decides which tools to call and when
llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = create_react_agent(
    model=llm,
    tools=[search_docs, web_search, execute_python],
    prompt="You are a research assistant. Use tools to answer questions accurately."
)

# The agent will:
# 1. Think: "I need to search docs first"
# 2. Act: call search_docs
# 3. Observe: get results
# 4. Think: "I need more recent data"
# 5. Act: call web_search
# 6. Observe: get results
# 7. Generate: synthesize final answer

result = agent.invoke({"messages": [("user", "What is the latest RAG benchmark score for GPT-4o?")]})

import base64
from pathlib import Path

def describe_image(image_path: str) -> str:
    """Use GPT-4o Vision to generate a rich text description of an image."""
    img_data = base64.b64encode(Path(image_path).read_bytes()).decode()
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": "Describe this image in detail, including all text, data, charts, and visual elements."},
                {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_data}"}}
            ]
        }]
    )
    return response.choices[0].message.content

# Index: store both description (for retrieval) and image path (for generation)
def index_images(image_paths: list[str]):
    for path in image_paths:
        description = describe_image(path)
        vectorstore.add_texts(
            texts=[description],
            metadatas=[{"image_path": path, "type": "image"}]
        )

# Retrieve: get description, return original image to the LLM
def multimodal_rag(query: str) -> str:
    results = vectorstore.similarity_search(query, k=2, filter={"type": "image"})
    images_b64 = []
    for doc in results:
        img_data = base64.b64encode(Path(doc.metadata["image_path"]).read_bytes()).decode()
        images_b64.append(img_data)

    messages = [{"role": "user", "content": [
        {"type": "text", "text": f"Answer using these images: {query}"},
        *[{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img}"}} for img in images_b64]
    ]}]
    return client.chat.completions.create(model="gpt-4o", messages=messages).choices[0].message.content

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.node_parser import SemanticSplitterNodeParser
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.postprocessor import SentenceTransformerRerank
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.qdrant import QdrantVectorStore

# Ingestion pipeline: transform docs → nodes → embeddings → store
pipeline = IngestionPipeline(
    transformations=[
        SemanticSplitterNodeParser(
            embed_model=OpenAIEmbedding(model="text-embedding-3-small"),
            breakpoint_percentile_threshold=95
        ),
        OpenAIEmbedding(model="text-embedding-3-small"),  # embed nodes
    ],
    vector_store=QdrantVectorStore(client=qdrant_client, collection_name="docs")
)

nodes = pipeline.run(documents=SimpleDirectoryReader("./docs").load_data())

# Query with reranking (cross-encoder re-scores top-20 → return top-5)
index = VectorStoreIndex(nodes, storage_context=storage_ctx)
reranker = SentenceTransformerRerank(model="cross-encoder/ms-marco-MiniLM-L-6-v2", top_n=5)

query_engine = index.as_query_engine(
    similarity_top_k=20,          # retrieve 20
    node_postprocessors=[reranker], # rerank → 5
    response_mode="compact"
)

response = query_engine.query("Explain RAG-Fusion")
print(response.source_nodes[0].score)  # rerank score

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset

# Build evaluation dataset
eval_data = Dataset.from_dict({
    "question":  ["What is RAG?", "Who invented the Transformer?"],
    "answer":    ["RAG augments LLMs with external retrieval...", "The Transformer was introduced by Vaswani et al..."],
    "contexts":  [
        ["RAG stands for Retrieval-Augmented Generation..."],
        ["Attention Is All You Need, Vaswani et al., 2017..."]
    ],
    "ground_truth": ["RAG uses retrieval to augment generation.", "Vaswani et al. invented the Transformer in 2017."]
})

results = evaluate(
    eval_data,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)

print(results.to_pandas()[["faithfulness","answer_relevancy","context_precision","context_recall"]])
# faithfulness  answer_relevancy  context_precision  context_recall
#         0.92              0.87               0.91            0.78

from deepeval import assert_test
from deepeval.metrics import FaithfulnessMetric, HallucinationMetric, AnswerRelevancyMetric
from deepeval.test_case import LLMTestCase

# DeepEval integrates with pytest — RAG quality as CI/CD gates
def test_rag_faithfulness():
    test_case = LLMTestCase(
        input="What is CRAG?",
        actual_output="CRAG uses a relevance grader to decide when to use web search as fallback.",
        retrieval_context=["CRAG corrects retrieval by scoring doc relevance..."],
    )

    faithfulness = FaithfulnessMetric(threshold=0.8, model="gpt-4o", include_reason=True)
    hallucination = HallucinationMetric(threshold=0.2)

    assert_test(test_case, [faithfulness, hallucination])
    # Fails CI if faithfulness < 0.8 or hallucination > 0.2

import redis, hashlib, asyncio
from openai import AsyncOpenAI

aclient = AsyncOpenAI()
cache = redis.Redis(host="localhost", port=6379)

def cache_key(text: str) -> str:
    return f"emb:{hashlib.sha256(text.encode()).hexdigest()}"

async def embed_cached(text: str) -> list[float]:
    """Embedding with Redis cache — avoids re-embedding identical queries."""
    key = cache_key(text)
    if cached := cache.get(key):
        return eval(cached)  # deserialize
    emb = (await aclient.embeddings.create(model="text-embedding-3-small", input=text)).data[0].embedding
    cache.setex(key, 3600, str(emb))  # 1hr TTL
    return emb

async def rag_stream(question: str, vectorstore):
    """Full async RAG with streaming generator output."""
    q_emb = await embed_cached(question)
    docs = await asyncio.to_thread(vectorstore.similarity_search_by_vector, q_emb, k=3)
    context = "\n\n".join(d.page_content for d in docs)

    prompt = f"Context:\n{context}\n\nQuestion: {question}\nAnswer:"
    stream = await aclient.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True   # ← stream tokens as they arrive
    )
    async for chunk in stream:
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content  # stream to client

import anthropic

client = anthropic.Anthropic()

def claude_rag(question: str, context_docs: list[str]) -> str:
    """
    Claude 3.7 Sonnet with extended thinking — ideal for complex
    multi-step RAG where reasoning quality matters most.
    """
    context = "\n\n---\n\n".join(context_docs)

    response = client.messages.create(
        model="claude-3-7-sonnet-20250219",
        max_tokens=8000,
        thinking={"type": "enabled", "budget_tokens": 5000},  # extended thinking
        messages=[{
            "role": "user",
            "content": f"""You are a research assistant. Use ONLY the provided context.


{context}


{question}

Answer with citations. If the context is insufficient, say so."""
        }]
    )

    # Extract thinking + answer separately
    thinking_text = next((b.thinking for b in response.content if b.type == "thinking"), "")
    answer = next(b.text for b in response.content if b.type == "text")
    return answer

# Also works with prompt caching — reduces costs up to 90% on repeated context
response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": [{
            "type": "text",
            "text": large_context,
            "cache_control": {"type": "ephemeral"}  # cache the context prefix
        }, {
            "type": "text",
            "text": question
        }]
    }]
)

import boto3
from botocore.exceptions import ClientError

s3 = boto3.client("s3", region_name="us-east-1")

# ── Upload a file ──
s3.upload_file(
    Filename="report.pdf",
    Bucket="my-bucket",
    Key="reports/2025/report.pdf",
    ExtraArgs={
        "ContentType": "application/pdf",
        "ServerSideEncryption": "AES256",   # always encrypt at rest
        "StorageClass": "STANDARD_IA",       # cheaper for infrequent reads
    }
)

# ── Generate pre-signed URL (expires in 1 hour) ──
url = s3.generate_presigned_url(
    "get_object",
    Params={"Bucket": "my-bucket", "Key": "reports/2025/report.pdf"},
    ExpiresIn=3600
)
print(url)  # share with client — no AWS credentials needed

# ── Multipart upload for large files ──
from boto3.s3.transfer import TransferConfig
config = TransferConfig(multipart_threshold=25 * 1024 * 1024)  # 25MB threshold
s3.upload_file("bigfile.zip", "my-bucket", "uploads/bigfile.zip", Config=config)

# ── Set lifecycle rule — move to Glacier after 90 days, delete after 365 ──
s3.put_bucket_lifecycle_configuration(
    Bucket="my-bucket",
    LifecycleConfiguration={"Rules": [{
        "ID": "archive-old-reports",
        "Status": "Enabled",
        "Filter": {"Prefix": "reports/"},
        "Transitions": [{"Days": 90, "StorageClass": "GLACIER"}],
        "Expiration": {"Days": 365}
    }]}
)

# ── Trigger Lambda on new object ──
# (done in S3 console → Event notifications → Lambda function)
# Lambda handler receives: event["Records"][0]["s3"]["bucket"]["name"] + key

# Launch an EC2 instance (al2023, t3.micro, us-east-1)
aws ec2 run-instances \
  --image-id ami-0c02fb55956c7d316 \
  --instance-type t3.micro \
  --key-name my-keypair \
  --security-group-ids sg-xxxxxxxxxx \
  --subnet-id subnet-xxxxxxxxxx \
  --iam-instance-profile Name=MyAppRole \
  --user-data '#!/bin/bash
    yum update -y
    yum install -y docker
    systemctl start docker
    systemctl enable docker
    docker pull my-app:latest
    docker run -d -p 80:8000 my-app:latest' \
  --tag-specifications 'ResourceType=instance,Tags=[{Key=Name,Value=my-app}]' \
  --count 1

# Connect via SSM Session Manager (no SSH key needed — more secure)
aws ssm start-session --target i-xxxxxxxxxxxxxxxxx

# Create a snapshot of EBS volume
aws ec2 create-snapshot \
  --volume-id vol-xxxxxxxxxx \
  --description "Before deployment snapshot $(date +%Y-%m-%d)"

# Allocate and associate Elastic IP
EIP=$(aws ec2 allocate-address --domain vpc --query AllocationId --output text)
aws ec2 associate-address --instance-id i-xxxxxxxxxx --allocation-id $EIP

import json, boto3, os

# ── Pattern 1: API Gateway trigger (REST API) ──
def handler_api(event, context):
    body = json.loads(event.get("body", "{}"))
    name = body.get("name", "World")
    return {
        "statusCode": 200,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"message": f"Hello, {name}!"})
    }

# ── Pattern 2: S3 trigger (process uploaded file) ──
def handler_s3(event, context):
    s3 = boto3.client("s3")
    for record in event["Records"]:
        bucket = record["s3"]["bucket"]["name"]
        key    = record["s3"]["object"]["key"]
        obj    = s3.get_object(Bucket=bucket, Key=key)
        content = obj["Body"].read().decode("utf-8")
        print(f"Processing {key}: {len(content)} bytes")
        # ... process and write result back to S3

# ── Pattern 3: SQS trigger (process messages) ──
def handler_sqs(event, context):
    for record in event["Records"]:
        message = json.loads(record["body"])
        process_job(message)
    # SQS automatically deletes messages on success

# ── Powertools for best practices ──
from aws_lambda_powertools import Logger, Tracer, Metrics
from aws_lambda_powertools.metrics import MetricUnit

logger  = Logger(service="my-service")
tracer  = Tracer()
metrics = Metrics(namespace="MyApp")

@logger.inject_lambda_context
@tracer.capture_lambda_handler
@metrics.log_metrics
def handler(event, context):
    metrics.add_metric(name="RequestCount", unit=MetricUnit.Count, value=1)
    logger.info("Processing request", extra={"event": event})
    # structured JSON logs + X-Ray traces + CloudWatch metrics — all wired up

import boto3, psycopg2

# ── RDS PostgreSQL via psycopg2 ──
# Get connection string from Secrets Manager (never hardcode)
def get_db_connection():
    sm = boto3.client("secretsmanager")
    secret = sm.get_secret_value(SecretId="prod/postgres/main")
    creds = json.loads(secret["SecretString"])
    return psycopg2.connect(
        host=creds["host"], port=5432,
        database=creds["dbname"],
        user=creds["username"], password=creds["password"],
        sslmode="require"         # always require SSL on RDS
    )

# ── DynamoDB single-table design ──
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("app-table")

# Write an item
table.put_item(Item={
    "PK": "USER#usr_123",          # partition key
    "SK": "PROFILE#usr_123",       # sort key
    "name": "Alice",
    "email": "[email protected]",
    "ttl": int(time.time()) + 86400 * 30,  # auto-expire in 30 days
})

# Query all items for a user (one partition key, many sort keys)
response = table.query(
    KeyConditionExpression=Key("PK").eq("USER#usr_123") & Key("SK").begins_with("ORDER#"),
    ScanIndexForward=False,   # newest first
    Limit=20
)

# Conditional write (optimistic locking)
try:
    table.update_item(
        Key={"PK": "PRODUCT#p1", "SK": "PRODUCT#p1"},
        UpdateExpression="SET stock = stock - :qty",
        ConditionExpression="stock >= :qty",
        ExpressionAttributeValues={":qty": 5}
    )
except dynamodb.meta.client.exceptions.ConditionalCheckFailedException:
    print("Out of stock!")

# DynamoDB Streams → trigger Lambda on every write
# (wired in AWS console / CDK / Terraform)

import boto3

route53 = boto3.client("route53")

# Update a DNS record (e.g. after new deployment)
route53.change_resource_record_sets(
    HostedZoneId="Z1234567890ABC",
    ChangeBatch={"Changes": [{
        "Action": "UPSERT",
        "ResourceRecordSet": {
            "Name": "api.example.com",
            "Type": "A",
            "AliasTarget": {
                "HostedZoneId": "Z35SXDOTRQ7X7K",   # ALB hosted zone ID
                "DNSName": "my-alb-1234567890.us-east-1.elb.amazonaws.com",
                "EvaluateTargetHealth": True,
            }
        }
    }]}
)

# ALB — register targets and check health
elbv2 = boto3.client("elbv2")

# Get unhealthy targets
response = elbv2.describe_target_health(
    TargetGroupArn="arn:aws:elasticloadbalancing:us-east-1:123456789:targetgroup/my-tg/abc"
)
unhealthy = [t for t in response["TargetHealthDescriptions"] if t["TargetHealth"]["State"] != "healthy"]
print(f"Unhealthy targets: {unhealthy}")

import boto3

ecs = boto3.client("ecs", region_name="us-east-1")

def deploy_new_version(cluster: str, service: str, new_image: str):
    """Rolling deploy: update task definition image → force new deployment."""

    # 1. Get current task definition
    svc = ecs.describe_services(cluster=cluster, services=[service])["services"][0]
    current_td_arn = svc["taskDefinition"]
    td = ecs.describe_task_definition(taskDefinition=current_td_arn)["taskDefinition"]

    # 2. Update the image in the container definition
    containers = td["containerDefinitions"]
    for c in containers:
        if c["name"] == "app":
            c["image"] = new_image    # e.g. "123456.dkr.ecr.../my-app:v2.0.1"

    # 3. Register new task definition revision
    new_td = ecs.register_task_definition(
        family=td["family"],
        containerDefinitions=containers,
        taskRoleArn=td["taskRoleArn"],
        executionRoleArn=td["executionRoleArn"],
        networkMode=td["networkMode"],
        requiresCompatibilities=td["requiresCompatibilities"],
        cpu=td["cpu"], memory=td["memory"],
    )
    new_td_arn = new_td["taskDefinition"]["taskDefinitionArn"]

    # 4. Update service with new task definition → rolling deploy begins
    ecs.update_service(
        cluster=cluster, service=service,
        taskDefinition=new_td_arn,
        forceNewDeployment=True,
        deploymentConfiguration={
            "maximumPercent": 200,        # allow double capacity during rollout
            "minimumHealthyPercent": 100, # never go below 100% healthy
        }
    )
    print(f"Deploying {new_td_arn} to {service}")

deploy_new_version("production", "api-service", "123.dkr.ecr.us-east-1.amazonaws.com/api:v2.1")

import boto3, json

iam = boto3.client("iam")
sm  = boto3.client("secretsmanager")

# ── Attach a least-privilege policy to a role ──
policy_doc = json.dumps({
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Action": [
            "s3:GetObject",
            "s3:PutObject"
        ],
        "Resource": "arn:aws:s3:::my-bucket/uploads/*"   # narrow scope!
    }, {
        "Effect": "Allow",
        "Action": ["sqs:ReceiveMessage", "sqs:DeleteMessage"],
        "Resource": "arn:aws:sqs:us-east-1:123456789:my-queue"
    }]
})

policy = iam.create_policy(
    PolicyName="AppLeastPrivilege",
    PolicyDocument=policy_doc
)
iam.attach_role_policy(
    RoleName="my-app-role",
    PolicyArn=policy["Policy"]["Arn"]
)

# ── Read a secret (the right way in app code) ──
def get_secret(secret_name: str) -> dict:
    response = sm.get_secret_value(SecretId=secret_name)
    return json.loads(response["SecretString"])

db_creds = get_secret("prod/postgres/main")
# {"username": "app_user", "password": "...", "host": "...", "port": 5432}

# ── Enable auto-rotation for a database secret ──
sm.rotate_secret(
    SecretId="prod/postgres/main",
    RotationLambdaARN="arn:aws:lambda:us-east-1:123:function:SecretsManagerRotation",
    RotationRules={"AutomaticallyAfterDays": 30}
)

import boto3, json

sqs = boto3.client("sqs", region_name="us-east-1")
QUEUE_URL = "https://sqs.us-east-1.amazonaws.com/123456789/my-queue"

# ── Send a message ──
sqs.send_message(
    QueueUrl=QUEUE_URL,
    MessageBody=json.dumps({"job_id": "abc123", "type": "process_image"}),
    MessageAttributes={
        "priority": {"StringValue": "high", "DataType": "String"}
    }
)

# ── Send batch (up to 10 messages, 256KB total) ──
sqs.send_message_batch(
    QueueUrl=QUEUE_URL,
    Entries=[{"Id": str(i), "MessageBody": json.dumps({"item": i})} for i in range(10)]
)

# ── Consumer (long polling — cheaper than short polling) ──
def process_queue():
    while True:
        response = sqs.receive_message(
            QueueUrl=QUEUE_URL,
            MaxNumberOfMessages=10,   # batch up to 10
            WaitTimeSeconds=20,       # long poll — up to 20s wait
            VisibilityTimeout=300,    # 5 min to process before requeue
        )
        for msg in response.get("Messages", []):
            try:
                body = json.loads(msg["Body"])
                process_job(body)
                # Delete on success — removes from queue
                sqs.delete_message(QueueUrl=QUEUE_URL, ReceiptHandle=msg["ReceiptHandle"])
            except Exception as e:
                print(f"Failed: {e}")
                # Don't delete → message returns to queue after VisibilityTimeout
                # After maxReceiveCount → moves to DLQ automatically

# ── SNS fanout: one publish → many subscribers ──
sns = boto3.client("sns")
sns.publish(
    TopicArn="arn:aws:sns:us-east-1:123456789:order-events",
    Subject="OrderPlaced",
    Message=json.dumps({"order_id": "ord_789", "amount": 59.99}),
)
# All subscribers (SQS queues, Lambda, email) receive this simultaneously

# Core packages
pip install langchain langchain-openai langchain-anthropic langchain-community
pip install langchain-chroma langchain-qdrant langsmith

# Check versions
python -c "import langchain; print(langchain.__version__)"

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_community.llms import Ollama

# OpenAI
gpt4o = ChatOpenAI(model="gpt-4o", temperature=0, streaming=True)

# Anthropic Claude
claude = ChatAnthropic(model="claude-3-7-sonnet-20250219", temperature=0)

# Local model via Ollama (no API cost)
llama = Ollama(model="llama3.1:8b")

# All three share the same interface — swap freely
response = gpt4o.invoke("What is RAG?")
print(response.content)

# Streaming
for chunk in gpt4o.stream("Explain LangChain in 3 sentences"):
    print(chunk.content, end="", flush=True)

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda, RunnableParallel
from pydantic import BaseModel

llm    = ChatOpenAI(model="gpt-4o", temperature=0)
parser = StrOutputParser()

# ── Basic chain: prompt | model | parser ──
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Be concise."),
    ("human", "{question}")
])
chain = prompt | llm | parser
print(chain.invoke({"question": "What is LCEL?"}))

# ── Streaming ──
for chunk in chain.stream({"question": "Explain RAG in 3 steps"}):
    print(chunk, end="", flush=True)

# ── Batch (parallel) ──
results = chain.batch([
    {"question": "What is S3?"},
    {"question": "What is Lambda?"},
    {"question": "What is DynamoDB?"},
], config={"max_concurrency": 5})

# ── Parallel branches (run two things at once, merge) ──
parallel = RunnableParallel({
    "answer":   chain,
    "keywords": ChatPromptTemplate.from_template("List 5 keywords for: {question}") | llm | parser,
})
out = parallel.invoke({"question": "Explain GraphRAG"})
# out = {"answer": "...", "keywords": "..."}

# ── Structured output with Pydantic ──
class Movie(BaseModel):
    title: str
    year: int
    genre: str

structured_chain = prompt | llm.with_structured_output(Movie)
movie = structured_chain.invoke({"question": "Tell me about Inception"})
print(movie.title, movie.year)   # Inception  2010

# ── Fallbacks ──
fast_chain   = ChatPromptTemplate.from_template("{q}") | ChatOpenAI(model="gpt-4o-mini") | parser
strong_chain = ChatPromptTemplate.from_template("{q}") | ChatOpenAI(model="gpt-4o") | parser
chain_with_fallback = fast_chain.with_fallbacks([strong_chain])

# ── Pass-through + inject extra context ──
rag_chain = (
    RunnableParallel({"context": retriever, "question": RunnablePassthrough()})
    | ChatPromptTemplate.from_template("Context: {context}\n\nQ: {question}")
    | llm | parser
)

from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import MessagesPlaceholder
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Step 1: Rewrite query using chat history ("it" → resolve pronoun)
contextualize_prompt = ChatPromptTemplate.from_messages([
    ("system", "Rewrite the user question to be standalone, using the chat history if needed."),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
])
history_aware_retriever = create_history_aware_retriever(llm, retriever, contextualize_prompt)

# Step 2: Answer with retrieved docs
qa_prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer using only this context:\n\n{context}"),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
])
question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)

# Step 3: Compose into conversational RAG chain
rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)

# Step 4: Add session-scoped memory
store = {}  # session_id → ChatMessageHistory
def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

conversational_rag = RunnableWithMessageHistory(
    rag_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
    output_messages_key="answer",
)

# Multi-turn conversation — "it" resolved from history
r1 = conversational_rag.invoke({"input": "What is RAG?"}, config={"configurable": {"session_id": "u1"}})
r2 = conversational_rag.invoke({"input": "How is it different from fine-tuning?"}, config={"configurable": {"session_id": "u1"}})
print(r2["answer"])

from langchain.agents import AgentExecutor, create_react_agent
from langchain import hub
from langchain.tools import tool
from langchain_community.tools.tavily_search import TavilySearchResults

# ── Define custom tools with @tool decorator ──
@tool
def calculate_compound_interest(principal: float, rate: float, years: int) -> str:
    """Calculate compound interest. Args: principal (USD), rate (annual %), years."""
    amount = principal * (1 + rate / 100) ** years
    return f"${amount:,.2f} after {years} years"

@tool
def get_stock_price(ticker: str) -> str:
    """Get the current stock price for a given ticker symbol."""
    # In production: call a real finance API
    prices = {"AAPL": 189.30, "GOOGL": 173.50, "AMZN": 185.20}
    return f"{ticker}: ${prices.get(ticker.upper(), 'Not found')}"

@tool
def search_knowledge_base(query: str) -> str:
    """Search internal knowledge base for company-specific information."""
    docs = retriever.invoke(query)
    return "\n".join(d.page_content for d in docs[:3])

# ── Create agent ──
tools = [
    TavilySearchResults(max_results=3),
    calculate_compound_interest,
    get_stock_price,
    search_knowledge_base,
]

prompt = hub.pull("hwchase17/react")   # standard ReAct prompt
agent  = create_react_agent(llm=ChatOpenAI(model="gpt-4o", temperature=0), tools=tools, prompt=prompt)

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,          # shows Thought / Action / Observation loop
    max_iterations=10,     # safety cap — prevents infinite loops
    handle_parsing_errors=True,
    early_stopping_method="generate",
)

# Agent decides: search web → use calculator → search KB → synthesise
result = executor.invoke({
    "input": "If I invest $10,000 in AAPL today at their historical 15% annual growth, what do I have in 10 years?"
})
print(result["output"])

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, Annotated
import operator

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ── 1. Define State ──
# Annotated[list, operator.add] means: each node APPENDS to messages (not replaces)
class State(TypedDict):
    messages: Annotated[list, operator.add]
    query: str
    documents: list
    answer: str

# ── 2. Define Nodes (each is a plain Python function) ──
def retrieve(state: State) -> State:
    """Retrieve relevant documents."""
    docs = retriever.invoke(state["query"])
    return {"documents": docs}

def generate(state: State) -> State:
    """Generate answer from retrieved docs."""
    context = "\n".join(d.page_content for d in state["documents"])
    response = llm.invoke([
        HumanMessage(content=f"Context:\n{context}\n\nQuestion: {state['query']}")
    ])
    return {
        "answer": response.content,
        "messages": [AIMessage(content=response.content)]
    }

def grade_answer(state: State) -> str:
    """Conditional edge: route based on answer quality."""
    # Simple heuristic — in production use an LLM grader
    if len(state["answer"]) < 50 or "I don't know" in state["answer"]:
        return "retry"   # loop back to retrieve with better query
    return "done"

# ── 3. Build Graph ──
builder = StateGraph(State)

builder.add_node("retrieve", retrieve)
builder.add_node("generate", generate)

# Edges
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "generate")

# Conditional edge: done → END, retry → retrieve (loop!)
builder.add_conditional_edges(
    "generate",
    grade_answer,
    {"done": END, "retry": "retrieve"}
)

# ── 4. Compile with checkpointer (enables persistence + interrupt) ──
memory = MemorySaver()   # in-memory; swap for SqliteSaver / RedisSaver in prod
graph  = builder.compile(checkpointer=memory)

# ── 5. Run with thread_id (each thread = isolated conversation) ──
config = {"configurable": {"thread_id": "user-123-session-1"}}
result = graph.invoke({"query": "What is CRAG?", "messages": [], "documents": [], "answer": ""}, config)
print(result["answer"])

# Resume same thread later — state is persisted!
result2 = graph.invoke({"query": "How does it compare to Self-RAG?", "messages": result["messages"], "documents": [], "answer": ""}, config)

# Print ASCII representation
graph.get_graph().print_ascii()

# Export as PNG (requires pygraphviz)
from IPython.display import Image
Image(graph.get_graph().draw_mermaid_png())

# Or get Mermaid markdown
print(graph.get_graph().draw_mermaid())

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.store.memory import InMemoryStore
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import SystemMessage
import json

# ── Thread-scoped memory (per conversation) ──
# SqliteSaver: persists to disk — survives restarts
db_path = "checkpoints.db"
with SqliteSaver.from_conn_string(db_path) as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "user-42-chat-7"}}
    graph.invoke({"messages": [HumanMessage("My name is Alice")]}, config)
    # Restart app — state still there!
    graph.invoke({"messages": [HumanMessage("What's my name?")]}, config)
    # → "Your name is Alice" ✓

# ── Cross-thread memory (user profile, facts) ──
store = InMemoryStore()   # swap for PostgresStore in production

def chatbot_with_memory(state, config, *, store):
    user_id = config["configurable"]["user_id"]
    namespace = ("users", user_id, "memories")

    # Recall existing memories for this user
    memories = store.search(namespace, query=state["messages"][-1].content, limit=3)
    memory_text = "\n".join(m.value["fact"] for m in memories) if memories else ""

    system = f"""You are a personal assistant.
User facts you know:
{memory_text}"""

    response = llm.invoke([SystemMessage(content=system)] + state["messages"])

    # Extract and save new facts mentioned in this message
    new_facts = extract_facts(state["messages"][-1].content)
    for fact in new_facts:
        store.put(namespace, key=fact[:50], value={"fact": fact})

    return {"messages": [response]}

# "My dog is called Max" → stored as fact
# Next session: "What's my dog's name?" → "Max" (retrieved from store)

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent
from typing import Literal

# ── Create specialised worker agents ──
researcher = create_react_agent(
    ChatOpenAI(model="gpt-4o"), tools=[tavily_search, vectorstore_search],
    prompt="You are a research specialist. Find accurate information."
)

coder = create_react_agent(
    ChatOpenAI(model="gpt-4o"), tools=[python_repl, code_interpreter],
    prompt="You are a Python expert. Write clean, working code."
)

analyst = create_react_agent(
    ChatOpenAI(model="gpt-4o"), tools=[sql_tool, chart_tool],
    prompt="You are a data analyst. Query data and visualise insights."
)

# ── Supervisor state ──
class SupervisorState(TypedDict):
    messages: Annotated[list, operator.add]
    next_agent: str

# ── Supervisor node: LLM decides which agent to call next ──
SUPERVISOR_PROMPT = """You are a supervisor managing: researcher, coder, analyst, FINISH.
Given the conversation, decide which agent should act next, or FINISH if done.
Respond with just the agent name."""

def supervisor(state: SupervisorState) -> SupervisorState:
    response = llm.invoke([
        SystemMessage(content=SUPERVISOR_PROMPT),
        *state["messages"]
    ])
    return {"next_agent": response.content.strip()}

def route(state) -> Literal["researcher", "coder", "analyst", END]:
    next_a = state["next_agent"]
    if next_a == "FINISH": return END
    return next_a

# ── Build supervisor graph ──
builder = StateGraph(SupervisorState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", lambda s: {"messages": researcher.invoke(s)["messages"]})
builder.add_node("coder",      lambda s: {"messages": coder.invoke(s)["messages"]})
builder.add_node("analyst",    lambda s: {"messages": analyst.invoke(s)["messages"]})

builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", route)

# All workers report back to supervisor after each turn
for worker in ["researcher", "coder", "analyst"]:
    builder.add_edge(worker, "supervisor")

graph = builder.compile(checkpointer=MemorySaver())

# Complex task: supervisor orchestrates multiple agents automatically
result = graph.invoke({
    "messages": [HumanMessage(content=
        "Research LangGraph's architecture, write a Python example of a multi-agent system, "
        "and analyse what percentage of GitHub repos use LangChain vs LangGraph"
    )],
    "next_agent": ""
}, config={"configurable": {"thread_id": "complex-task-1"}})

import torch
import torch.nn.functional as F
import math

def scaled_dot_product_attention(Q, K, V, mask=None):
    """
    The heart of every transformer.
    Q, K, V: [batch, heads, seq_len, head_dim]
    """
    d_k = Q.size(-1)

    # Attention scores: how much each token should attend to each other token
    scores = Q @ K.transpose(-2, -1) / math.sqrt(d_k)  # [B, H, seq, seq]

    # Causal mask: decoder can't see future tokens (autoregressive)
    if mask is not None:
        scores = scores.masked_fill(mask == 0, float('-inf'))

    # Convert scores to probabilities
    attn_weights = F.softmax(scores, dim=-1)

    # Weighted sum of values
    return attn_weights @ V, attn_weights   # [B, H, seq, head_dim]

class MultiHeadAttention(torch.nn.Module):
    def __init__(self, d_model=768, n_heads=12):
        super().__init__()
        self.n_heads  = n_heads
        self.d_head   = d_model // n_heads
        self.W_q = torch.nn.Linear(d_model, d_model, bias=False)
        self.W_k = torch.nn.Linear(d_model, d_model, bias=False)
        self.W_v = torch.nn.Linear(d_model, d_model, bias=False)
        self.W_o = torch.nn.Linear(d_model, d_model, bias=False)

    def forward(self, x, mask=None):
        B, T, C = x.shape   # batch, seq_len, d_model

        # Project → split into heads
        def split_heads(w): return w.view(B, T, self.n_heads, self.d_head).transpose(1, 2)
        Q, K, V = split_heads(self.W_q(x)), split_heads(self.W_k(x)), split_heads(self.W_v(x))

        # Attention
        out, weights = scaled_dot_product_attention(Q, K, V, mask)

        # Merge heads → project out
        out = out.transpose(1, 2).contiguous().view(B, T, C)
        return self.W_o(out), weights

# Flash Attention (PyTorch 2.0+) — same math, 3× faster, 10× less memory
# Use this in production, not the above:
with torch.backends.cuda.sdp_kernel(enable_flash=True):
    out = F.scaled_dot_product_attention(Q, K, V, is_causal=True)