Deploy to Production -- Hosting, Monitoring, and Scaling -- פריסה לפרודקשן

ארכיטקטורת Production

יש לכם סוכן AI שעובד. הוא רץ ב-terminal, עונה על שאלות, משתמש ב-tools, ומחזיר תשובות מצוינות. מעולה -- אבל זה כמו להכין אוכל מדהים רק במטבח שלכם. כדי לפתוח מסעדה, צריך מטבח תעשייתי, מלצרים, קופה, ניהול מלאי, וביטוח. הפרק הזה הוא על המעבר מ-demo ל-production.

ה-Production Agent Stack

כל סוכן production בנוי מאותם שכבות. הנה הארכיטקטורה המלאה:

┌─────────────────────────────────────────────────────────┐
│                    CLIENTS                               │
│  Web App  │  Mobile App  │  CLI  │  Slack Bot  │  API   │
└────────────────────┬────────────────────────────────────┘
                     │ HTTPS / WSS
            ┌────────▼─────────┐
            │  Load Balancer   │   (Nginx / CloudFlare / ALB)
            │  + Rate Limiter  │
            └────────┬─────────┘
                     │
            ┌────────▼─────────┐
            │   API Server     │   (FastAPI / Express)
            │  - Auth (JWT)    │   - REST endpoints
            │  - Validation    │   - SSE streaming
            │  - Rate limits   │   - WebSocket (optional)
            └────────┬─────────┘
                     │
     ┌───────────────┼───────────────┐
     │               │               │
┌────▼────┐   ┌──────▼──────┐  ┌─────▼──────┐
│  Queue  │   │ Agent       │  │ Background │
│ (Redis/ │   │ Runtime     │  │ Workers    │
│  SQS)   │   │ (LangGraph/ │  │ (Celery/   │
│         │   │  Pydantic/  │  │  Bull)     │
└────┬────┘   │  custom)    │  └────────────┘
     │        └──────┬──────┘
     │               │
     │    ┌──────────┼──────────┐
     │    │          │          │
     │ ┌──▼───┐ ┌───▼────┐ ┌──▼──────┐
     │ │Tools │ │  LLM   │ │ Memory  │
     │ │(MCP) │ │Provider│ │ (Redis/ │
     │ │      │ │(Claude/│ │  Postgres│
     │ └──────┘ │ GPT-4) │ │  /Pinec.)│
     │          └────────┘ └─────────┘
     │
┌────▼────────────────────┐
│  Observability Layer    │
│  - OpenTelemetry traces │
│  - Structured logs      │
│  - Metrics (Prometheus) │
│  - Alerts (PagerDuty)   │
└─────────────────────────┘

Stateless vs. Stateful Agents

ההחלטה הראשונה והקריטית ביותר: האם הסוכן שלכם stateless או stateful?

Containerization עם Docker

Docker הוא השפה המשותפת של deployment. לא משנה אם אתם פורסים ל-AWS, GCP, DigitalOcean, או הרספברי-פאי בבית -- אם יש לכם Dockerfile, אתם פורסים.

# Dockerfile -- Production Agent
FROM python:3.12-slim

# Security: non-root user
RUN groupadd -r agent && useradd -r -g agent agent

WORKDIR /app

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

# Application code
COPY . .

# Don't run as root
USER agent

# Health check
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

EXPOSE 8000

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

// Dockerfile -- TypeScript Agent
FROM node:20-slim

RUN groupadd -r agent && useradd -r -g agent agent

WORKDIR /app

COPY package*.json ./
RUN npm ci --only=production

COPY . .
RUN npm run build

USER agent

HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD curl -f http://localhost:3000/health || exit 1

EXPOSE 3000

CMD ["node", "dist/server.js"]

mkdir production-agent && cd production-agent
mkdir -p src config monitoring
cp ../your-agent/main.py src/  # או main.ts
touch Dockerfile docker-compose.yml .env.example
echo "ANTHROPIC_API_KEY=sk-ant-..." > .env.example
echo "PORT=8000" >> .env.example

אפשרויות אירוח

שלוש קטגוריות עיקריות: self-hosted (אתם מנהלים הכל), serverless (אתם כותבים קוד, הספק מנהל), ו-managed agent platforms (הספק מנהל גם את ה-agent runtime).

Self-hosted (VPS/VM)

שליטה מלאה, הכי זול ב-scale, הכי הרבה עבודה. מתאים לצוותים עם ידע DevOps.

Docker Compose לפריסות קטנות -- אם יש לכם עד כמה אלפי בקשות ביום, docker-compose על VPS בודד עם Nginx reverse proxy זה מספיק:

# docker-compose.yml
version: "3.9"
services:
  agent:
    build: .
    ports:
      - "8000:8000"
    env_file: .env
    restart: unless-stopped
    deploy:
      resources:
        limits:
          memory: 2G
    depends_on:
      - redis

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./certs:/etc/nginx/certs
    depends_on:
      - agent

volumes:
  redis_data:

Serverless

Zero ops, pay-per-use, cold starts. מושלם לסוכנים stateless עם תנועה לא אחידה.

Managed Agent Platforms

הכי פחות עבודה, הכי יקר. הספק מנהל את כל ה-infrastructure כולל ה-agent runtime.

השוואת עלויות -- טבלה מלאה

שימו לב: עלויות ה-hosting הן רק חלק קטן מהעלות הכוללת. ב-90% מהמקרים, עלויות ה-LLM API גדולות פי 10-100 מעלויות ה-hosting. סוכן שמשתמש ב-Claude Sonnet ושולח 1,000 tokens בממוצע לבקשה + מקבל 2,000 tokens -- עולה ~$0.012 לבקשה. ב-100K בקשות זה $1,200/חודש ב-LLM בלבד.

עיצוב API לשירותי סוכנים

הסוכן שלכם צריך דלת כניסה -- API שלקוחות (web app, mobile, CLI, Slack bot) יכולים לדבר איתו. הנה ה-API הסטנדרטי לשירות סוכן:

Endpoints

Python: FastAPI Production Server

# src/server.py -- Production Agent API
import os
import uuid
import time
import logging
from fastapi import FastAPI, HTTPException, Depends, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from anthropic import Anthropic

# --- Structured Logging ---
logging.basicConfig(
    format='{"time":"%(asctime)s","level":"%(levelname)s","msg":"%(message)s"}',
    level=logging.INFO
)
logger = logging.getLogger(__name__)

app = FastAPI(title="Production Agent API", version="1.0.0")
client = Anthropic()

# --- Request/Response Models ---
class AgentRequest(BaseModel):
    prompt: str
    model: str = "claude-sonnet-4-20250514"
    max_tokens: int = 4096
    user_id: str | None = None

class AgentResponse(BaseModel):
    id: str
    response: str
    model: str
    tokens_used: int
    cost_usd: float
    latency_ms: float

# --- Authentication ---
API_KEYS = set(os.getenv("API_KEYS", "").split(","))

async def verify_api_key(request: Request):
    api_key = request.headers.get("X-API-Key")
    if not api_key or api_key not in API_KEYS:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return api_key

# --- Rate Limiting (simple in-memory, use Redis in prod) ---
from collections import defaultdict
rate_limits: dict[str, list[float]] = defaultdict(list)
RATE_LIMIT = 60  # requests per minute

def check_rate_limit(api_key: str):
    now = time.time()
    window = [t for t in rate_limits[api_key] if now - t < 60]
    rate_limits[api_key] = window
    if len(window) >= RATE_LIMIT:
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
    rate_limits[api_key].append(now)

# --- Endpoints ---
@app.get("/health")
async def health():
    return {"status": "healthy", "timestamp": time.time()}

@app.post("/agent/run", response_model=AgentResponse)
async def run_agent(req: AgentRequest, api_key: str = Depends(verify_api_key)):
    check_rate_limit(api_key)
    request_id = str(uuid.uuid4())
    start = time.time()

    logger.info(f"request_id={request_id} user={req.user_id} model={req.model}")

    try:
        result = client.messages.create(
            model=req.model,
            max_tokens=req.max_tokens,
            messages=[{"role": "user", "content": req.prompt}]
        )

        latency = (time.time() - start) * 1000
        input_tokens = result.usage.input_tokens
        output_tokens = result.usage.output_tokens
        cost = (input_tokens * 0.003 + output_tokens * 0.015) / 1000  # Sonnet pricing

        logger.info(
            f"request_id={request_id} tokens_in={input_tokens} "
            f"tokens_out={output_tokens} cost=${cost:.4f} latency={latency:.0f}ms"
        )

        return AgentResponse(
            id=request_id,
            response=result.content[0].text,
            model=req.model,
            tokens_used=input_tokens + output_tokens,
            cost_usd=cost,
            latency_ms=latency
        )
    except Exception as e:
        logger.error(f"request_id={request_id} error={str(e)}")
        raise HTTPException(status_code=500, detail="Agent execution failed")

# --- SSE Streaming Endpoint ---
@app.post("/agent/stream")
async def stream_agent(req: AgentRequest, api_key: str = Depends(verify_api_key)):
    check_rate_limit(api_key)

    async def generate():
        with client.messages.stream(
            model=req.model,
            max_tokens=req.max_tokens,
            messages=[{"role": "user", "content": req.prompt}]
        ) as stream:
            for text in stream.text_stream:
                yield f"data: {text}\n\n"
        yield "data: [DONE]\n\n"

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

TypeScript: Express Production Server

// src/server.ts -- Production Agent API (TypeScript)
import express from 'express';
import Anthropic from '@anthropic-ai/sdk';
import { randomUUID } from 'crypto';

const app = express();
app.use(express.json());

const client = new Anthropic();
const API_KEYS = new Set((process.env.API_KEYS || '').split(','));

// --- Auth Middleware ---
function authMiddleware(req: express.Request, res: express.Response, next: express.NextFunction) {
  const apiKey = req.headers['x-api-key'] as string;
  if (!apiKey || !API_KEYS.has(apiKey)) {
    return res.status(401).json({ error: 'Invalid API key' });
  }
  next();
}

// --- Health Check ---
app.get('/health', (req, res) => {
  res.json({ status: 'healthy', timestamp: Date.now() });
});

// --- Run Agent ---
app.post('/agent/run', authMiddleware, async (req, res) => {
  const requestId = randomUUID();
  const start = Date.now();
  const { prompt, model = 'claude-sonnet-4-20250514', max_tokens = 4096 } = req.body;

  try {
    const result = await client.messages.create({
      model,
      max_tokens,
      messages: [{ role: 'user', content: prompt }],
    });

    const latency = Date.now() - start;
    const text = result.content[0].type === 'text' ? result.content[0].text : '';

    console.log(JSON.stringify({
      request_id: requestId,
      tokens_in: result.usage.input_tokens,
      tokens_out: result.usage.output_tokens,
      latency_ms: latency,
    }));

    res.json({
      id: requestId,
      response: text,
      model,
      tokens_used: result.usage.input_tokens + result.usage.output_tokens,
      latency_ms: latency,
    });
  } catch (error) {
    console.error(JSON.stringify({ request_id: requestId, error: String(error) }));
    res.status(500).json({ error: 'Agent execution failed' });
  }
});

// --- SSE Streaming ---
app.post('/agent/stream', authMiddleware, async (req, res) => {
  const { prompt, model = 'claude-sonnet-4-20250514', max_tokens = 4096 } = req.body;

  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');

  const stream = await client.messages.stream({
    model,
    max_tokens,
    messages: [{ role: 'user', content: prompt }],
  });

  for await (const event of stream) {
    if (event.type === 'content_block_delta' && event.delta.type === 'text_delta') {
      res.write(`data: ${event.delta.text}\n\n`);
    }
  }
  res.write('data: [DONE]\n\n');
  res.end();
});

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => console.log(`Agent API running on port ${PORT}`));

# Python
pip install fastapi uvicorn anthropic pydantic

# TypeScript
npm install express @anthropic-ai/sdk
npm install -D @types/express typescript

אסטרטגיות Scaling

הסוכן שלכם עובד מצוין עם 10 בקשות ביום. מה קורה כשיש 10,000? 100,000? מיליון? הדרך מ-10 ל-10M עוברת דרך ארבע אסטרטגיות מפתח.

1. Horizontal Scaling

במקום שרת אחד חזק, הרבה שרתים קטנים מאחורי load balancer. כל instance מטפל ב-X בקשות במקביל.

כמה concurrent sessions לכל instance? סוכן AI הוא I/O-bound (מחכה ל-LLM, ל-tools) ולא CPU-bound. לכן instance עם 4 vCPU יכול לטפל ב-50-200 sessions במקביל (תלוי ב-agent complexity). מעבר לזה, צריך instance נוסף.

2. Queue-Based Architecture

במקום שהAPI server מריץ את הסוכן ישירות, הוא שם את הבקשה בתור (queue) ו-worker נפרד מעבד אותה. זה מפריד בין קבלת בקשות לעיבוד:

# Queue-based architecture with Redis
# producer.py -- API Server pushes to queue
import redis
import json

r = redis.Redis()

async def enqueue_agent_request(request_id: str, prompt: str):
    job = json.dumps({"id": request_id, "prompt": prompt})
    r.lpush("agent:queue", job)
    return request_id

# consumer.py -- Worker pulls from queue
def process_queue():
    while True:
        _, job_data = r.brpop("agent:queue")
        job = json.loads(job_data)

        try:
            result = run_agent(job["prompt"])
            r.set(f"agent:result:{job['id']}", json.dumps(result), ex=3600)
        except Exception as e:
            # Send to Dead Letter Queue
            r.lpush("agent:dlq", json.dumps({**job, "error": str(e)}))

3. Caching

אל תקראו ל-LLM לשאלה שכבר ענו עליה. שלוש שכבות cache:

4. Model Routing

לא כל שאלה צריכה את המודל הכי חזק. שאלה פשוטה כמו "מה השעה?" לא צריכה Claude Sonnet ($3/$15 per M tokens). Haiku ($0.25/$1.25) מספיק. ה-pattern:

# Model routing -- save 60-80% on LLM costs
def route_to_model(prompt: str, complexity: str = "auto") -> str:
    if complexity == "auto":
        # Use a cheap model to classify complexity
        classification = client.messages.create(
            model="claude-haiku-3-5-20241022",
            max_tokens=10,
            messages=[{"role": "user", "content": f"Rate complexity 1-3: {prompt[:200]}"}]
        )
        score = int(classification.content[0].text.strip())
    else:
        score = {"low": 1, "medium": 2, "high": 3}[complexity]

    models = {
        1: "claude-haiku-3-5-20241022",    # Simple: $0.25/$1.25 per M tokens
        2: "claude-sonnet-4-20250514",     # Medium: $3/$15 per M tokens
        3: "claude-sonnet-4-20250514",     # Complex: best available
    }
    return models[score]

Durable Execution -- סוכנים ששורדים קריסות

סוכן AI לא כמו API רגיל שמחזיר תשובה בתוך 100ms. סוכן יכול לרוץ דקות, שעות, ואפילו ימים -- חושב, קורא ל-tools, מחכה לאישור אנושי, חוזר וחושב. מה קורה כש:

• השרת קורס באמצע ריצה של סוכן?
• עדכון deployment מכבה את ה-container?
• ה-LLM API חוזר עם timeout ב-step 7 מתוך 10?
• הסוכן מחכה לאישור אנושי שיגיע רק מחר?

בלי durable execution, כל העבודה אבדה. הסוכן מתחיל מאפס. עם durable execution, הסוכן ממשיך מהנקודה שבה עצר.

Durable execution חצה את ה-mainstream ב-2026: AWS שחרר Durable Functions, Cloudflare השיק Workflows GA, ו-Vercel השיק Workflow DevKit. זה כבר לא נישה -- זה standard.

פלטפורמות Durable Execution

דוגמה: LangGraph Checkpointing

# Durable execution with LangGraph checkpointing
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from typing import TypedDict

class AgentState(TypedDict):
    messages: list
    current_step: str
    tool_results: dict
    approved: bool

# Create graph with persistent checkpointing
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/agent_state"
)

graph = StateGraph(AgentState)

# Define nodes (each one is a checkpoint)
graph.add_node("research", research_step)
graph.add_node("analyze", analyze_step)
graph.add_node("wait_approval", human_approval_step)
graph.add_node("execute", execute_step)

# Define edges
graph.add_edge("research", "analyze")
graph.add_edge("analyze", "wait_approval")
graph.add_conditional_edges(
    "wait_approval",
    lambda state: "execute" if state["approved"] else END
)
graph.add_edge("execute", END)

graph.set_entry_point("research")
agent = graph.compile(checkpointer=checkpointer)

# Run -- if server crashes after "research", it resumes from "analyze"
config = {"configurable": {"thread_id": "task-123"}}
result = agent.invoke({"messages": [...], "current_step": "start"}, config)

# Later: resume after human approval
agent.update_state(config, {"approved": True})
result = agent.invoke(None, config)  # Continues from wait_approval

דוגמה: Temporal Workflow (TypeScript)

// Durable agent workflow with Temporal
import { proxyActivities, sleep } from '@temporalio/workflow';
import type { AgentActivities } from './activities';

const { research, analyze, sendApprovalRequest, executeAction } =
  proxyActivities({ startToCloseTimeout: '5 minutes' });

// This workflow survives server crashes, restarts, deployments
export async function agentWorkflow(taskId: string, prompt: string) {
  // Step 1: Research (checkpointed automatically)
  const researchResults = await research(prompt);

  // Step 2: Analyze (if server crashes here, resumes from this point)
  const analysis = await analyze(researchResults);

  // Step 3: Wait for human approval (can wait days!)
  await sendApprovalRequest(taskId, analysis);
  // Temporal persists state -- even if server restarts, it remembers
  // that we're waiting for approval

  // Step 4: Execute after approval signal
  const result = await executeAction(analysis);
  return result;
}

טיפול בשגיאות וחוסן

סוכני AI נכשלים בדרכים ייחודיות שלא קיימות בתוכנה רגילה. מעבר ל-bugs רגילים, יש:

Circuit Breaker Pattern

# Circuit breaker for agent tool calls
import time

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, reset_timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.last_failure_time = 0
        self.state = "CLOSED"  # CLOSED = normal, OPEN = blocking, HALF_OPEN = testing

    def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = "HALF_OPEN"
            else:
                raise Exception("Circuit breaker is OPEN -- service unavailable")

        try:
            result = func(*args, **kwargs)
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure_time = time.time()
            if self.failures >= self.failure_threshold:
                self.state = "OPEN"
            raise e

# Usage
llm_breaker = CircuitBreaker(failure_threshold=3, reset_timeout=30)

def call_llm_with_protection(prompt: str):
    try:
        return llm_breaker.call(client.messages.create, model="claude-sonnet-4-20250514", ...)
    except Exception:
        # Fallback: return cached response or graceful error
        return {"text": "I'm temporarily unable to process requests. Please try again shortly."}

Graceful Degradation

כשמשהו נכשל, הסוכן לא צריך לקרוס. הוא צריך להתדרדר בחינניות:

# Graceful degradation pattern
async def agent_with_graceful_degradation(prompt: str):
    results = {}

    # Try primary data source
    try:
        results["live_data"] = await fetch_live_analytics()
    except Exception:
        logger.warning("Live analytics unavailable, using cached data")
        results["live_data"] = await get_cached_analytics()  # 1-hour old

    # Try primary LLM
    try:
        response = await call_claude_sonnet(prompt, context=results)
    except Exception:
        logger.warning("Claude Sonnet unavailable, falling back to Haiku")
        try:
            response = await call_claude_haiku(prompt, context=results)
        except Exception:
            logger.error("All LLMs unavailable")
            response = "I'm experiencing technical difficulties. " \
                       "Here's what I can tell you from cached data: " \
                       f"{summarize_cached(results)}"

    return response

Monitoring ו-Observability

אתם לא יכולים לתקן מה שאתם לא רואים. Observability היא היכולת להבין מה קורה בתוך המערכת מבחוץ. לסוכני AI, זה קריטי במיוחד כי ההתנהגות שלהם לא דטרמיניסטית -- אותה שאלה יכולה לייצר תשובות שונות, tool calls שונים, ומספר steps שונה.

ארבעת עמודי ה-Observability

OpenTelemetry Instrumentation

# OpenTelemetry setup for agent monitoring
from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter

# Setup tracing
trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://otel-collector:4317"))
)
tracer = trace.get_tracer("agent-service")

# Setup metrics
metrics.set_meter_provider(MeterProvider())
meter = metrics.get_meter("agent-service")

# Custom metrics for AI agents
request_counter = meter.create_counter("agent.requests.total")
token_counter = meter.create_counter("agent.tokens.total")
cost_counter = meter.create_counter("agent.cost.usd")
latency_histogram = meter.create_histogram("agent.latency.ms")

# Instrumented agent call
@app.post("/agent/run")
async def run_agent(req: AgentRequest):
    with tracer.start_as_current_span("agent.run") as span:
        span.set_attribute("agent.model", req.model)
        span.set_attribute("agent.user_id", req.user_id or "anonymous")

        # Track LLM call
        with tracer.start_as_current_span("llm.call"):
            result = client.messages.create(...)

        # Record metrics
        request_counter.add(1, {"model": req.model, "status": "success"})
        token_counter.add(result.usage.input_tokens, {"type": "input"})
        token_counter.add(result.usage.output_tokens, {"type": "output"})
        cost = calculate_cost(result.usage)
        cost_counter.add(cost, {"model": req.model})
        latency_histogram.record(latency_ms, {"model": req.model})

LLM-Specific Monitoring

מעבר ל-metrics סטנדרטיים, סוכני AI צריכים monitoring ייחודי:

שליטה בעלויות Production

עלות LLM ב-production היא ההוצאה הכי גדולה שלכם -- גדולה פי 10-100 מעלות ה-hosting. סוכן שמשתמש ב-Claude Sonnet ב-100K בקשות/חודש יכול לעלות $1,000-$3,000 בחודש ב-LLM בלבד. אבל עם אופטימיזציה נכונה, אפשר לחסוך 60-90%.

מנופי אופטימיזציית עלות

תרחיש סיוט: Runaway Agent Loops

הכי מפחיד ב-production: באג שגורם לסוכן לרוץ בלולאה אינסופית. כל iteration קוראת ל-LLM, כל קריאה עולה כסף. סוכן עם באג שרץ 1,000 iterations ב-Sonnet יכול לעלות $50-200 בדקות.

# Cost protection: budget limits and runaway detection
class CostGuard:
    def __init__(self, max_cost_per_request: float = 1.0,
                 max_daily_cost: float = 100.0,
                 max_steps_per_request: int = 15):
        self.max_cost_per_request = max_cost_per_request
        self.max_daily_cost = max_daily_cost
        self.max_steps = max_steps_per_request
        self.daily_cost = 0.0
        self.request_cost = 0.0
        self.step_count = 0

    def check_budget(self, cost_increment: float):
        self.request_cost += cost_increment
        self.daily_cost += cost_increment
        self.step_count += 1

        if self.step_count > self.max_steps:
            raise BudgetExceeded(f"Max steps exceeded: {self.step_count}/{self.max_steps}")
        if self.request_cost > self.max_cost_per_request:
            raise BudgetExceeded(f"Request budget exceeded: ${self.request_cost:.2f}")
        if self.daily_cost > self.max_daily_cost:
            raise BudgetExceeded(f"Daily budget exceeded: ${self.daily_cost:.2f}")
            # Also: send PagerDuty alert!

class BudgetExceeded(Exception):
    pass

אבטחה ב-Production

סוכן AI ב-production הוא משטח תקיפה ענק. הוא מקבל input חופשי (natural language), ומריץ tools שיכולים לגשת ל-APIs, databases, ואפילו file systems. כל שכבת אבטחה שחסרה היא דלת פתוחה לתוקף.

Authentication ו-Authorization

# JWT Authentication for agent API
import jwt
from datetime import datetime, timedelta

SECRET_KEY = os.getenv("JWT_SECRET")

def create_token(user_id: str, role: str = "user") -> str:
    payload = {
        "user_id": user_id,
        "role": role,
        "exp": datetime.utcnow() + timedelta(hours=24),
        "iat": datetime.utcnow()
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

def verify_token(token: str) -> dict:
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

# Role-based access to agent tools
TOOL_PERMISSIONS = {
    "user":  ["search", "summarize", "translate"],
    "admin": ["search", "summarize", "translate", "database_query", "deploy"],
    "readonly": ["search", "summarize"]
}

def check_tool_access(role: str, tool_name: str):
    allowed = TOOL_PERMISSIONS.get(role, [])
    if tool_name not in allowed:
        raise HTTPException(
            status_code=403,
            detail=f"Role '{role}' cannot use tool '{tool_name}'"
        )

Input Validation -- נגד Prompt Injection

# Input validation layer
import re

def validate_agent_input(prompt: str) -> str:
    # 1. Length limit
    if len(prompt) > 10000:
        raise HTTPException(status_code=400, detail="Prompt too long (max 10K chars)")

    # 2. Detect common prompt injection patterns
    injection_patterns = [
        r"ignore\s+(all\s+)?previous\s+instructions",
        r"system\s*:\s*you\s+are\s+now",
        r"forget\s+(everything|all|your)\s+(instructions|rules)",
        r"override\s+safety",
        r"act\s+as\s+if\s+you\s+have\s+no\s+restrictions",
    ]
    for pattern in injection_patterns:
        if re.search(pattern, prompt, re.IGNORECASE):
            logger.warning(f"Prompt injection attempt detected: {pattern}")
            raise HTTPException(status_code=400, detail="Invalid input detected")

    # 3. Strip potentially dangerous content
    prompt = prompt.replace("\x00", "")  # null bytes

    return prompt

Output Sanitization

# Output sanitization -- strip PII before returning
def sanitize_output(response: str) -> str:
    # Remove Israeli ID numbers (9 digits)
    response = re.sub(r'\b\d{9}\b', '[REDACTED-ID]', response)

    # Remove credit card numbers
    response = re.sub(r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b', '[REDACTED-CC]', response)

    # Remove email addresses
    response = re.sub(r'[\w.-]+@[\w.-]+\.\w+', '[REDACTED-EMAIL]', response)

    # Remove Israeli phone numbers
    response = re.sub(r'\b0[2-9]\d{1}-?\d{7}\b', '[REDACTED-PHONE]', response)

    return response

Secrets Management

CI/CD לסוכנים

סוכן AI שונה מתוכנה רגילה בגלל שני ערוצי שינוי: קוד (כמו כל תוכנה) ו-prompts (system prompt, tool descriptions, guardrails). שניהם צריכים versioning, testing, ו-deployment pipeline.

Testing Pipeline לסוכנים

# Pipeline: unit tests --> integration tests --> eval suite --> staging --> production

# 1. Unit Tests (no LLM calls, fast)
# test_tools.py
def test_search_tool_formatting():
    result = format_search_results([{"title": "Test", "url": "https://test.com"}])
    assert "Test" in result
    assert "https://test.com" in result

def test_cost_guard_limits():
    guard = CostGuard(max_cost_per_request=1.0)
    guard.check_budget(0.5)  # OK
    with pytest.raises(BudgetExceeded):
        guard.check_budget(0.6)  # Exceeds $1.0

# 2. Integration Tests (real LLM calls, slower)
def test_agent_responds_to_simple_query():
    response = client.post("/agent/run", json={"prompt": "What is 2+2?"})
    assert response.status_code == 200
    assert "4" in response.json()["response"]

# 3. Eval Suite (quality checks)
def test_agent_eval_suite():
    test_cases = [
        {"prompt": "Summarize this article: ...", "expected_contains": ["key point"]},
        {"prompt": "Translate to Hebrew: Hello", "expected_contains": ["שלום"]},
    ]
    passed = 0
    for case in test_cases:
        response = run_agent(case["prompt"])
        if all(kw in response for kw in case["expected_contains"]):
            passed += 1
    pass_rate = passed / len(test_cases)
    assert pass_rate >= 0.8, f"Eval pass rate {pass_rate:.0%} below 80% threshold"

GitHub Actions CI/CD Pipeline

# .github/workflows/agent-deploy.yml
name: Agent CI/CD Pipeline

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: pip install -r requirements.txt -r requirements-dev.txt

      - name: Unit tests
        run: pytest tests/unit/ -v

      - name: Integration tests
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: pytest tests/integration/ -v --timeout=60

      - name: Eval suite
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: python tests/eval_suite.py --min-pass-rate 0.8

  deploy-staging:
    needs: test
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Build Docker image
        run: docker build -t agent:${{ github.sha }} .

      - name: Push to registry
        run: |
          docker tag agent:${{ github.sha }} registry.example.com/agent:${{ github.sha }}
          docker push registry.example.com/agent:${{ github.sha }}

      - name: Deploy to staging
        run: |
          # Deploy to staging environment
          kubectl set image deployment/agent-staging agent=registry.example.com/agent:${{ github.sha }}

  canary-deploy:
    needs: deploy-staging
    runs-on: ubuntu-latest
    steps:
      - name: Canary deploy (5% traffic)
        run: |
          # Route 5% of production traffic to new version
          kubectl apply -f k8s/canary-5-percent.yml

      - name: Monitor canary (10 minutes)
        run: |
          # Check error rate and latency for 10 minutes
          python scripts/monitor_canary.py --duration 600 --max-error-rate 0.05

      - name: Promote to 100% or rollback
        run: |
          if [ $? -eq 0 ]; then
            kubectl apply -f k8s/production-100-percent.yml
          else
            kubectl apply -f k8s/rollback.yml
            exit 1
          fi

Prompt Versioning

Prompts הם קוד -- הם צריכים version control. שינוי של מילה אחת ב-system prompt יכול לשנות את ההתנהגות של כל הסוכן.

# prompts/v2.3.0/system_prompt.txt
You are a customer support agent for ClickFlow.
Your role is to help users with technical issues.

Rules:
- Always be helpful and professional
- If you don't know the answer, say so
- Never share internal system details
- Escalate billing issues to humans

# Version history:
# v2.3.0 - Added escalation rule for billing
# v2.2.0 - Improved professional tone guidelines
# v2.1.0 - Added "don't share internal details" rule
# v2.0.0 - Major rewrite for new product features
# v1.0.0 - Initial version

Compliance ו-Audit

כשסוכן AI מטפל בלקוחות, מנתח נתונים, או מקבל החלטות -- צריך לתעד הכל. לא רק ל-debugging, אלא ל-compliance:

מה לתעד

GDPR ו-Right to Deletion

אם משתמש מבקש למחוק את הנתונים שלו (GDPR Article 17), אתם צריכים למחוק:

• כל ה-conversations שלו (inputs + outputs)
• כל ה-embeddings שנוצרו מהנתונים שלו (vector DB)
• כל ה-cached responses שקשורות אליו
• metadata שמזהה אותו (email, user ID ב-logs)

# GDPR deletion endpoint
@app.delete("/user/{user_id}/data")
async def delete_user_data(user_id: str, api_key: str = Depends(verify_admin)):
    # 1. Delete conversations from database
    db.execute("DELETE FROM conversations WHERE user_id = ?", [user_id])

    # 2. Delete from vector database
    vector_db.delete(filter={"user_id": user_id})

    # 3. Delete from cache
    for key in redis.scan_iter(f"cache:*:{user_id}:*"):
        redis.delete(key)

    # 4. Anonymize logs (don't delete -- anonymize for compliance)
    db.execute(
        "UPDATE audit_logs SET user_id = 'DELETED' WHERE user_id = ?",
        [user_id]
    )

    logger.info(f"GDPR deletion completed for user {user_id}")
    return {"status": "deleted", "user_id": user_id}

שגרת עבודה -- Production Agent

סוכן ב-production דורש פיקוח מתמשך. הנה השגרה שמבטיחה שהכל עובד:

צ'קליסט -- סיום פרק 19

• בניתי Dockerfile לסוכן עם non-root user, health check, ו-multi-stage build
• כתבתי docker-compose.yml עם agent service, Redis, ו-Nginx reverse proxy
• בניתי production API server (FastAPI או Express) עם /agent/run ו-/agent/stream
• הוספתי SSE streaming -- tokens מגיעים ב-real-time, לא תשובה מלאה אחרי 10 שניות
• הוספתי authentication (API keys או JWT) ו-rate limiting
• הגדרתי cost guards: max cost per request, max daily cost, max steps per request
• הוספתי circuit breaker לקריאות LLM ו-tool calls
• הוספתי structured logging (JSON) עם request_id, tokens, cost, latency
• הוספתי monitoring endpoint (/metrics) עם token usage, cost, ו-error rate
• הוספתי input validation נגד prompt injection ו-output sanitization לPII
• כתבתי unit tests (3+) ו-eval suite (5+ test cases) לסוכן
• יצרתי GitHub Actions workflow שמריץ tests על כל push
• מכיר/ה את durable execution ויודע/ת מתי צריך LangGraph checkpointing או Temporal
• מכיר/ה את The Scaling Playbook ויודע/ת באיזה שלב הסוכן שלי נמצא
• מבין/ה את GDPR implications ויש לי data retention policy לנתוני הסוכן