# The old approach
prompt = "Write a comprehensive research report on quantum computing"
response = llm.generate(prompt)
# Hope it's good 🤞

# The agentic approach
outline = llm.generate("Create an outline for quantum computing report")
research_data = web_search(extract_topics(outline))
first_draft = llm.generate(f"Write report with this outline and data...")
critique = llm.reflect("Review this draft for gaps and weaknesses...")
final_report = llm.revise(first_draft, critique, research_data)

def calculate_interest(principal, rate, time):
    return principal * rate * time  # Wrong formula!

# Generation
code_v1 = llm.generate("Write compound interest calculator")

# Reflection
critique = llm.reflect("""
Review this code for correctness:
{code_v1}

Check: formulas, edge cases, variable names
""")
# Output: "Uses simple interest formula, should be compound"

# Revision
code_v2 = llm.revise(code_v1, critique)
# Correct implementation

def generate_with_reflection(task: str, max_iterations: int = 3):
    """
    Generate with self-correction loop
    """
    output = llm.generate(task)
    
    for i in range(max_iterations):
        # External feedback (execution, validation, etc.)
        feedback = validate_output(output)
        
        if feedback.is_valid:
            return output
            
        # Reflect with concrete feedback
        critique = llm.reflect(f"""
        Task: {task}
        Output: {output}
        Issues found: {feedback.errors}
        
        Provide specific improvements.
        """)
        
        output = llm.revise(output, critique)
    
    return output

# 1. You define a function
def get_weather(city: str, units: str = "celsius") -> dict:
    """
    Get current weather for a city.
    
    Args:
        city: City name
        units: Temperature units (celsius/fahrenheit)
    """
    return weather_api.fetch(city, units)

# 2. LLM sees this description
{
  "name": "get_weather",
  "description": "Get current weather for a city.",
  "parameters": {
    "city": {"type": "string"},
    "units": {"type": "string", "enum": ["celsius", "fahrenheit"]}
  }
}

# 3. When user asks "What's the weather in Tokyo?"
# LLM outputs:
{
  "tool_call": {
    "function": "get_weather",
    "arguments": {"city": "Tokyo", "units": "celsius"}
  }
}

# 4. YOU execute the function (LLM doesn't do it)
result = get_weather("Tokyo", "celsius")

# 5. Feed result back to LLM
# LLM generates: "It's currently 18°C in Tokyo..."

# Instead of creating tools for:
# - calculate_compound_interest()
# - calculate_roi()
# - calculate_depreciation()
# - calculate_... (hundreds more)

# Just give it ONE tool:
def execute_code(code: str) -> str:
    """Execute Python code and return output"""
    return run_in_sandbox(code)

# Now it can do anything:
llm: "What's the compound interest on $1000 at 5% for 10 years?"
# Generates:
principal = 1000
rate = 0.05
time = 10
result = principal * (1 + rate) ** time
print(f"${result:.2f}")
# Output: $1628.89

# LLM generated this once:
import os
os.system("rm -rf *.py")  # 😱

# ❌ Dangerous
exec(llm_generated_code)

# ✅ Safe
import docker

def execute_safely(code: str, timeout: int = 30):
    client = docker.from_env()
    container = client.containers.run(
        "python:3.11-slim",
        command=f"python -c '{code}'",
        timeout=timeout,
        mem_limit="256m",
        network_disabled=True,
        remove=True
    )
    return container.decode()

# Using GitHub MCP server
async with mcp.connect("github") as github:
    readme = await github.get_file("org/repo", "README.md")
    prs = await github.list_pull_requests("org/repo")
    
# Your LLM can now naturally interact with GitHub

# Just collect 10-15 examples
test_cases = [
    {"input": "...", "expected": "..."},
    # ... 9 more
]

# Simple pass/fail
correct = sum(1 for t in test_cases if system(t["input"]) == t["expected"])
print(f"Accuracy: {correct}/{len(test_cases)}")

class EvaluationSuite:
    def __init__(self):
        self.test_cases = load_test_cases()  # Now 50+ examples
        
    def evaluate_end_to_end(self):
        """Overall system performance"""
        return {
            "accuracy": self._calculate_accuracy(),
            "latency_p95": self._measure_latency(),
            "cost_per_request": self._calculate_cost()
        }
    
    def evaluate_component(self, component_name):
        """Individual component performance"""
        return self._run_component_eval(component_name)

prompt = """
Which output is better?

Output A: {output_a}
Output B: {output_b}

Answer: A or B
"""
# LLMs consistently favor the first option ~60% of the time

rubric = """
Evaluate this chart on these criteria (0 or 1 point each):
1. Has clear, descriptive title
2. Axis labels are present and readable
3. Uses appropriate chart type for data
4. Data points are clearly visible
5. Legend present (if needed)
6. Professional appearance
7. No visual clutter

Return JSON: {"scores": [1,1,0,1,1,1,0], "total": 5, "explanation": "..."}
"""

score = llm_judge(output, rubric)

# 1. Focus on failures only
failures = [test for test in test_suite if test.failed()]

# 2. Examine traces for each failure
for failure in failures:
    traces = failure.get_intermediate_outputs()
    # Look at output of each component
    
# 3. Build error taxonomy spreadsheet

# Instead of running full workflow every time...
# (which might take 60s and cost $0.50)

# Just evaluate the component you're improving
# (takes 5s, costs $0.02)

def evaluate_search_component():
    """Faster iteration on search quality"""
    gold_standard = {
        "quantum computing": ["arxiv.org/...", "nature.com/...", "quantum.gov/..."],
        "black holes": ["nasa.gov/...", "eso.org/...", "arxiv.org/..."],
        # ...
    }
    
    for query, expected_sources in gold_standard.items():
        actual_results = search_engine(query)
        f1_score = calculate_overlap(actual_results, expected_sources)
        # Quick feedback on search quality

# Developer decides the steps
step1 = search_web(query)
step2 = select_sources(step1)
step3 = write_essay(step2)

# LLM generates a plan, then executes it
plan = llm.plan(query, available_tools)
# Plan might be: [search, validate, search_more, synthesize]

for step in plan:
    result = execute_step(step)

# Deterministic approach (breaks easily):
if "order" in query:
    check_order_status()
elif "return" in query:
    process_return()
elif "product" in query:
    search_catalog()
# ... 50 more elif statements

# Planning approach:
plan = llm.create_plan(query, tools=[
    check_order_status,
    process_return,
    search_catalog,
    check_inventory,
    contact_support_team
])
# LLM decides which tools to use and in what order

system_prompt = """
You have access to these tools:
- get_item_descriptions(category): Get product descriptions
- check_inventory(item_ids): Check stock levels
- get_item_price(item_ids): Get current prices
- process_return(order_id, item_id): Handle returns

Create a step-by-step plan in JSON to answer user queries.

Format:
{
  "plan": [
    {
      "step": 1,
      "description": "What this step does",
      "tool": "tool_name",
      "arguments": {...},
      "depends_on": [] // Previous step numbers needed
    }
  ]
}
"""

{
  "plan": [
    {
      "step": 1,
      "description": "Find all round sunglasses",
      "tool": "get_item_descriptions",
      "arguments": {"category": "sunglasses", "filter": "round"},
      "depends_on": []
    },
    {
      "step": 2,
      "description": "Check which round sunglasses are in stock",
      "tool": "check_inventory",
      "arguments": {"item_ids": "from_step_1"},
      "depends_on": [1]
    },
    {
      "step": 3,
      "description": "Get prices for in-stock items",
      "tool": "get_item_price",
      "arguments": {"item_ids": "from_step_2"},
      "depends_on": [2]
    }
  ]
}

def execute_plan(plan, context):
    results = {}
    
    for step in plan["plan"]:
        # Wait for dependencies
        for dep in step["depends_on"]:
            if dep not in results:
                raise Exception(f"Step {step['step']} depends on incomplete step {dep}")
        
        # Resolve arguments from previous steps
        args = resolve_arguments(step["arguments"], results)
        
        # Execute
        tool = get_tool(step["tool"])
        results[step["step"]] = tool(**args)
    
    return results

# You'd need dozens of specific tools:
get_column_max()
get_column_mean()
filter_rows()
get_unique_values()
sort_by_column()
group_by()
# ... hundreds more

system_prompt = """
You have access to Python with pandas library.
Write code to answer the user's query.

Return format:
<execute_python>
# Your code here
</execute_python>
"""

# User: "What were the amounts of the last 5 transactions?"

# LLM generates:

import pandas as pd

# Step 1: Load data
df = pd.read_csv('transactions.csv')

# Step 2: Ensure proper date parsing
df['date'] = pd.to_datetime(df['date'])

# Step 3: Sort by date (most recent first)
df_sorted = df.sort_values('date', ascending=False)

# Step 4: Get last 5 transactions
last_5 = df_sorted.head(5)

# Step 5: Extract amounts
amounts = last_5['amount'].tolist()

print(amounts)

def safe_code_execution(code: str, timeout: int = 30):
    """
    Execute LLM-generated code safely
    """
    # 1. Sandbox environment
    container = create_sandbox()
    
    # 2. Resource limits
    container.set_memory_limit("512MB")
    container.set_cpu_limit(1.0)
    
    # 3. Network isolation
    container.disable_network()
    
    # 4. Timeout
    try:
        result = container.run(code, timeout=timeout)
    except TimeoutError:
        return {"error": "Execution timeout"}
    
    return result

# User: "Build a REST API for a todo app with auth"

# LLM generates plan:
"""
☐ Step 1: Design database schema
☐ Step 2: Implement user model and authentication
☐ Step 3: Create todo model with user association
☐ Step 4: Build API endpoints (CRUD for todos)
☐ Step 5: Add input validation and error handling
☐ Step 6: Write unit tests
☐ Step 7: Write integration tests
☐ Step 8: Create API documentation
"""

# Then executes each step systematically
# If tests fail → revise → retest
# Until working application delivered

def planning_with_guardrails(query, max_steps=10):
    """
    Planning with safety limits
    """
    plan = llm.generate_plan(query)
    
    # Validate plan before execution
    if len(plan) > max_steps:
        return {"error": "Plan too complex, please simplify query"}
    
    # Check for dangerous patterns
    if contains_risky_operations(plan):
        return {"error": "Plan contains operations requiring human approval"}
    
    # Execute with monitoring
    for step in plan:
        result = execute_with_monitoring(step)
        if result.error_rate > 0.5:
            return {"error": "Plan execution failing, aborting"}
    
    return result

# One agent doing everything (gets confused)
marketing_agent = LLM("""
You're responsible for research, design, and copywriting...
""")
# Tries to do too much, does nothing well

researcher = LLM("Expert at market analysis and competitor research")
designer = LLM("Skilled at data visualization and graphic design")
writer = LLM("Compelling marketing copywriter")

# Each does what it's best at

# Clean handoffs, predictable flow
research = researcher.analyze("sunglasses market trends")
graphics = designer.create(research)
brochure = writer.compose(research, graphics)

manager_prompt = """
You're a marketing manager with this team:
- Researcher: Market analysis expert
- Designer: Visual content creator
- Writer: Marketing copywriter

Create a plan and delegate tasks.
"""

manager = LLM(manager_prompt, tools=[
    call_researcher,
    call_designer,
    call_writer
])

# Manager decides who does what and when
plan = manager.create_plan("Summer sunglasses campaign")
# Might decide: researcher → designer + writer in parallel → final review

# Everyone can talk to everyone
agents = [researcher, designer, writer, editor]

for agent in agents:
    agent.team_members = [a for a in agents if a != agent]

# Free-form collaboration
conversation = []
while not consensus_reached():
    for agent in agents:
        message = agent.respond_to(conversation)
        conversation.append(message)

final_output = synthesize(conversation)

class Agent:
    def __init__(self, role: str, tools: list, prompt_template: str):
        self.role = role
        self.tools = tools
        self.prompt = prompt_template
        
    def execute(self, task: str, context: dict = None):
        """
        Execute task with role-specific behavior
        """
        full_prompt = self.prompt.format(
            task=task,
            context=context or {}
        )
        
        return llm.generate(
            full_prompt,
            tools=self.tools
        )

# Create specialized agents
researcher = Agent(
    role="Market Researcher",
    tools=[web_search, fetch_article],
    prompt_template="""
    You're an expert market researcher. Analyze:
    - Market trends and patterns
    - Competitor positioning
    - Customer preferences
    
    Task: {task}
    Context: {context}
    
    Provide data-driven insights with citations.
    """
)

designer = Agent(
    role="Graphic Designer",
    tools=[generate_image, create_chart, code_execution],
    prompt_template="""
    You're a skilled graphic designer. Create:
    - Data visualizations
    - Marketing graphics
    - Brand-consistent artwork
    
    Task: {task}
    Context: {context}
    
    Focus on visual impact and clarity.
    """
)

from langgraph.graph import Graph

# Define workflow
workflow = Graph()

# Add agents as nodes
workflow.add_node("researcher", researcher.execute)
workflow.add_node("designer", designer.execute)
workflow.add_node("writer", writer.execute)

# Define edges (communication flow)
workflow.add_edge("researcher", "designer")
workflow.add_edge("researcher", "writer")
workflow.add_edge("designer", "writer")

# Compile and run
app = workflow.compile()
result = app.invoke({"query": "Create sunglasses campaign"})

import time
from dataclasses import dataclass

@dataclass
class PerformanceMetrics:
    component: str
    duration_ms: float
    cost_dollars: float

def benchmark_workflow(query):
    metrics = []
    
    # Time each component
    start = time.time()
    terms = generate_search_terms(query)
    metrics.append(PerformanceMetrics(
        "generate_terms",
        (time.time() - start) * 1000,
        calculate_cost(terms)
    ))
    
    start = time.time()
    results = web_search(terms)
    metrics.append(PerformanceMetrics(
        "web_search",
        (time.time() - start) * 1000,
        0.016  # API cost
    ))
    
    # ... rest of workflow
    
    return metrics

# Example output:
# generate_terms: 7000ms, $0.04
# web_search: 5000ms, $0.016
# fetch_pages: 11000ms, $0.040
# write_essay: 18000ms, $2.80
# TOTAL: 41000ms, $2.896

# Before: 18s for essay generation with GPT-4
essay = await llm.generate(
    model="gpt-4o",
    prompt=essay_prompt
)

# Strategy 1: Faster model (if quality acceptable)
essay = await llm.generate(
    model="gpt-4o-mini",  # 3x faster, 70% cheaper
    prompt=essay_prompt
)
# Now: 6s

# Strategy 2: Streaming (feels faster)
async for chunk in llm.stream(model="gpt-4o", prompt=essay_prompt):
    yield chunk  # User sees progress immediately

# Strategy 3: Specialized provider (same model, faster hardware)
essay = await llm.generate(
    model="groq:llama3-70b",  # Specialized hardware
    prompt=essay_prompt
)
# Now: 3s

# Before: Sequential fetching (15s total)
page1 = fetch(url1)  # 5s
page2 = fetch(url2)  # 5s
page3 = fetch(url3)  # 5s

# After: Parallel fetching (5s total)
import asyncio

pages = await asyncio.gather(
    fetch(url1),
    fetch(url2),
    fetch(url3)
)
# Now: 5s (max of all requests)

# Savings: 10s per request
# At 1000 requests/day: 2.8 hours saved daily

def analyze_costs():
    """
    Break down costs by component
    """
    costs = {
        'generate_search_terms': {
            'model': 'gpt-4o',
            'avg_tokens': 200,
            'cost_per_request': 0.04,
            'requests_per_day': 1000,
            'daily_cost': 40.00
        },
        'web_search': {
            'cost_per_request': 0.016,
            'requests_per_day': 1000,
            'daily_cost': 16.00
        },
        'write_essay': {
            'model': 'gpt-4o',
            'avg_tokens': 3000,
            'cost_per_request': 2.80,
            'requests_per_day': 1000,
            'daily_cost': 2800.00  # ← 97% of total cost!
        }
    }
    
    total_daily = sum(c['daily_cost'] for c in costs.values())
    # $2,856/day = $85,680/month 😱
    
    return costs

# Before: GPT-4 for everything
write_essay_cost = 2.80 * 1000 = $2,800/day

# After: Strategic model selection
costs = {
    # Non-critical: Use cheap models
    'search_terms': ('claude-haiku', 0.01),  # Was $40, now $10
    'source_selection': ('gpt-3.5', 0.02),   # Was $40, now $20
    
    # Critical quality: Keep premium model
    'essay_writing': ('gpt-4o', 2.80),       # Still $2,800
    
    # OR: Use cheaper premium model
    'essay_writing': ('claude-sonnet', 0.80), # Now $800 (71% savings!)
}

# New total: $830/day = $24,900/month
# Savings: $60,780/month 💰

from functools import lru_cache
import hashlib

class SmartCache:
    def __init__(self, redis_client):
        self.cache = redis_client
        self.ttl = 3600  # 1 hour
    
    def cached_llm_call(self, prompt: str, model: str) -> str:
        """
        Cache LLM responses for identical prompts
        """
        cache_key = hashlib.sha256(
            f"{model}:{prompt}".encode()
        ).hexdigest()
        
        # Check cache
        if cached := self.cache.get(cache_key):
            return cached  # $0.00 cost!
        
        # Call LLM
        response = llm.generate(model=model, prompt=prompt)
        
        # Store in cache
        self.cache.setex(cache_key, self.ttl, response)
        
        return response

# Real-world impact:
# - Research queries: 30% cache hit rate
# - Savings: 30% of $24,900 = $7,470/month

# Before: Individual processing
for item in items:  # 1000 items
    result = llm.process(item)  # $0.10 each = $100

# After: Batch processing
batch_size = 10
for i in range(0, len(items), batch_size):
    batch = items[i:i+batch_size]
    results = llm.process_batch(batch)  # $0.50 per batch
    
# Cost: 100 batches × $0.50 = $50 (50% savings)

class ProductionReadiness:
    def check_quality(self):
        """
        Minimum quality thresholds
        """
        assert self.end_to_end_accuracy > 0.90, "Not accurate enough"
        assert self.critical_component_accuracy > 0.95, "Component failing"
        assert self.error_analysis_complete, "Don't know failure modes"
        assert self.edge_cases_handled, "Will break on edge cases"
    
    def check_performance(self):
        """
        Performance requirements
        """
        assert self.p95_latency < self.sla_requirement
        assert self.cost_per_request < self.budget_limit
        assert self.parallelization_optimized
    
    def check_reliability(self):
        """
        Production hardening
        """
        assert self.error_handling_comprehensive
        assert self.retry_logic_implemented
        assert self.timeout_protection_enabled
        assert self.rate_limiting_configured
        assert self.graceful_degradation_works
    
    def check_safety(self):
        """
        Security and safety
        """
        assert self.code_execution_sandboxed
        assert self.input_validation_active
        assert self.output_filtering_enabled
        assert self.pii_protection_configured
        assert self.content_safety_checked
    
    def check_observability(self):
        """
        Monitoring and debugging
        """
        assert self.logging_comprehensive
        assert self.trace_storage_enabled
        assert self.metrics_dashboard_ready
        assert self.alerting_configured
        assert self.cost_tracking_active

import structlog
from prometheus_client import Counter, Histogram

# Structured logging
logger = structlog.get_logger()

# Metrics
requests_total = Counter('agent_requests_total', 'Total requests')
request_duration = Histogram('agent_request_duration_seconds', 'Request duration')
request_cost = Histogram('agent_request_cost_dollars', 'Request cost')
errors_total = Counter('agent_errors_total', 'Total errors', ['component'])

class AgentObservability:
    def trace_request(self, request_id: str):
        """
        Comprehensive request tracing
        """
        with request_duration.time():
            try:
                # Log request start
                logger.info("request_started",
                    request_id=request_id,
                    query=request.query
                )
                
                # Execute workflow
                result = self.execute_workflow(request)
                
                # Log success
                logger.info("request_completed",
                    request_id=request_id,
                    latency_ms=result.latency,
                    cost_dollars=result.cost,
                    output_length=len(result.output)
                )
                
                # Update metrics
                requests_total.inc()
                request_cost.observe(result.cost)
                
                return result
                
            except Exception as e:
                # Log failure
                logger.error("request_failed",
                    request_id=request_id,
                    error=str(e),
                    component=e.component
                )
                
                # Update error metrics
                errors_total.labels(component=e.component).inc()
                
                raise

"""
INCIDENT RESPONSE PLAYBOOK

1. DETECT
   - Alert fires from monitoring
   - User reports issues
   - Cost spike detected

2. TRIAGE
   - Check dashboard: Which component failing?
   - Review recent traces: Pattern in failures?
   - Check error logs: New error types?

3. IMMEDIATE MITIGATION
   - Can we roll back recent changes?
   - Can we disable problematic component?
   - Can we route to fallback system?

4. ROOT CAUSE ANALYSIS
   - Examine traces of failed requests
   - Compare to successful requests
   - Check for:
     * Model provider issues
     * API rate limits hit
     * Input pattern changes
     * Prompt regressions

5. FIX & VERIFY
   - Implement fix
   - Test with recent failures
   - Monitor closely for 24h

6. POSTMORTEM
   - Document what happened
   - Update evals to catch this
   - Improve monitoring
   - Update runbook
"""

week_1_2 = {
    "build": "Simple agent with deterministic workflow",
    "focus": "Output quality above all else",
    "team_size": "2-3 engineers"
}

week_3_4 = {
    "build": "Basic eval suite (20-30 examples)",
    "focus": "Identify failure modes",
    "deliverable": "Error analysis document"
}

week_5_8 = {
    "build": "Iterate based on evals",
    "focus": "Get to 85-90% accuracy",
    "gate": "Quality threshold before proceeding"
}

month_3 = {
    "build": [
        "Comprehensive error handling",
        "Retry logic and timeouts",
        "Monitoring and alerting",
        "Component-level evals"
    ],
    "focus": "Reliability engineering"
}

month_4 = {
    "build": [
        "Latency optimization",
        "Cost optimization (if needed)",
        "Load testing",
        "Incident playbooks"
    ],
    "focus": "Performance and ops readiness"
}

month_5 = {
    "build": [
        "Expand eval set (50-100 examples)",
        "Add reflection where beneficial",
        "Implement caching layer"
    ],
    "focus": "Quality improvements"
}

month_6 = {
    "build": [
        "Planning capabilities (if needed)",
        "Multi-agent (if justified)",
        "Reusable agent components"
    ],
    "focus": "Advanced capabilities"
}

monthly_costs = {
    "compute": {
        "development": 2000,   # Experimentation
        "production": 10000,   # 10K queries/day
        "overhead": 2000       # Monitoring, logs
    },
    "people": {
        "engineers": 6 * 15000,  # Loaded cost
        "pm": 12000
    },
    "tools": {
        "monitoring": 500,
        "frameworks": 200,
        "misc": 300
    }
}

total_monthly = (
    sum(monthly_costs["compute"].values()) +
    sum(monthly_costs["people"].values()) +
    sum(monthly_costs["tools"].values())
)
# = $116,000/month

# With 10K queries/day at $5 value each
revenue_potential = 10000 * 5 * 30  # $1.5M/month

# ROI = 1300% (if you capture the value!)