How to Build
Generative AI
Applications —
The Complete Developer
Guide

from anthropic import Anthropic

client = Anthropic()

# ── 1. ZERO-SHOT ──────────────────────────────────────────────
# No examples. Works for well-defined tasks the model knows well.
zero_shot = {
    "role": "user",
    "content": "Classify the sentiment of this review as POSITIVE, NEGATIVE, or NEUTRAL.\n\nReview: 'The battery life is terrible but the screen is beautiful.'"
}

# ── 2. FEW-SHOT ────────────────────────────────────────────────
# Provide 2-5 examples to anchor the output format and style.
few_shot_system = """You extract structured data from unstructured text.

Examples:
Input: "John Smith, 34, joined us from Google last Monday."
Output: {"name": "John Smith", "age": 34, "previous_company": "Google"}

Input: "Maria Garcia (28) previously at Meta. Started yesterday."
Output: {"name": "Maria Garcia", "age": 28, "previous_company": "Meta"}

Now extract from the input. Respond ONLY with valid JSON, no other text."""

# ── 3. CHAIN-OF-THOUGHT (CoT) ──────────────────────────────────
# Ask the model to reason step by step before answering.
# Critical for maths, logic, and multi-step reasoning tasks.
cot_prompt = """Analyse whether this customer qualifies for a premium refund.
Policy: Full refund if: purchase <30 days ago AND item is defective AND
        customer has <2 previous refund requests.

Customer data: Purchase 12 days ago. Item has a reported screen crack.
               Previous refund requests: 1.

Think step by step, then give your final decision as JSON:
{"qualifies": true/false, "reason": "brief explanation"}"""

# ── 4. STRUCTURED OUTPUT ──────────────────────────────────────
# Force specific JSON schema for reliable parsing downstream.
resp = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=500,
    system="""Extract the requested information and respond ONLY with a
JSON object matching this exact schema. No other text.
Schema: {"intent": string, "entities": object, "confidence": float}""",
    messages=[{"role": "user", "content": "Cancel my subscription for account ACC-9842"}]
)
import json
parsed = json.loads(resp.content[0].text)

# ── 5. ROLE + PERSONA ─────────────────────────────────────────
# Give the model a clear identity to constrain tone and behaviour.
persona_system = """You are a senior software engineer at a fintech company.
You review code for: security vulnerabilities, performance issues, and
compliance with PCI-DSS. You are direct. You reference specific line
numbers. You never make vague suggestions like 'improve error handling'
without showing the exact code change needed."""

# ── 6. SELF-CONSISTENCY / VERIFICATION ───────────────────────
# Ask the model to verify its own output before returning.
verify_prompt = """Answer the question, then verify your answer.

Question: What is 15% of $847.60?

Format:
Step 1 — calculation: [your working]
Answer: [result]
Verification: [re-check the maths a different way]
Final answer: [confirmed result]"""

from langchain.document_loaders import PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_pinecone import PineconeVectorStore
from langchain.chains import create_retrieval_chain
from langchain_core.prompts import ChatPromptTemplate

# ── INGESTION ────────────────────────────────────────────────

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

def ingest_documents(docs_path: str, index_name: str):
    # Load all PDFs in a directory
    loader = DirectoryLoader(docs_path, glob="**/*.pdf",
                              loader_cls=PyPDFLoader)
    raw_docs = loader.load()

    # Chunk: 1000 chars, 100 char overlap to preserve context at boundaries
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000, chunk_overlap=100,
        separators=["\n\n", "\n", ". ", " "]
    )
    chunks = splitter.split_documents(raw_docs)

    # Embed and store
    vectorstore = PineconeVectorStore.from_documents(
        chunks, embeddings, index_name=index_name
    )
    return vectorstore

# ── QUERY ─────────────────────────────────────────────────────

RAG_SYSTEM_PROMPT = """You are a precise Q&A assistant.
Answer ONLY from the provided context. If the answer is not in the context,
say "I don't have that information in my knowledge base."
Never guess. Cite the source document and page number.

Context: {context}"""

def build_rag_chain(index_name: str):
    vectorstore = PineconeVectorStore.from_existing_index(
        index_name, embeddings
    )
    # Fetch top-5 most relevant chunks per query
    retriever = vectorstore.as_retriever(
        search_type="mmr",      # Maximum Marginal Relevance — reduces duplicates
        search_kwargs={"k": 5, "fetch_k": 20}
    )

    prompt = ChatPromptTemplate.from_messages([
        ("system", RAG_SYSTEM_PROMPT),
        ("human", "{input}")
    ])

    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    return create_retrieval_chain(retriever, prompt | llm)

# Usage:
# chain = build_rag_chain("my-knowledge-base")
# result = chain.invoke({"input": "What is the cancellation policy?"})
# print(result["answer"])

import json
from openai import OpenAI

client = OpenAI()

# Fine-tuning requires JSONL training data in chat format.
# Each line = one training example with system, user, and assistant turns.
# Rule of thumb: 50+ examples for basic improvement, 200+ for reliable results.

training_examples = [
    {
        "messages": [
            {"role": "system",    "content": "You are a customer support agent for Acme Corp. Always respond in JSON."},
            {"role": "user",      "content": "I want to cancel my subscription"},
            {"role": "assistant", "content": '{"intent": "cancel_subscription", "action": "collect_account_id", "response": "I can help you cancel your subscription. Could you provide your account ID?"}'}
        ]
    },
    # ... 200+ more examples
]

# Save training data as JSONL
with open("training_data.jsonl", "w") as f:
    for example in training_examples:
        f.write(json.dumps(example) + "\n")

# Upload training file
training_file = client.files.create(
    file=open("training_data.jsonl", "rb"),
    purpose="fine-tune"
)

# Create fine-tuning job
job = client.fine_tuning.jobs.create(
    training_file=training_file.id,
    model="gpt-4o-mini-2024-07-18",  # Fine-tune the cheap fast model
    hyperparameters={
        "n_epochs": 3,               # 3 is typically a good starting point
        "learning_rate_multiplier": 0.1
    },
    suffix="support-agent-v1"
)
print(f"Fine-tuning job started: {job.id}")

# Monitor: client.fine_tuning.jobs.retrieve(job.id)
# Use: model="ft:gpt-4o-mini-2024-07-18:your-org:support-agent-v1:abc123"

import anthropic, json, requests

client = anthropic.Anthropic()

# Define the tools your agent can use
TOOLS = [
    {
        "name": "web_search",
        "description": "Search the web for current information. Use for recent news, prices, facts.",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "The search query"}},
            "required": ["query"]
        }
    },
    {
        "name": "get_order_details",
        "description": "Retrieve order details from the database by order ID",
        "input_schema": {
            "type": "object",
            "properties": {"order_id": {"type": "string"}},
            "required": ["order_id"]
        }
    }
]

def execute_tool(tool_name: str, tool_input: dict) -> str:
    if tool_name == "web_search":
        # Integrate your search API (Tavily, SerpAPI, etc.)
        return f"Search results for '{tool_input['query']}': [results here]"
    elif tool_name == "get_order_details":
        # Query your actual database
        return json.dumps({"order_id": tool_input["order_id"],
                              "status": "shipped", "eta": "2026-05-20"})
    return "Tool not found"

def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            tools=TOOLS,
            messages=messages
        )

        # If stop reason is end_turn — we have the final answer
        if response.stop_reason == "end_turn":
            return response.content[0].text

        # If stop reason is tool_use — execute the tool and loop back
        if response.stop_reason == "tool_use":
            # Append model's tool call to messages
            messages.append({"role": "assistant", "content": response.content})

            # Execute each tool call and collect results
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })

            # Feed results back to model and continue loop
            messages.append({"role": "user", "content": tool_results})

from fastapi import FastAPI, HTTPException, Depends
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from langfuse import Langfuse
import anthropic, redis, json, time

app = FastAPI()
lf  = Langfuse()            # LLM observability
r   = redis.Redis(host="redis", decode_responses=True)
ai  = anthropic.Anthropic()

class ChatRequest(BaseModel):
    session_id: str
    message:    str

def get_history(session_id: str) -> list:
    raw = r.get(f"session:{session_id}")
    return json.loads(raw) if raw else []

@app.post("/chat/stream")
async def stream_chat(req: ChatRequest):
    history = get_history(req.session_id)
    history.append({"role": "user", "content": req.message})

    # Trace this call in Langfuse
    trace = lf.trace(name="chat", session_id=req.session_id,
                       input=req.message)

    async def token_stream():
        start        = time.time()
        full_response = ""
        input_tokens  = 0
        output_tokens = 0

        with ai.messages.stream(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            system="You are a helpful assistant. Be concise and accurate.",
            messages=history[-12:]  # Keep last 12 turns
        ) as stream:
            for chunk in stream.text_stream:
                full_response += chunk
                yield f"data: {json.dumps({'delta': chunk})}\n\n"

            usage = stream.get_final_message().usage
            input_tokens  = usage.input_tokens
            output_tokens = usage.output_tokens

        # Persist history + log metrics
        history.append({"role": "assistant", "content": full_response})
        r.setex(f"session:{req.session_id}", 3600,
                json.dumps(history))
        trace.update(output=full_response,
                      usage={"input": input_tokens, "output": output_tokens,
                             "latency_ms": int((time.time()-start)*1000)})
        yield "data: [DONE]\n\n"

    return StreamingResponse(token_stream(), media_type="text/event-stream")

import json
from anthropic import Anthropic
from dataclasses import dataclass
from typing import Optional

judge = Anthropic()

@dataclass
class EvalResult:
    faithfulness:  float   # 0–1: Is answer grounded in context?
    relevance:     float   # 0–1: Does it answer the question?
    completeness:  float   # 0–1: Are important aspects covered?
    overall:       float   # Weighted average
    notes:         str     # Judge's reasoning

def evaluate_rag_response(
    question: str,
    context:  str,
    answer:   str
) -> EvalResult:
    prompt = f"""Evaluate this RAG system response as an expert judge.

QUESTION: {question}

RETRIEVED CONTEXT:
{context}

SYSTEM ANSWER:
{answer}

Score each dimension 0.0 to 1.0 and explain your reasoning.
Respond ONLY with JSON:
{{
  "faithfulness": 0.0,   // Is every claim in the answer supported by context?
  "relevance": 0.0,      // Does the answer actually address the question?
  "completeness": 0.0,   // Are all important aspects of the question covered?
  "notes": "brief reasoning"
}}"""

    resp = judge.messages.create(
        model="claude-opus-4-20250514",  # Use best model for judging
        max_tokens=400,
        messages=[{"role": "user", "content": prompt}]
    )

    scores = json.loads(resp.content[0].text)
    return EvalResult(
        faithfulness=scores["faithfulness"],
        relevance=scores["relevance"],
        completeness=scores["completeness"],
        overall=0.5*scores["faithfulness"] + 0.3*scores["relevance"]
               + 0.2*scores["completeness"],
        notes=scores["notes"]
    )

# Run eval suite over your golden test set
# Flag any response with overall < 0.7 for human review

import re
from openai import OpenAI

client = OpenAI()

# ── PROMPT INJECTION DETECTION ────────────────────────────────
INJECTION_PATTERNS = [
    re.compile(r'ignore (all )?(previous |prior )?instructions?', re.I),
    re.compile(r'(system|assistant|human)\s*:', re.I),
    re.compile(r'you are now (dan|evil|unfiltered|jailbreak)', re.I),
    re.compile(r'disregard (your|the) (training|guidelines|rules)', re.I),
    re.compile(r'print (the )?system prompt', re.I),
]

def check_prompt_injection(text: str) -> bool:
    return any(p.search(text) for p in INJECTION_PATTERNS)

# ── PII DETECTION IN OUTPUT ───────────────────────────────────
PII_PATTERNS = {
    "credit_card":  re.compile(r'\b\d{4}[\s\-]?\d{4}[\s\-]?\d{4}[\s\-]?\d{4}\b'),
    "ssn":          re.compile(r'\b\d{3}-\d{2}-\d{4}\b'),
    "email":        re.compile(r'[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}'),
    "phone_india":  re.compile(r'[6-9]\d{9}'),
}

def redact_pii(text: str) -> str:
    for label, pattern in PII_PATTERNS.items():
        text = pattern.sub(f"[{label.upper()} REDACTED]", text)
    return text

# ── OPENAI MODERATION ─────────────────────────────────────────
def check_content_policy(text: str) -> bool:
    result = client.moderations.create(input=text)
    return result.results[0].flagged  # True = policy violation

# ── SAFETY WRAPPER ────────────────────────────────────────────
def safe_generate(user_input: str, generate_fn) -> str:
    # 1. Check input
    if check_prompt_injection(user_input):
        return "I can't process that request."
    if check_content_policy(user_input):
        return "Your message contains content I can't help with."

    # 2. Generate
    response = generate_fn(user_input)

    # 3. Check and clean output
    response = redact_pii(response)
    if check_content_policy(response):
        return "I'm unable to provide that response. Please rephrase your question."

    return response