Capstone — Deploy an AI Agent to Production

deploy/
├── api/
│   ├── __init__.py
│   ├── main.py           # FastAPI app
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── pipeline.py   # Pipeline endpoints
│   │   └── health.py     # Health check endpoints
│   ├── middleware/
│   │   ├── __init__.py
│   │   ├── rate_limit.py
│   │   └── cost_tracker.py
│   └── models.py         # Pydantic request/response models
├── ui/
│   └── app.py            # Streamlit UI
├── pipeline/             # Day 73 pipeline code (copied here)
│   ├── agents/
│   ├── evaluator.py
│   ├── security.py
│   └── pipeline.py
├── Dockerfile.api
├── Dockerfile.ui
├── docker-compose.yml
├── render.yaml           # Render deployment config
└── requirements.txt

# script_id: day_097_capstone_deploy_to_production/api_main
# api/main.py
import os
import logging
import time
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse

from api.routers.pipeline import router as pipeline_router
from api.routers.health import router as health_router

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Startup and shutdown events."""
    logger.info("Starting content pipeline API...")
    # Initialize any connections, warm up models, etc.
    yield
    logger.info("Shutting down content pipeline API...")


app = FastAPI(
    title="AI Content Pipeline",
    description="Multi-agent content creation with human review",
    version="1.0.0",
    lifespan=lifespan,
)

# CORS for the Streamlit frontend
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # In production: specify your UI domain
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Request logging middleware
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    logger.info(
        f"{request.method} {request.url.path} "
        f"status={response.status_code} duration={duration:.2f}s"
    )
    return response

app.include_router(pipeline_router, prefix="/api/v1")
app.include_router(health_router)

# script_id: day_097_capstone_deploy_to_production/api_models
# api/models.py
from pydantic import BaseModel, Field
from typing import Optional, Literal
from enum import Enum


class ContentType(str, Enum):
    blog_post = "blog_post"
    social_media = "social_media"
    email = "email"
    summary = "summary"


class PipelineRequest(BaseModel):
    topic: str = Field(..., min_length=5, max_length=500, description="Content topic")
    content_type: ContentType = ContentType.blog_post
    target_audience: str = Field(default="general audience", max_length=200)
    require_human_review: bool = Field(default=False, description="Skip human review for API")
    min_quality_threshold: float = Field(default=0.7, ge=0.0, le=1.0)

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "topic": "How AI agents are transforming software development",
                    "content_type": "blog_post",
                    "target_audience": "software engineers",
                    "require_human_review": False,
                    "min_quality_threshold": 0.7,
                }
            ]
        }
    }


class PipelineResponse(BaseModel):
    success: bool
    run_id: Optional[str] = None
    content: Optional[str] = None
    evaluation: Optional[dict] = None
    revision_count: Optional[int] = None
    total_tokens: Optional[int] = None
    estimated_cost_usd: Optional[float] = None
    error: Optional[str] = None

# script_id: day_097_capstone_deploy_to_production/api_router_pipeline
# api/routers/pipeline.py
import asyncio
import json
from fastapi import APIRouter, HTTPException, Depends
from fastapi.responses import StreamingResponse
from api.models import PipelineRequest, PipelineResponse
from api.middleware.rate_limit import check_rate_limit
from api.middleware.cost_tracker import track_cost
from pipeline.pipeline import ContentPipeline

router = APIRouter(tags=["pipeline"])


@router.post("/run", response_model=PipelineResponse)
async def run_pipeline(
    request: PipelineRequest,
    _rate_limit: None = Depends(check_rate_limit),
):
    """
    Run the content pipeline synchronously.
    Returns when the pipeline completes.
    """
    try:
        pipeline = ContentPipeline(
            min_quality_threshold=request.min_quality_threshold,
            max_revisions=2,
            require_human_review=request.require_human_review,
        )

        # Run in thread pool to avoid blocking the event loop
        loop = asyncio.get_event_loop()
        result = await loop.run_in_executor(
            None,
            lambda: pipeline.run(
                raw_topic=request.topic,
                content_type=request.content_type.value,
                target_audience=request.target_audience,
            ),
        )

        # Track and log the estimated cost, then attach it to the response
        cost = await track_cost(
            result.get("run_id"), result, request.content_type.value
        )
        result["estimated_cost_usd"] = cost

        return PipelineResponse(**result)

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.post("/stream")
async def stream_pipeline(request: PipelineRequest):
    """
    Run the content pipeline with streaming progress updates.
    Returns Server-Sent Events.
    """

    async def event_generator():
        """Generate SSE events for pipeline progress."""
        try:
            yield f"data: {json.dumps({'stage': 'started', 'topic': request.topic})}\n\n"

            pipeline = ContentPipeline(
                min_quality_threshold=request.min_quality_threshold,
                max_revisions=2,
                require_human_review=False,  # Can't pause for human-in-the-loop review (HITL) over a streaming connection
            )

            # Run in thread pool
            loop = asyncio.get_event_loop()
            result = await loop.run_in_executor(
                None,
                lambda: pipeline.run(
                    raw_topic=request.topic,
                    content_type=request.content_type.value,
                    target_audience=request.target_audience,
                ),
            )

            yield f"data: {json.dumps({'stage': 'complete', 'result': result})}\n\n"

        except Exception as e:
            yield f"data: {json.dumps({'stage': 'error', 'error': str(e)})}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )

# script_id: day_097_capstone_deploy_to_production/api_router_health
# api/routers/health.py
import os
from fastapi import APIRouter
from pydantic import BaseModel

router = APIRouter(tags=["health"])


class HealthResponse(BaseModel):
    status: str
    version: str
    openai_configured: bool


@router.get("/health", response_model=HealthResponse)
async def health_check():
    """Health check endpoint for deployment monitoring."""
    return HealthResponse(
        status="healthy",
        version="1.0.0",
        openai_configured=bool(os.environ.get("OPENAI_API_KEY")),
    )


@router.get("/")
async def root():
    return {"message": "AI Content Pipeline API", "docs": "/docs"}

# script_id: day_097_capstone_deploy_to_production/api_middleware_rate_limit
# api/middleware/rate_limit.py
import time
from collections import defaultdict
from fastapi import Request, HTTPException

# Simple in-memory rate limiter (use Redis in production for multi-instance)
request_counts = defaultdict(list)
RATE_LIMIT = 10       # requests per window
WINDOW_SECONDS = 60   # window size


async def check_rate_limit(request: Request):
    """
    Rate limit by IP address.
    In production: use slowapi or a Redis-backed rate limiter.
    """
    client_ip = request.client.host
    now = time.time()
    window_start = now - WINDOW_SECONDS

    # Clean up old requests
    request_counts[client_ip] = [t for t in request_counts[client_ip] if t > window_start]

    if len(request_counts[client_ip]) >= RATE_LIMIT:
        raise HTTPException(
            status_code=429,
            detail=f"Rate limit exceeded. Max {RATE_LIMIT} requests per {WINDOW_SECONDS}s.",
        )

    request_counts[client_ip].append(now)

# script_id: day_097_capstone_deploy_to_production/api_middleware_cost_tracker
# api/middleware/cost_tracker.py
import sqlite3
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

# Rough $/1K-token estimate (blended input+output, directional);
# verify current pricing at the provider (as of 2026-06).
COST_PER_1K_TOKENS = {
    "gpt-4o": 0.0025,
    "gpt-4o-mini": 0.00015,
}


def estimate_pipeline_cost(content_length: int) -> float:
    """
    Rough cost estimate for a pipeline run.
    A blog post pipeline uses ~5k-10k tokens across all agents.
    """
    estimated_tokens = max(5000, content_length * 3)
    return estimated_tokens * COST_PER_1K_TOKENS["gpt-4o-mini"] / 1000


def log_cost(run_id: str, estimated_cost: float, content_type: str):
    """Log pipeline cost to SQLite for monitoring."""
    try:
        conn = sqlite3.connect("./costs.db")
        conn.execute("""
            CREATE TABLE IF NOT EXISTS pipeline_costs (
                run_id TEXT, timestamp TEXT, estimated_cost REAL, content_type TEXT
            )
        """)
        conn.execute(
            "INSERT INTO pipeline_costs VALUES (?, ?, ?, ?)",
            (run_id, datetime.now().isoformat(), estimated_cost, content_type),
        )
        conn.commit()
        conn.close()
    except Exception as e:
        logger.error(f"Failed to log cost: {e}")


async def track_cost(run_id: str, result: dict, content_type: str):
    """Track and log the cost of a pipeline run."""
    content_length = len(result.get("content", ""))
    cost = estimate_pipeline_cost(content_length)
    log_cost(run_id, cost, content_type)
    logger.info(f"Pipeline {run_id}: estimated cost ${cost:.4f}")
    return cost

# script_id: day_097_capstone_deploy_to_production/streamlit_ui
# ui/app.py
import streamlit as st
import requests
import json

API_URL = "http://localhost:8000"  # or your deployed API URL

st.set_page_config(
    page_title="AI Content Pipeline",
    page_icon="✍️",
    layout="wide",
)

st.title("AI Content Pipeline")
st.caption("Multi-agent content creation with quality evaluation")

# Sidebar configuration
with st.sidebar:
    st.header("Settings")
    content_type = st.selectbox(
        "Content Type",
        ["blog_post", "social_media", "email", "summary"],
        index=0,
    )
    target_audience = st.text_input("Target Audience", value="software engineers")
    min_quality = st.slider("Minimum Quality Threshold", 0.5, 1.0, 0.7, 0.05)
    st.divider()
    st.caption("Each run costs approximately $0.01-0.05 in API calls.")

# Main input
topic = st.text_area(
    "What should we write about?",
    placeholder="e.g., How AI agents are transforming software development workflows",
    height=100,
)

if st.button("Generate Content", type="primary", disabled=not topic):
    if not topic.strip():
        st.error("Please enter a topic.")
    else:
        with st.spinner("Running content pipeline..."):
            try:
                response = requests.post(
                    f"{API_URL}/api/v1/run",
                    json={
                        "topic": topic,
                        "content_type": content_type,
                        "target_audience": target_audience,
                        "min_quality_threshold": min_quality,
                        "require_human_review": False,
                    },
                    timeout=120,
                )
                result = response.json()

                if result.get("success"):
                    # Quality metrics
                    col1, col2, col3 = st.columns(3)
                    eval_data = result.get("evaluation", {})
                    col1.metric("Quality Score", f"{eval_data.get('normalized_score', 0):.0%}")
                    col2.metric("Revisions", result.get("revision_count", 0))
                    col3.metric("Run ID", result.get("run_id", "N/A"))

                    # Content
                    st.divider()
                    st.subheader("Generated Content")
                    st.markdown(result["content"])

                    # Download button
                    st.download_button(
                        "Download as Markdown",
                        data=result["content"],
                        file_name=f"content_{result.get('run_id', 'output')}.md",
                        mime="text/markdown",
                    )

                    # Evaluation details
                    with st.expander("Evaluation Details"):
                        st.json(eval_data)

                else:
                    st.error(f"Pipeline failed: {result.get('error', 'Unknown error')}")

            except requests.exceptions.Timeout:
                st.error("Request timed out. The pipeline is taking longer than expected.")
            except Exception as e:
                st.error(f"Error connecting to API: {e}")

# Dockerfile.api
FROM python:3.11-slim

WORKDIR /app

# Install dependencies first (layer caching)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY api/ ./api/
COPY pipeline/ ./pipeline/

EXPOSE 8000

# Non-root user for security
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

CMD ["uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8000"]

# Dockerfile.ui
FROM python:3.11-slim

WORKDIR /app

COPY requirements-ui.txt .
RUN pip install --no-cache-dir -r requirements-ui.txt

COPY ui/ ./ui/

EXPOSE 8501

RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

CMD ["streamlit", "run", "ui/app.py", "--server.port=8501", "--server.address=0.0.0.0"]

# docker-compose.yml
services:
  api:
    build:
      context: .
      dockerfile: Dockerfile.api
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./data:/app/data  # Persist SQLite databases
    healthcheck:
      test: ["CMD", "python", "-c", "import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://localhost:8000/health').status==200 else 1)"]
      interval: 30s
      timeout: 10s
      retries: 3
    restart: unless-stopped

  ui:
    build:
      context: .
      dockerfile: Dockerfile.ui
    ports:
      - "8501:8501"
    environment:
      - API_URL=http://api:8000
    depends_on:
      api:
        condition: service_healthy
    restart: unless-stopped

# Build and run locally
docker-compose up --build

# API: http://localhost:8000
# UI: http://localhost:8501
# API docs: http://localhost:8000/docs

# render.yaml
services:
  - type: web
    name: content-pipeline-api
    env: docker
    dockerfilePath: ./Dockerfile.api
    plan: free
    healthCheckPath: /health
    envVars:
      - key: OPENAI_API_KEY
        sync: false  # Set this in Render dashboard, not here
      - key: PYTHON_ENV
        value: production

  - type: web
    name: content-pipeline-ui
    env: docker
    dockerfilePath: ./Dockerfile.ui
    plan: free
    envVars:
      - key: API_URL
        fromService:
          name: content-pipeline-api
          type: web
          property: host

# Install Railway CLI
npm install -g @railway/cli

railway login
railway init
railway up

# Set env vars
railway variables set OPENAI_API_KEY=sk-...

# script_id: day_097_capstone_deploy_to_production/api_router_health_detailed
# api/routers/health.py (expanded)
import os
import time
import psutil  # add `psutil` to requirements.txt — it's a third-party package
from fastapi import APIRouter
from pydantic import BaseModel
from typing import Optional

router = APIRouter(tags=["health"])

START_TIME = time.time()


class DetailedHealthResponse(BaseModel):
    status: str
    version: str
    uptime_seconds: float
    openai_configured: bool
    memory_mb: Optional[float] = None
    cpu_percent: Optional[float] = None


@router.get("/health/detailed", response_model=DetailedHealthResponse)
async def detailed_health():
    """Detailed health check with system metrics."""
    try:
        memory_mb = psutil.Process().memory_info().rss / 1024 / 1024
        cpu_percent = psutil.cpu_percent(interval=0.1)
    except Exception:
        memory_mb = None
        cpu_percent = None

    return DetailedHealthResponse(
        status="healthy",
        version="1.0.0",
        uptime_seconds=time.time() - START_TIME,
        openai_configured=bool(os.environ.get("OPENAI_API_KEY")),
        memory_mb=round(memory_mb, 1) if memory_mb else None,
        cpu_percent=cpu_percent,
    )