Operating Agents in Production

123456789101112
{
  "agentVersion": "2.4.1",
  "deployedAt": "2024-01-15T10:30:00Z",
  "components": {
    "codeVersion": "abc123",
    "promptVersion": "prompts-v2.4.0",
    "toolsVersion": "tools-v1.8.2",
    "guardrailsVersion": "safety-v3.1.0",
    "modelConfig": {
      "primary": "gpt-4-0125-preview",
      "fallback": "gpt-4-1106-preview",
      "embedding": "text-embedding-3-small"

123456789101112
import asyncio
import json
import boto3
from datetime import datetime

lambda_client = boto3.client('lambda')
kinesis_client = boto3.client('kinesis')

async def process_with_shadow(event, context):
    # Process with production agent
    production_start = datetime.utcnow()
    production_result = await invoke_production_agent(event)

123456789101112
import functools
import time
from dataclasses import dataclass
from typing import Optional
import boto3

@dataclass
class CostRecord:
    request_id: str
    user_id: str
    task_type: str
    model: str

123456789101112
import boto3
import json
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict

@dataclass
class AgentVersion:
    version_id: str
    model_config: Dict
    prompt_template: str
    tool_definitions: List[Dict]

123456789101112
import boto3
import json
from datetime import datetime, timedelta
from enum import Enum
from typing import Dict, List, Optional

class IncidentSeverity(Enum):
    SEV1 = 1  # Complete outage
    SEV2 = 2  # Major degradation
    SEV3 = 3  # Minor degradation
    SEV4 = 4  # Warning condition

123456789101112
import boto3
from decimal import Decimal
from datetime import datetime, timedelta
import json

class AgentCostTracker:
    """Real-time cost tracking for AI agents."""
    
    # Pricing per 1K tokens (as of 2024)
    MODEL_PRICING = {
        'claude-3-opus': {'input': 0.015, 'output': 0.075},
        'claude-3-sonnet': {'input': 0.003, 'output': 0.015},