Workflow-006: Building AI Agents with Claude

Document Control

Workflow ID: 006
Version: 1.0
Status: Active
Complexity: High
Duration: 4-16 hours
Team Size: 1-3 developers

Overview

Building AI Agents with Claude involves creating autonomous systems that can perform complex tasks, make decisions, and interact with various tools and APIs. This workflow covers the complete process from agent design to deployment and monitoring.

┌─────────────────────────────────────────────────────────────────────┐
│                        AI AGENT BUILDING FLOW                        │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│   [1] DESIGN                 [2] IMPLEMENT                          │
│   ┌──────────────┐           ┌──────────────┐                      │
│   │ Agent Goals  │           │ Core Logic   │                      │
│   │ Capabilities │ ────────► │ Tool Integration│                   │
│   │ Boundaries   │           │ Memory System│                      │
│   └──────────────┘           └──────────────┘                      │
│          │                           │                               │
│          └─────────┬─────────────────┘                              │
│                    ▼                                                 │
│   [3] TRAIN                  [4] DEPLOY                             │
│   ┌──────────────┐           ┌──────────────┐                      │
│   │ Prompt Eng.  │           │ Production   │                      │
│   │ Testing      │ ────────► │ Monitoring   │                      │
│   │ Validation   │           │ Scaling      │                      │
│   └──────────────┘           └──────────────┘                      │
│          │                           │                               │
│          └───────────────────────────┴─── [5] OPTIMIZE              │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘

Prerequisites

Required SOPs

Environment Setup

Claude API access with high rate limits
MCP (Model Context Protocol) configured
Tool integration framework
Database for agent memory/state
Monitoring and logging infrastructure
Testing environment for agent validation

Conceptual Prerequisites

Understanding of autonomous systems
Knowledge of tool integration patterns
Experience with state management
Familiarity with API design
Background in prompt engineering

Phase 1: DESIGN - Agent Architecture

Objective

Design the agent's goals, capabilities, and operational boundaries.

Steps

1.1 Define Agent Purpose and Goals

bash

# Clarify the agent's primary mission
claude-code "think hard: What is this agent's primary purpose and how should it operate?"

# Agent specification template:
"""
Agent Specification: Customer Support Bot

Primary Goal:
Resolve customer inquiries efficiently while maintaining high satisfaction

Core Capabilities:
├─ Natural language understanding
├─ Knowledge base retrieval
├─ Ticket creation and management
├─ Escalation decision making
├─ Multi-channel communication
└─ Learning from interactions

Success Metrics:
├─ Resolution rate >80%
├─ Customer satisfaction >4/5
├─ Response time <2 minutes
├─ Escalation rate <20%
└─ Accuracy >95%

Operational Boundaries:
├─ Can't process payments
├─ Can't delete customer accounts
├─ Must escalate legal issues
├─ Limited to business hours
└─ Can't access personal data without consent
"""

1.2 Design Agent Architecture

bash

# Create the overall system architecture
claude-code "Design a modular architecture for this AI agent"

# Architecture components:
"""
AI Agent Architecture:

┌─────────────────────────────────────────────────┐
│                 Agent Controller                 │
├─────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │   Memory    │ │   Decision  │ │   Planning  │ │
│ │  Manager    │ │   Engine    │ │   Module    │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
├─────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │    Tool     │ │    Claude   │ │   Monitor   │ │
│ │ Interface   │ │   Client    │ │   Logger    │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
├─────────────────────────────────────────────────┤
│               External Integrations              │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │  Database   │ │     APIs    │ │   Message   │ │
│ │   Systems   │ │   Services  │ │   Queues    │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────┘

Core Components:

1. Agent Controller: Main orchestration logic
2. Memory Manager: Context and state management  
3. Decision Engine: Choice and action selection
4. Planning Module: Multi-step task planning
5. Tool Interface: External tool integration
6. Claude Client: AI model communication
7. Monitor Logger: Observability and debugging
"""

1.3 Define Tool and API Requirements

bash

# Identify required integrations and tools
claude-code "List all tools and APIs this agent will need to access"

# Tool requirements specification:
"""
Tool Integration Requirements:

Core Tools:
├─ Web Search: Real-time information retrieval
├─ Database Access: Customer data and history
├─ Email/SMS: Communication channels
├─ Knowledge Base: Documentation and FAQs
├─ Ticketing System: Issue tracking
└─ Calendar: Scheduling and appointments

API Integrations:
├─ CRM System (Salesforce/HubSpot)
├─ Payment Gateway (Stripe/PayPal)
├─ Communication (Slack/Discord)
├─ Analytics (Google Analytics)
├─ Monitoring (DataDog/New Relic)
└─ File Storage (AWS S3/Google Cloud)

Security Requirements:
├─ OAuth 2.0 authentication
├─ Rate limiting compliance
├─ Data encryption in transit
├─ PII handling protocols
├─ Audit logging
└─ Permission scoping
"""

1.4 Design Memory and State Management

bash

# Plan how the agent will store and retrieve information
claude-code "Design a memory system for this agent to maintain context and state"

# Memory architecture:
"""
Agent Memory Architecture:

1. Working Memory (Short-term):
   ├─ Current conversation context
   ├─ Active task state
   ├─ Recent tool outputs
   └─ Decision rationale

2. Episodic Memory (Medium-term):
   ├─ Conversation history (24-48h)
   ├─ Task completion records
   ├─ Error and recovery logs
   └─ Performance metrics

3. Semantic Memory (Long-term):
   ├─ Domain knowledge base
   ├─ Process procedures
   ├─ Best practice patterns
   └─ Learned optimizations

Storage Implementation:
├─ Redis: Working memory (fast access)
├─ PostgreSQL: Structured data storage
├─ Vector DB: Semantic search
└─ S3: Document and artifact storage

Memory Management Patterns:
├─ Automatic context pruning
├─ Relevance-based retrieval
├─ Periodic memory consolidation
└─ Privacy-aware data retention
"""

Quality Gate 1: Design Complete

[ ] Agent purpose clearly defined
[ ] Architecture designed and documented
[ ] Tool requirements identified
[ ] Memory system planned
[ ] Boundaries and constraints established

Phase 2: IMPLEMENT - Core Agent Development

Objective

Build the core agent infrastructure and integrate necessary tools.

Steps

2.1 Setup Project Structure

bash

# Create organized project structure
claude-code "Create a well-structured project for this AI agent"

# Project structure:
"""
ai-agent/
├── src/
│   ├── agent/
│   │   ├── controller.py         # Main agent orchestration
│   │   ├── memory.py             # Memory management
│   │   ├── planner.py            # Task planning
│   │   └── decision_engine.py    # Decision making logic
│   ├── tools/
│   │   ├── __init__.py
│   │   ├── base_tool.py          # Tool interface
│   │   ├── web_search.py         # Web search tool
│   │   ├── database.py           # Database operations
│   │   └── communication.py      # Email/SMS tools
│   ├── clients/
│   │   ├── claude_client.py      # Claude API integration
│   │   ├── api_clients.py        # External API clients
│   │   └── mcp_client.py         # MCP integration
│   ├── utils/
│   │   ├── logging.py            # Logging utilities
│   │   ├── config.py             # Configuration management
│   │   └── validators.py         # Input validation
│   └── main.py                   # Application entry point
├── tests/
│   ├── unit/
│   ├── integration/
│   └── e2e/
├── config/
│   ├── development.yaml
│   ├── production.yaml
│   └── agent_prompts.yaml
├── docker/
│   ├── Dockerfile
│   └── docker-compose.yml
├── docs/
│   ├── architecture.md
│   ├── api_reference.md
│   └── deployment.md
└── requirements.txt
"""

2.2 Implement Core Agent Controller

bash

# Build the main agent orchestration logic
claude-code "Implement the core agent controller class"

# Example implementation:
"""
# src/agent/controller.py
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
import asyncio
import logging

@dataclass
class AgentState:
    task_id: str
    context: Dict[str, Any]
    memory: Dict[str, Any]
    tools_used: List[str]
    status: str

class AgentController:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.memory_manager = MemoryManager(config)
        self.decision_engine = DecisionEngine()
        self.planner = TaskPlanner()
        self.claude_client = ClaudeClient(config['claude_api_key'])
        self.tools = self._initialize_tools()
        self.logger = logging.getLogger(__name__)
    
    async def process_request(self, request: str, context: Dict = None) -> str:
        \"\"\"Main request processing pipeline\"\"\"
        
        # Initialize task
        task_id = self._generate_task_id()
        state = AgentState(
            task_id=task_id,
            context=context or {},
            memory={},
            tools_used=[],
            status="processing"
        )
        
        try:
            # Step 1: Understand the request
            understanding = await self._understand_request(request, state)
            
            # Step 2: Plan the approach
            plan = await self._create_plan(understanding, state)
            
            # Step 3: Execute the plan
            result = await self._execute_plan(plan, state)
            
            # Step 4: Validate and respond
            response = await self._generate_response(result, state)
            
            return response
            
        except Exception as e:
            self.logger.error(f"Error processing request {task_id}: {e}")
            return await self._handle_error(e, state)
    
    async def _understand_request(self, request: str, state: AgentState) -> Dict:
        \"\"\"Use Claude to understand the request intent and requirements\"\"\"
        
        # Retrieve relevant context from memory
        relevant_context = await self.memory_manager.get_relevant_context(request)
        
        # Build understanding prompt
        prompt = f\"\"\"
        You are an AI agent analyzing a user request.
        
        Request: {request}
        
        Relevant Context: {relevant_context}
        
        Please analyze this request and provide:
        1. Intent classification
        2. Required capabilities
        3. Expected outcome
        4. Complexity assessment
        5. Required tools
        
        Respond in JSON format.
        \"\"\"
        
        response = await self.claude_client.complete(prompt)
        understanding = self._parse_json_response(response)
        
        # Store in working memory
        state.memory['understanding'] = understanding
        
        return understanding
"""

2.3 Implement Tool Integration Framework

bash

# Create a flexible tool integration system
claude-code "Create a tool integration framework with MCP support"

# Tool framework implementation:
"""
# src/tools/base_tool.py
from abc import ABC, abstractmethod
from typing import Dict, Any, List
from dataclasses import dataclass

@dataclass
class ToolResult:
    success: bool
    data: Any
    error_message: Optional[str] = None
    metadata: Dict[str, Any] = None

class BaseTool(ABC):
    \"\"\"Base class for all agent tools\"\"\"
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.name = self.__class__.__name__
        
    @abstractmethod
    async def execute(self, **kwargs) -> ToolResult:
        \"\"\"Execute the tool with given parameters\"\"\"
        pass
    
    @abstractmethod
    def get_schema(self) -> Dict[str, Any]:
        \"\"\"Return tool schema for LLM understanding\"\"\"
        pass
    
    def validate_params(self, params: Dict[str, Any]) -> bool:
        \"\"\"Validate input parameters\"\"\"
        schema = self.get_schema()
        # Implement validation logic
        return True

# src/tools/web_search.py
class WebSearchTool(BaseTool):
    \"\"\"Tool for web search functionality\"\"\"
    
    async def execute(self, query: str, max_results: int = 5) -> ToolResult:
        try:
            # Implement web search logic
            # Using search API (Google, Bing, etc.)
            results = await self._perform_search(query, max_results)
            
            return ToolResult(
                success=True,
                data=results,
                metadata={'query': query, 'result_count': len(results)}
            )
        except Exception as e:
            return ToolResult(
                success=False,
                data=None,
                error_message=str(e)
            )
    
    def get_schema(self) -> Dict[str, Any]:
        return {
            'name': 'web_search',
            'description': 'Search the web for information',
            'parameters': {
                'type': 'object',
                'properties': {
                    'query': {'type': 'string', 'description': 'Search query'},
                    'max_results': {'type': 'integer', 'default': 5}
                },
                'required': ['query']
            }
        }

# src/tools/database.py
class DatabaseTool(BaseTool):
    \"\"\"Tool for database operations\"\"\"
    
    async def execute(self, operation: str, **kwargs) -> ToolResult:
        try:
            if operation == 'select':
                result = await self._select_data(**kwargs)
            elif operation == 'insert':
                result = await self._insert_data(**kwargs)
            elif operation == 'update':
                result = await self._update_data(**kwargs)
            else:
                raise ValueError(f"Unsupported operation: {operation}")
            
            return ToolResult(success=True, data=result)
        except Exception as e:
            return ToolResult(
                success=False,
                data=None,
                error_message=str(e)
            )
"""

2.4 Implement Memory Management

bash

# Build sophisticated memory system
claude-code "Implement a comprehensive memory management system"

# Memory implementation:
"""
# src/agent/memory.py
import asyncio
from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
import json
import redis
import psycopg2
from sentence_transformers import SentenceTransformer

class MemoryManager:
    \"\"\"Manages agent memory across different time horizons\"\"\"
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.redis_client = redis.Redis(**config['redis'])
        self.pg_connection = psycopg2.connect(**config['postgres'])
        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
        
    async def store_working_memory(self, key: str, data: Any, ttl: int = 3600):
        \"\"\"Store short-term working memory in Redis\"\"\"
        serialized_data = json.dumps(data, default=str)
        self.redis_client.setex(key, ttl, serialized_data)
    
    async def get_working_memory(self, key: str) -> Optional[Any]:
        \"\"\"Retrieve from working memory\"\"\"
        data = self.redis_client.get(key)
        if data:
            return json.loads(data.decode('utf-8'))
        return None
    
    async def store_episodic_memory(self, episode: Dict[str, Any]):
        \"\"\"Store conversation episodes in PostgreSQL\"\"\"
        cursor = self.pg_connection.cursor()
        
        insert_query = \"\"\"
        INSERT INTO episodic_memory (
            timestamp, agent_id, user_id, request, response,
            tools_used, success, metadata
        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
        \"\"\"
        
        cursor.execute(insert_query, (
            datetime.now(),
            episode['agent_id'],
            episode['user_id'],
            episode['request'],
            episode['response'],
            json.dumps(episode['tools_used']),
            episode['success'],
            json.dumps(episode.get('metadata', {}))
        ))
        
        self.pg_connection.commit()
        cursor.close()
    
    async def get_relevant_context(self, query: str, limit: int = 5) -> List[Dict]:
        \"\"\"Retrieve relevant context using semantic search\"\"\"
        
        # Generate query embedding
        query_embedding = self.embedding_model.encode([query])[0]
        
        # Search similar episodes using vector similarity
        cursor = self.pg_connection.cursor()
        search_query = \"\"\"
        SELECT request, response, tools_used, metadata,
               embedding <-> %s as distance
        FROM episodic_memory 
        WHERE timestamp > %s
        ORDER BY embedding <-> %s
        LIMIT %s
        \"\"\"
        
        week_ago = datetime.now() - timedelta(days=7)
        cursor.execute(search_query, (
            query_embedding.tolist(),
            week_ago,
            query_embedding.tolist(),
            limit
        ))
        
        results = cursor.fetchall()
        cursor.close()
        
        return [
            {
                'request': row[0],
                'response': row[1],
                'tools_used': json.loads(row[2]),
                'metadata': json.loads(row[3]),
                'relevance_score': 1 - row[4]  # Convert distance to similarity
            }
            for row in results
        ]
    
    async def update_semantic_memory(self, knowledge: Dict[str, Any]):
        \"\"\"Update long-term semantic knowledge\"\"\"
        cursor = self.pg_connection.cursor()
        
        # Upsert knowledge into semantic memory
        upsert_query = \"\"\"
        INSERT INTO semantic_memory (topic, knowledge, embedding, updated_at)
        VALUES (%s, %s, %s, %s)
        ON CONFLICT (topic) 
        DO UPDATE SET 
            knowledge = EXCLUDED.knowledge,
            embedding = EXCLUDED.embedding,
            updated_at = EXCLUDED.updated_at
        \"\"\"
        
        topic = knowledge['topic']
        knowledge_text = knowledge['content']
        embedding = self.embedding_model.encode([knowledge_text])[0]
        
        cursor.execute(upsert_query, (
            topic,
            json.dumps(knowledge),
            embedding.tolist(),
            datetime.now()
        ))
        
        self.pg_connection.commit()
        cursor.close()
"""

Quality Gate 2: Implementation Complete

[ ] Project structure created
[ ] Core agent controller implemented
[ ] Tool integration framework built
[ ] Memory management system working
[ ] Basic functionality tested

Phase 3: TRAIN - Prompt Engineering and Testing

Objective

Optimize agent prompts and validate behavior through comprehensive testing.

Steps

3.1 Design Agent Prompts

bash

# Create effective system prompts for the agent
claude-code "Design comprehensive system prompts for this AI agent"

# System prompt template:
"""
# config/agent_prompts.yaml
system_prompt: |
  You are a Customer Support AI Agent with the following capabilities:
  
  CORE IDENTITY:
  - Professional, empathetic customer service representative
  - Expert in company products and policies
  - Focused on resolution and customer satisfaction
  
  CAPABILITIES:
  - Access customer information and order history
  - Search knowledge base for product information
  - Create and manage support tickets
  - Process returns and refunds (within policy limits)
  - Schedule callbacks and escalate complex issues
  
  TOOLS AVAILABLE:
  1. web_search(query): Search for general information
  2. database_query(table, filters): Query customer/order data
  3. create_ticket(description, priority): Create support ticket
  4. send_email(recipient, subject, body): Send email communication
  5. schedule_callback(customer_id, time): Schedule callback
  
  DECISION FRAMEWORK:
  1. Always prioritize customer satisfaction
  2. Follow company policies strictly
  3. Escalate when uncertain or for complex issues
  4. Document all interactions thoroughly
  5. Be transparent about limitations
  
  RESPONSE FORMAT:
  - Start with empathy and acknowledgment
  - Provide clear, actionable solutions
  - Include next steps and timelines
  - End with satisfaction check
  
  CONSTRAINTS:
  - Cannot process payments directly
  - Cannot delete customer accounts
  - Must escalate legal/compliance issues
  - Cannot offer discounts >20% without approval
  - Must maintain customer privacy

task_planning_prompt: |
  Given a customer request, create a step-by-step plan to resolve it:
  
  1. ANALYZE the request for:
     - Customer intent and emotion
     - Required information
     - Complexity level
     - Potential solutions
  
  2. PLAN the resolution:
     - Information gathering steps
     - Tool usage sequence
     - Decision points
     - Escalation criteria
  
  3. EXECUTE systematically:
     - Follow plan step by step
     - Validate each step's success
     - Adjust plan if needed
     - Document outcomes
  
  4. RESPOND with:
     - Clear solution or status
     - Next actions for customer
     - Follow-up timeline
     - Satisfaction confirmation

error_handling_prompt: |
  When encountering errors or unexpected situations:
  
  1. ACKNOWLEDGE the issue to the customer
  2. EXPLAIN what went wrong (if appropriate)
  3. PROVIDE alternative solutions or next steps
  4. ESCALATE if unable to resolve
  5. FOLLOW UP to ensure resolution
  
  Never claim capabilities you don't have.
  Always be honest about limitations.
  Prioritize customer experience over perfect solutions.
"""

3.2 Implement Testing Framework

bash

# Create comprehensive testing for agent behavior
claude-code "Create a testing framework for validating agent behavior"

# Testing framework:
"""
# tests/agent_test_framework.py
import pytest
import asyncio
from typing import Dict, List, Any
from unittest.mock import Mock, patch
from src.agent.controller import AgentController

class AgentTestFramework:
    \"\"\"Framework for testing agent behavior and responses\"\"\"
    
    def __init__(self, agent_config: Dict[str, Any]):
        self.agent = AgentController(agent_config)
        self.test_scenarios = self._load_test_scenarios()
    
    async def run_behavioral_tests(self) -> Dict[str, Any]:
        \"\"\"Run comprehensive behavioral tests\"\"\"
        results = {}
        
        for scenario in self.test_scenarios:
            result = await self._test_scenario(scenario)
            results[scenario['name']] = result
        
        return results
    
    async def _test_scenario(self, scenario: Dict[str, Any]) -> Dict[str, Any]:
        \"\"\"Test a single scenario\"\"\"
        try:
            # Setup test environment
            with patch.multiple(
                self.agent,
                tools=scenario.get('mock_tools', {}),
                memory_manager=Mock()
            ):
                # Execute test
                response = await self.agent.process_request(
                    scenario['input'],
                    scenario.get('context', {})
                )
                
                # Validate response
                validation_result = self._validate_response(
                    response, scenario['expected']
                )
                
                return {
                    'success': validation_result['passed'],
                    'response': response,
                    'validation': validation_result,
                    'metrics': self._calculate_metrics(response, scenario)
                }
                
        except Exception as e:
            return {
                'success': False,
                'error': str(e),
                'response': None
            }
    
    def _load_test_scenarios(self) -> List[Dict[str, Any]]:
        \"\"\"Load test scenarios from configuration\"\"\"
        return [
            {
                'name': 'order_status_inquiry',
                'input': 'What is the status of my order #12345?',
                'context': {'customer_id': 'CUST001'},
                'expected': {
                    'contains_order_status': True,
                    'uses_database_tool': True,
                    'polite_tone': True,
                    'includes_next_steps': True
                },
                'mock_tools': {
                    'database_query': Mock(return_value={
                        'order_id': '12345',
                        'status': 'shipped',
                        'tracking': 'TRACK123'
                    })
                }
            },
            {
                'name': 'complex_technical_issue',
                'input': 'My device keeps crashing when I use feature X',
                'expected': {
                    'escalation_triggered': True,
                    'diagnostic_questions': True,
                    'ticket_created': True
                }
            },
            {
                'name': 'refund_request',
                'input': 'I want to return this product and get a refund',
                'expected': {
                    'policy_check': True,
                    'return_process_explained': True,
                    'timeline_provided': True
                }
            }
        ]

# tests/test_agent_scenarios.py
class TestAgentScenarios:
    \"\"\"Specific test cases for agent scenarios\"\"\"
    
    @pytest.mark.asyncio
    async def test_order_status_happy_path(self):
        \"\"\"Test successful order status inquiry\"\"\"
        agent = AgentController(test_config)
        
        # Mock successful database response
        with patch.object(agent.tools['database'], 'execute') as mock_db:
            mock_db.return_value = ToolResult(
                success=True,
                data={'order_id': '12345', 'status': 'shipped'}
            )
            
            response = await agent.process_request(
                "What's the status of order 12345?",
                {'customer_id': 'CUST001'}
            )
            
            # Validate response quality
            assert 'shipped' in response.lower()
            assert 'order 12345' in response
            assert mock_db.called
    
    @pytest.mark.asyncio
    async def test_error_handling(self):
        \"\"\"Test agent behavior when tools fail\"\"\"
        agent = AgentController(test_config)
        
        # Mock tool failure
        with patch.object(agent.tools['database'], 'execute') as mock_db:
            mock_db.return_value = ToolResult(
                success=False,
                error_message="Database connection failed"
            )
            
            response = await agent.process_request(
                "What's my order status?",
                {'customer_id': 'CUST001'}
            )
            
            # Should handle error gracefully
            assert 'unable to access' in response.lower() or 'technical issue' in response.lower()
            assert 'apologize' in response.lower() or 'sorry' in response.lower()
    
    @pytest.mark.asyncio
    async def test_escalation_trigger(self):
        \"\"\"Test that complex issues trigger escalation\"\"\"
        agent = AgentController(test_config)
        
        response = await agent.process_request(
            "I think there's a security vulnerability in your system",
            {'customer_id': 'CUST001'}
        )
        
        # Should trigger escalation
        assert 'escalate' in response.lower() or 'specialist' in response.lower()
        # Should not attempt to resolve directly
"""

3.3 Performance and Quality Testing

bash

# Test agent performance and consistency
claude-code "Create performance and quality tests for the agent"

# Performance testing:
"""
# tests/test_performance.py
import time
import asyncio
import pytest
from concurrent.futures import ThreadPoolExecutor
from src.agent.controller import AgentController

class TestAgentPerformance:
    \"\"\"Performance and load testing for the agent\"\"\"
    
    @pytest.mark.asyncio
    async def test_response_time(self):
        \"\"\"Test agent response time meets SLA\"\"\"
        agent = AgentController(test_config)
        
        start_time = time.time()
        response = await agent.process_request(
            "What are your business hours?",
            {}
        )
        end_time = time.time()
        
        response_time = end_time - start_time
        
        # Should respond within 5 seconds
        assert response_time < 5.0, f"Response took {response_time:.2f}s"
        assert len(response) > 0
    
    @pytest.mark.asyncio
    async def test_concurrent_requests(self):
        \"\"\"Test handling multiple concurrent requests\"\"\"
        agent = AgentController(test_config)
        
        # Create multiple concurrent requests
        requests = [
            agent.process_request(f"Test request {i}", {'id': i})
            for i in range(10)
        ]
        
        start_time = time.time()
        responses = await asyncio.gather(*requests)
        end_time = time.time()
        
        # All requests should succeed
        assert len(responses) == 10
        assert all(len(r) > 0 for r in responses)
        
        # Should handle concurrency efficiently
        total_time = end_time - start_time
        assert total_time < 30.0  # 10 requests in under 30 seconds
    
    @pytest.mark.asyncio
    async def test_memory_usage(self):
        \"\"\"Test agent memory usage doesn't grow unbounded\"\"\"
        import psutil
        import os
        
        agent = AgentController(test_config)
        process = psutil.Process(os.getpid())
        
        initial_memory = process.memory_info().rss
        
        # Process many requests
        for i in range(100):
            await agent.process_request(
                f"Simple request {i}",
                {'session_id': f'session_{i}'}
            )
        
        final_memory = process.memory_info().rss
        memory_growth = final_memory - initial_memory
        
        # Memory growth should be reasonable (< 100MB)
        assert memory_growth < 100 * 1024 * 1024

class TestAgentQuality:
    \"\"\"Quality and consistency testing\"\"\"
    
    @pytest.mark.asyncio
    async def test_response_consistency(self):
        \"\"\"Test that similar requests get consistent responses\"\"\"
        agent = AgentController(test_config)
        
        # Ask the same question multiple times
        responses = []
        for _ in range(5):
            response = await agent.process_request(
                "What are your business hours?",
                {}
            )
            responses.append(response)
        
        # Responses should be similar
        # (implement similarity scoring logic)
        similarity_scores = []
        for i in range(len(responses)):
            for j in range(i+1, len(responses)):
                score = self._calculate_similarity(responses[i], responses[j])
                similarity_scores.append(score)
        
        avg_similarity = sum(similarity_scores) / len(similarity_scores)
        assert avg_similarity > 0.8  # 80% similarity threshold
    
    def _calculate_similarity(self, text1: str, text2: str) -> float:
        \"\"\"Calculate semantic similarity between texts\"\"\"
        # Implement using sentence transformers or similar
        from sentence_transformers import SentenceTransformer
        model = SentenceTransformer('all-MiniLM-L6-v2')
        
        embeddings = model.encode([text1, text2])
        similarity = cosine_similarity([embeddings[0]], [embeddings[1]])[0][0]
        
        return similarity
"""

Quality Gate 3: Training Complete

[ ] System prompts optimized
[ ] Behavioral tests passing
[ ] Performance benchmarks met
[ ] Quality thresholds achieved
[ ] Error handling validated

Phase 4: DEPLOY - Production Deployment

Objective

Deploy the agent to production with monitoring and scaling capabilities.

Steps

4.1 Production Environment Setup

bash

# Prepare production deployment
claude-code "Create production deployment configuration for this AI agent"

# Production deployment setup:
"""
# docker/Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    postgresql-client \
    redis-tools \
    curl \
    && rm -rf /var/lib/apt/lists/*

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY src/ ./src/
COPY config/ ./config/

# Create non-root user
RUN useradd --create-home --shell /bin/bash agent
USER agent

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

EXPOSE 8000

CMD ["python", "-m", "src.main"]

# docker/docker-compose.yml
version: '3.8'

services:
  agent:
    build: .
    ports:
      - "8000:8000"
    environment:
      - CLAUDE_API_KEY=${CLAUDE_API_KEY}
      - DATABASE_URL=${DATABASE_URL}
      - REDIS_URL=${REDIS_URL}
      - LOG_LEVEL=INFO
    depends_on:
      - postgres
      - redis
    restart: unless-stopped
    
  postgres:
    image: postgres:15
    environment:
      - POSTGRES_DB=agent_memory
      - POSTGRES_USER=agent
      - POSTGRES_PASSWORD=${DB_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./sql/init.sql:/docker-entrypoint-initdb.d/init.sql
    restart: unless-stopped
    
  redis:
    image: redis:7-alpine
    restart: unless-stopped
    
  monitoring:
    image: prom/prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
    restart: unless-stopped

volumes:
  postgres_data:
"""

4.2 API Interface and Scaling

bash

# Create scalable API interface
claude-code "Create a REST API interface for the agent with proper scaling"

# API implementation:
"""
# src/api/server.py
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
import asyncio
import logging
from typing import Optional, Dict, Any
import uvicorn

from src.agent.controller import AgentController
from src.utils.config import load_config
from src.utils.auth import verify_token

app = FastAPI(
    title="AI Customer Support Agent",
    version="1.0.0",
    description="Autonomous customer support agent API"
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Configure for production
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

security = HTTPBearer()

# Request/Response models
class ChatRequest(BaseModel):
    message: str
    customer_id: Optional[str] = None
    session_id: Optional[str] = None
    context: Optional[Dict[str, Any]] = None

class ChatResponse(BaseModel):
    response: str
    session_id: str
    metadata: Dict[str, Any]
    success: bool

class HealthResponse(BaseModel):
    status: str
    version: str
    dependencies: Dict[str, str]

# Global agent instance
agent_controller = None

@app.on_event("startup")
async def startup_event():
    global agent_controller
    config = load_config()
    agent_controller = AgentController(config)
    logging.info("AI Agent started successfully")

@app.get("/health", response_model=HealthResponse)
async def health_check():
    \"\"\"Health check endpoint\"\"\"
    return HealthResponse(
        status="healthy",
        version="1.0.0",
        dependencies={
            "database": "connected",
            "redis": "connected",
            "claude_api": "available"
        }
    )

@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(
    request: ChatRequest,
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    \"\"\"Main chat endpoint\"\"\"
    try:
        # Verify authentication
        await verify_token(credentials.credentials)
        
        # Process request
        response = await agent_controller.process_request(
            request.message,
            {
                'customer_id': request.customer_id,
                'session_id': request.session_id,
                **request.context or {}
            }
        )
        
        return ChatResponse(
            response=response,
            session_id=request.session_id or "new_session",
            metadata={"processing_time": "1.2s", "tools_used": []},
            success=True
        )
        
    except Exception as e:
        logging.error(f"Error processing chat request: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/feedback")
async def feedback_endpoint(
    session_id: str,
    rating: int,
    comment: Optional[str] = None,
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    \"\"\"Collect user feedback\"\"\"
    try:
        await verify_token(credentials.credentials)
        
        # Store feedback for agent improvement
        await agent_controller.memory_manager.store_feedback({
            'session_id': session_id,
            'rating': rating,
            'comment': comment,
            'timestamp': datetime.now()
        })
        
        return {"status": "feedback_received"}
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    uvicorn.run(
        "src.api.server:app",
        host="0.0.0.0",
        port=8000,
        workers=4,
        log_level="info"
    )
"""

4.3 Monitoring and Observability

bash

# Implement comprehensive monitoring
claude-code "Create monitoring and observability for the production agent"

# Monitoring implementation:
"""
# src/utils/monitoring.py
import time
import logging
from typing import Dict, Any
from prometheus_client import Counter, Histogram, Gauge, start_http_server
from datadog import initialize, statsd
import sentry_sdk
from sentry_sdk.integrations.asyncio import AsyncioIntegration

class AgentMonitoring:
    \"\"\"Comprehensive monitoring for AI agent\"\"\"
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self._setup_prometheus_metrics()
        self._setup_datadog()
        self._setup_sentry()
        
    def _setup_prometheus_metrics(self):
        \"\"\"Setup Prometheus metrics\"\"\"
        self.request_count = Counter(
            'agent_requests_total',
            'Total agent requests',
            ['endpoint', 'status', 'customer_type']
        )
        
        self.request_duration = Histogram(
            'agent_request_duration_seconds',
            'Agent request duration',
            ['endpoint', 'complexity']
        )
        
        self.tool_usage = Counter(
            'agent_tool_usage_total',
            'Tool usage count',
            ['tool_name', 'success']
        )
        
        self.active_sessions = Gauge(
            'agent_active_sessions',
            'Number of active chat sessions'
        )
        
        self.error_rate = Counter(
            'agent_errors_total',
            'Total errors',
            ['error_type', 'severity']
        )
    
    def _setup_datadog(self):
        \"\"\"Setup DataDog integration\"\"\"
        if 'datadog' in self.config:
            initialize(**self.config['datadog'])
    
    def _setup_sentry(self):
        \"\"\"Setup Sentry error tracking\"\"\"
        if 'sentry_dsn' in self.config:
            sentry_sdk.init(
                dsn=self.config['sentry_dsn'],
                integrations=[AsyncioIntegration()],
                traces_sample_rate=0.1,
                environment=self.config.get('environment', 'production')
            )
    
    def track_request(self, endpoint: str, customer_type: str = 'standard'):
        \"\"\"Context manager for tracking requests\"\"\"
        class RequestTracker:
            def __init__(self, monitoring):
                self.monitoring = monitoring
                self.endpoint = endpoint
                self.customer_type = customer_type
                self.start_time = None
                
            def __enter__(self):
                self.start_time = time.time()
                self.monitoring.active_sessions.inc()
                return self
                
            def __exit__(self, exc_type, exc_val, exc_tb):
                duration = time.time() - self.start_time
                
                if exc_type is None:
                    status = 'success'
                    self.monitoring.request_count.labels(
                        endpoint=self.endpoint,
                        status=status,
                        customer_type=self.customer_type
                    ).inc()
                else:
                    status = 'error'
                    self.monitoring.error_rate.labels(
                        error_type=exc_type.__name__,
                        severity='high'
                    ).inc()
                
                self.monitoring.request_duration.labels(
                    endpoint=self.endpoint,
                    complexity='standard'
                ).observe(duration)
                
                self.monitoring.active_sessions.dec()
                
                # Send to DataDog
                if hasattr(self.monitoring, 'statsd'):
                    statsd.histogram('agent.request.duration', duration)
                    statsd.increment(f'agent.request.{status}')
        
        return RequestTracker(self)
    
    def track_tool_usage(self, tool_name: str, success: bool):
        \"\"\"Track tool usage metrics\"\"\"
        self.tool_usage.labels(
            tool_name=tool_name,
            success='success' if success else 'failure'
        ).inc()
        
        if hasattr(self, 'statsd'):
            statsd.increment(f'agent.tool.{tool_name}.{"success" if success else "failure"}')
    
    def log_performance_metrics(self, metrics: Dict[str, float]):
        \"\"\"Log performance metrics\"\"\"
        for metric_name, value in metrics.items():
            logging.info(f"Performance metric {metric_name}: {value}")
            
            if hasattr(self, 'statsd'):
                statsd.histogram(f'agent.performance.{metric_name}', value)

# Start Prometheus metrics server
start_http_server(8001)

# src/utils/logging.py
import logging
import json
from datetime import datetime
from typing import Dict, Any

class StructuredLogger:
    \"\"\"Structured logging for agent operations\"\"\"
    
    def __init__(self, name: str, level: str = "INFO"):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(getattr(logging, level))
        
        # JSON formatter for structured logs
        formatter = logging.Formatter(
            '{"timestamp": "%(asctime)s", "level": "%(levelname)s", "module": "%(name)s", "message": %(message)s}'
        )
        
        handler = logging.StreamHandler()
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
    
    def log_request(self, request_id: str, message: str, context: Dict[str, Any]):
        \"\"\"Log agent request with context\"\"\"
        log_data = {
            "request_id": request_id,
            "type": "agent_request",
            "message": message,
            "context": context,
            "timestamp": datetime.now().isoformat()
        }
        self.logger.info(json.dumps(log_data))
    
    def log_tool_execution(self, tool_name: str, success: bool, duration: float, metadata: Dict[str, Any]):
        \"\"\"Log tool execution\"\"\"
        log_data = {
            "type": "tool_execution",
            "tool_name": tool_name,
            "success": success,
            "duration_ms": duration * 1000,
            "metadata": metadata,
            "timestamp": datetime.now().isoformat()
        }
        self.logger.info(json.dumps(log_data))
    
    def log_error(self, error_type: str, error_message: str, context: Dict[str, Any]):
        \"\"\"Log errors with context\"\"\"
        log_data = {
            "type": "error",
            "error_type": error_type,
            "error_message": error_message,
            "context": context,
            "timestamp": datetime.now().isoformat()
        }
        self.logger.error(json.dumps(log_data))
"""

Quality Gate 4: Deployment Complete

[ ] Production environment configured
[ ] API interface deployed
[ ] Monitoring systems active
[ ] Security measures implemented
[ ] Performance benchmarks validated

Phase 5: OPTIMIZE - Performance and Learning

Objective

Continuously improve agent performance through monitoring and optimization.

Steps

5.1 Performance Analysis and Optimization

bash

# Analyze production performance data
claude-code "Create performance analysis and optimization system for the agent"

# Performance optimization:
"""
# src/optimization/performance_optimizer.py
import asyncio
from typing import Dict, List, Any
from datetime import datetime, timedelta
import pandas as pd
from sklearn.cluster import KMeans
import numpy as np

class PerformanceOptimizer:
    \"\"\"Analyzes and optimizes agent performance\"\"\"
    
    def __init__(self, monitoring_data_source):
        self.monitoring_data = monitoring_data_source
        self.optimization_rules = []
        
    async def analyze_performance_trends(self, days: int = 7) -> Dict[str, Any]:
        \"\"\"Analyze performance trends over time\"\"\"
        
        # Fetch performance data
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days)
        
        performance_data = await self._fetch_performance_data(start_date, end_date)
        
        analysis = {
            'response_time_trends': self._analyze_response_times(performance_data),
            'success_rate_trends': self._analyze_success_rates(performance_data),
            'tool_usage_patterns': self._analyze_tool_usage(performance_data),
            'error_patterns': self._analyze_errors(performance_data),
            'user_satisfaction_trends': self._analyze_satisfaction(performance_data)
        }
        
        return analysis
    
    def _analyze_response_times(self, data: pd.DataFrame) -> Dict[str, Any]:
        \"\"\"Analyze response time patterns\"\"\"
        
        # Group by hour to find peak times
        hourly_avg = data.groupby(data['timestamp'].dt.hour)['response_time'].mean()
        
        # Identify slow response patterns
        slow_requests = data[data['response_time'] > data['response_time'].quantile(0.95)]
        
        return {
            'hourly_averages': hourly_avg.to_dict(),
            'peak_hours': hourly_avg.nlargest(3).index.tolist(),
            'slow_request_patterns': self._identify_slow_patterns(slow_requests),
            'recommendations': self._generate_performance_recommendations(hourly_avg, slow_requests)
        }
    
    def _identify_slow_patterns(self, slow_requests: pd.DataFrame) -> Dict[str, Any]:
        \"\"\"Identify patterns in slow requests\"\"\"
        
        patterns = {
            'common_tools': slow_requests['tools_used'].value_counts().head().to_dict(),
            'request_types': slow_requests['request_type'].value_counts().head().to_dict(),
            'complexity_correlation': slow_requests[['complexity_score', 'response_time']].corr().iloc[0, 1]
        }
        
        return patterns
    
    async def optimize_prompt_performance(self) -> Dict[str, Any]:
        \"\"\"Optimize prompts based on performance data\"\"\"
        
        # Analyze prompt effectiveness
        prompt_performance = await self._analyze_prompt_effectiveness()
        
        optimization_results = {}
        
        for prompt_type, performance in prompt_performance.items():
            if performance['avg_success_rate'] < 0.85:
                # Prompt needs optimization
                optimized_prompt = await self._optimize_prompt(prompt_type, performance)
                optimization_results[prompt_type] = {
                    'original_performance': performance,
                    'optimized_prompt': optimized_prompt,
                    'expected_improvement': self._estimate_improvement(performance)
                }
        
        return optimization_results
    
    async def _optimize_prompt(self, prompt_type: str, performance_data: Dict[str, Any]) -> str:
        \"\"\"Use Claude to optimize underperforming prompts\"\"\"
        
        claude_optimization_prompt = f\"\"\"
        Analyze and optimize this AI agent prompt based on performance data:
        
        Current Prompt Type: {prompt_type}
        Performance Issues:
        - Success Rate: {performance_data['avg_success_rate']:.2%}
        - Average Response Time: {performance_data['avg_response_time']:.2f}s
        - Common Failure Patterns: {performance_data['failure_patterns']}
        
        Common Issues Found:
        {performance_data['common_issues']}
        
        Please provide an optimized version that:
        1. Addresses the identified failure patterns
        2. Improves clarity and specificity
        3. Reduces ambiguity and improves success rate
        4. Maintains the original intent and functionality
        
        Return the optimized prompt with explanation of changes made.
        \"\"\"
        
        # Call Claude API for optimization
        from src.clients.claude_client import ClaudeClient
        claude_client = ClaudeClient()
        
        optimization_response = await claude_client.complete(claude_optimization_prompt)
        
        return optimization_response

class ContinuousLearningSystem:
    \"\"\"System for continuous learning from user interactions\"\"\"
    
    def __init__(self, agent_controller):
        self.agent = agent_controller
        self.learning_data = []
        
    async def learn_from_feedback(self, feedback_data: List[Dict[str, Any]]):
        \"\"\"Learn from user feedback to improve responses\"\"\"
        
        # Analyze feedback patterns
        positive_patterns = self._extract_patterns(
            [f for f in feedback_data if f['rating'] >= 4]
        )
        
        negative_patterns = self._extract_patterns(
            [f for f in feedback_data if f['rating'] <= 2]
        )
        
        # Generate improvement suggestions
        improvements = await self._generate_improvements(
            positive_patterns, negative_patterns
        )
        
        return improvements
    
    def _extract_patterns(self, feedback_subset: List[Dict[str, Any]]) -> Dict[str, Any]:
        \"\"\"Extract patterns from feedback subset\"\"\"
        
        patterns = {
            'common_topics': {},
            'response_characteristics': {},
            'tool_usage_patterns': {},
            'timing_patterns': {}
        }
        
        for feedback in feedback_subset:
            # Extract topics
            topic = self._extract_topic(feedback['original_request'])
            patterns['common_topics'][topic] = patterns['common_topics'].get(topic, 0) + 1
            
            # Extract response characteristics
            response_length = len(feedback['agent_response'].split())
            patterns['response_characteristics']['length'] = patterns['response_characteristics'].get('length', []) + [response_length]
            
            # Extract tool usage
            tools_used = feedback.get('tools_used', [])
            for tool in tools_used:
                patterns['tool_usage_patterns'][tool] = patterns['tool_usage_patterns'].get(tool, 0) + 1
        
        return patterns
    
    async def update_knowledge_base(self, new_knowledge: List[Dict[str, Any]]):
        \"\"\"Update agent's knowledge base with new information\"\"\"
        
        for knowledge_item in new_knowledge:
            # Validate knowledge quality
            quality_score = await self._assess_knowledge_quality(knowledge_item)
            
            if quality_score > 0.8:
                # Add to semantic memory
                await self.agent.memory_manager.update_semantic_memory(knowledge_item)
                
        return len([k for k in new_knowledge if await self._assess_knowledge_quality(k) > 0.8])
"""

5.2 A/B Testing and Experimentation

bash

# Implement A/B testing for agent improvements
claude-code "Create A/B testing framework for agent optimization"

# A/B testing framework:
"""
# src/optimization/ab_testing.py
import random
from typing import Dict, Any, List
from enum import Enum
import asyncio
from datetime import datetime, timedelta

class ExperimentType(Enum):
    PROMPT_VARIANT = "prompt_variant"
    TOOL_SELECTION = "tool_selection"
    RESPONSE_STYLE = "response_style"
    ESCALATION_THRESHOLD = "escalation_threshold"

class ABTestManager:
    \"\"\"Manages A/B tests for agent optimization\"\"\"
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.active_experiments = {}
        self.experiment_results = {}
        
    def create_experiment(
        self,
        name: str,
        experiment_type: ExperimentType,
        variants: Dict[str, Any],
        traffic_split: Dict[str, float],
        success_metrics: List[str],
        duration_days: int = 7
    ) -> str:
        \"\"\"Create new A/B test experiment\"\"\"
        
        experiment_id = f"{name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        
        experiment = {
            'id': experiment_id,
            'name': name,
            'type': experiment_type,
            'variants': variants,
            'traffic_split': traffic_split,
            'success_metrics': success_metrics,
            'start_date': datetime.now(),
            'end_date': datetime.now() + timedelta(days=duration_days),
            'status': 'active',
            'results': {variant: {} for variant in variants.keys()}
        }
        
        self.active_experiments[experiment_id] = experiment
        return experiment_id
    
    def assign_variant(self, experiment_id: str, user_id: str) -> str:
        \"\"\"Assign user to experiment variant\"\"\"
        
        if experiment_id not in self.active_experiments:
            return 'control'
        
        experiment = self.active_experiments[experiment_id]
        
        # Use consistent hashing for user assignment
        user_hash = hash(f"{user_id}_{experiment_id}") % 100
        
        cumulative_percentage = 0
        for variant, percentage in experiment['traffic_split'].items():
            cumulative_percentage += percentage * 100
            if user_hash < cumulative_percentage:
                return variant
        
        return 'control'
    
    async def record_interaction(
        self,
        experiment_id: str,
        variant: str,
        user_id: str,
        interaction_data: Dict[str, Any]
    ):
        \"\"\"Record interaction data for experiment analysis\"\"\"
        
        if experiment_id not in self.active_experiments:
            return
        
        experiment = self.active_experiments[experiment_id]
        
        if variant not in experiment['results']:
            experiment['results'][variant] = {
                'interactions': [],
                'success_rate': 0.0,
                'avg_response_time': 0.0,
                'user_satisfaction': 0.0
            }
        
        experiment['results'][variant]['interactions'].append({
            'user_id': user_id,
            'timestamp': datetime.now(),
            'data': interaction_data
        })
        
        # Update aggregate metrics
        await self._update_experiment_metrics(experiment_id, variant)
    
    async def _update_experiment_metrics(self, experiment_id: str, variant: str):
        \"\"\"Update experiment metrics\"\"\"
        
        experiment = self.active_experiments[experiment_id]
        interactions = experiment['results'][variant]['interactions']
        
        if not interactions:
            return
        
        # Calculate success rate
        successful_interactions = [
            i for i in interactions 
            if i['data'].get('success', False)
        ]
        success_rate = len(successful_interactions) / len(interactions)
        
        # Calculate average response time
        response_times = [
            i['data'].get('response_time', 0) 
            for i in interactions 
            if 'response_time' in i['data']
        ]
        avg_response_time = sum(response_times) / len(response_times) if response_times else 0
        
        # Calculate user satisfaction
        satisfaction_scores = [
            i['data'].get('user_rating', 0) 
            for i in interactions 
            if 'user_rating' in i['data']
        ]
        avg_satisfaction = sum(satisfaction_scores) / len(satisfaction_scores) if satisfaction_scores else 0
        
        experiment['results'][variant].update({
            'success_rate': success_rate,
            'avg_response_time': avg_response_time,
            'user_satisfaction': avg_satisfaction,
            'sample_size': len(interactions)
        })
    
    async def analyze_experiment_results(self, experiment_id: str) -> Dict[str, Any]:
        \"\"\"Analyze experiment results for statistical significance\"\"\"
        
        experiment = self.active_experiments[experiment_id]
        results = experiment['results']
        
        # Statistical significance testing
        analysis = {
            'experiment_id': experiment_id,
            'status': experiment['status'],
            'duration_days': (datetime.now() - experiment['start_date']).days,
            'variants': {}
        }
        
        control_variant = 'control'
        
        for variant_name, variant_data in results.items():
            if variant_data.get('sample_size', 0) < 30:
                continue  # Insufficient data
            
            variant_analysis = {
                'sample_size': variant_data['sample_size'],
                'success_rate': variant_data['success_rate'],
                'avg_response_time': variant_data['avg_response_time'],
                'user_satisfaction': variant_data['user_satisfaction']
            }
            
            # Compare to control if not control variant
            if variant_name != control_variant and control_variant in results:
                variant_analysis['vs_control'] = self._calculate_statistical_significance(
                    variant_data, results[control_variant]
                )
            
            analysis['variants'][variant_name] = variant_analysis
        
        # Determine winning variant
        analysis['recommendation'] = self._determine_winner(analysis['variants'])
        
        return analysis
    
    def _calculate_statistical_significance(
        self, 
        variant_data: Dict[str, Any], 
        control_data: Dict[str, Any]
    ) -> Dict[str, Any]:
        \"\"\"Calculate statistical significance using appropriate tests\"\"\"
        
        # Simplified z-test for proportions (success rate)
        from scipy import stats
        import numpy as np
        
        variant_successes = variant_data['success_rate'] * variant_data['sample_size']
        control_successes = control_data['success_rate'] * control_data['sample_size']
        
        variant_sample = variant_data['sample_size']
        control_sample = control_data['sample_size']
        
        # Z-test for proportions
        z_stat, p_value = stats.proportions_ztest(
            [variant_successes, control_successes],
            [variant_sample, control_sample]
        )
        
        # Effect size (Cohen's h for proportions)
        p1 = variant_data['success_rate']
        p2 = control_data['success_rate']
        effect_size = 2 * (np.arcsin(np.sqrt(p1)) - np.arcsin(np.sqrt(p2)))
        
        return {
            'z_statistic': z_stat,
            'p_value': p_value,
            'significant': p_value < 0.05,
            'effect_size': effect_size,
            'improvement': (p1 - p2) / p2 * 100 if p2 > 0 else 0
        }

# Example usage in agent controller
class ExperimentalAgentController(AgentController):
    \"\"\"Agent controller with A/B testing capabilities\"\"\"
    
    def __init__(self, config: Dict[str, Any]):
        super().__init__(config)
        self.ab_test_manager = ABTestManager(config.get('ab_testing', {}))
        
    async def process_request_with_experiments(
        self, 
        request: str, 
        context: Dict = None
    ) -> str:
        \"\"\"Process request with A/B testing\"\"\"
        
        user_id = context.get('user_id', 'anonymous')
        
        # Check for active experiments
        prompt_experiment = 'prompt_optimization_v2'
        if prompt_experiment in self.ab_test_manager.active_experiments:
            variant = self.ab_test_manager.assign_variant(prompt_experiment, user_id)
            
            # Use experimental prompt
            if variant == 'experimental':
                # Use optimized prompt
                pass
            else:
                # Use control prompt
                pass
        
        # Process request normally but track for experiments
        start_time = time.time()
        response = await super().process_request(request, context)
        response_time = time.time() - start_time
        
        # Record experiment data
        await self.ab_test_manager.record_interaction(
            prompt_experiment,
            variant,
            user_id,
            {
                'success': True,  # Determine success criteria
                'response_time': response_time,
                'request_type': self._classify_request(request)
            }
        )
        
        return response
"""

Quality Gate 5: Optimization Complete

[ ] Performance monitoring active
[ ] Optimization systems implemented
[ ] A/B testing framework operational
[ ] Continuous learning enabled
[ ] Knowledge base updating

Advanced Agent Patterns

Multi-Agent Coordination

bash

# Design multi-agent systems
claude-code "Create a multi-agent coordination system"

# Multi-agent coordination:
"""
Multiple specialized agents working together:

├─ Customer Support Agent (Primary)
├─ Technical Specialist Agent
├─ Billing Agent
├─ Escalation Manager Agent
└─ Quality Assurance Agent

Coordination Patterns:
- Handoff between agents
- Collaborative problem solving
- Consensus decision making
- Load balancing
"""

Autonomous Learning Agents

bash

# Design self-improving agents
claude-code "Create an agent that learns and improves autonomously"

# Self-improving patterns:
"""
Autonomous Learning Components:

1. Experience Collection
2. Pattern Recognition
3. Hypothesis Generation
4. Experimentation
5. Result Evaluation
6. Knowledge Integration
"""

Success Metrics and KPIs

Agent Performance Metrics

Metric	Target	Calculation
Task Success Rate	>85%	Successful completions / Total attempts
User Satisfaction	>4.0/5.0	Average user rating
Response Time	<30s	Time to first response
Escalation Rate	<15%	Escalated tasks / Total tasks

Technical Performance Metrics

Metric	Target	Measurement
API Availability	>99.5%	Uptime monitoring
Error Rate	<2%	Failed requests / Total requests
Memory Usage	<2GB	Peak memory consumption
Tool Success Rate	>95%	Successful tool calls

Business Impact Metrics

Metric	Target	Tracking Method
Cost Reduction	40%	vs. human support cost
Resolution Time	60% faster	vs. human benchmark
Customer Retention	+15%	Post-interaction surveys
Support Volume	+200%	Requests handled capacity

Best Practices for AI Agents

Do's ✅

Start with clear, specific objectives
Design modular, testable architecture
Implement comprehensive monitoring
Plan for failure and edge cases
Use structured memory management
Test thoroughly before production
Monitor and optimize continuously
Implement proper security measures

Don'ts ❌

Don't over-engineer initially
Don't ignore error handling
Don't skip monitoring setup
Don't hard-code business logic
Don't ignore user feedback
Don't deploy without testing
Don't forget rate limiting
Don't ignore privacy requirements

Security and Compliance Considerations

Data Protection

bash

# Implement data protection measures
"""
Data Protection Measures:

1. PII Identification and Masking
2. Data Encryption (at rest and in transit)
3. Access Control and Authentication
4. Audit Logging
5. Data Retention Policies
6. Right to be Forgotten compliance
"""

AI Ethics and Fairness

bash

# Implement ethical AI practices
"""
Ethical AI Implementation:

1. Bias Detection and Mitigation
2. Transparent Decision Making
3. Human Oversight and Control
4. Fairness Across Demographics
5. Explainable AI Responses
6. Privacy by Design
"""

Troubleshooting Common Issues

Issue: Agent provides inconsistent responses

bash

# Solution: Improve prompt consistency and memory
claude-code "Standardize prompts and improve context management"

Issue: Poor performance with complex queries

bash

# Solution: Implement query decomposition and multi-step processing
claude-code "Break complex queries into manageable sub-tasks"

Issue: High API costs

bash

# Solution: Optimize prompts and implement caching
claude-code "Reduce token usage and implement intelligent caching"

Workflow-006: Building AI Agents with Claude ​

Document Control ​

Overview ​

Prerequisites ​

Required SOPs ​

Environment Setup ​

Conceptual Prerequisites ​

Phase 1: DESIGN - Agent Architecture ​

Objective ​

Steps ​

1.1 Define Agent Purpose and Goals ​

1.2 Design Agent Architecture ​

1.3 Define Tool and API Requirements ​

1.4 Design Memory and State Management ​

Quality Gate 1: Design Complete ​

Phase 2: IMPLEMENT - Core Agent Development ​

Objective ​

Steps ​

2.1 Setup Project Structure ​

2.2 Implement Core Agent Controller ​

2.3 Implement Tool Integration Framework ​

2.4 Implement Memory Management ​

Quality Gate 2: Implementation Complete ​

Phase 3: TRAIN - Prompt Engineering and Testing ​

Objective ​

Steps ​

3.1 Design Agent Prompts ​

3.2 Implement Testing Framework ​

3.3 Performance and Quality Testing ​

Quality Gate 3: Training Complete ​

Phase 4: DEPLOY - Production Deployment ​

Objective ​

Steps ​

4.1 Production Environment Setup ​

4.2 API Interface and Scaling ​

4.3 Monitoring and Observability ​

Quality Gate 4: Deployment Complete ​

Phase 5: OPTIMIZE - Performance and Learning ​

Objective ​

Steps ​

5.1 Performance Analysis and Optimization ​

5.2 A/B Testing and Experimentation ​

Quality Gate 5: Optimization Complete ​

Advanced Agent Patterns ​

Multi-Agent Coordination ​

Autonomous Learning Agents ​

Success Metrics and KPIs ​

Agent Performance Metrics ​

Technical Performance Metrics ​

Business Impact Metrics ​

Best Practices for AI Agents ​

Do's ✅ ​

Don'ts ❌ ​

Security and Compliance Considerations ​

Data Protection ​

AI Ethics and Fairness ​

Troubleshooting Common Issues ​

Issue: Agent provides inconsistent responses ​

Issue: Poor performance with complex queries ​

Issue: High API costs ​

See Also ​

Workflow-006: Building AI Agents with Claude

Document Control

Overview

Prerequisites

Required SOPs

Environment Setup

Conceptual Prerequisites

Phase 1: DESIGN - Agent Architecture

Objective

Steps

1.1 Define Agent Purpose and Goals

1.2 Design Agent Architecture

1.3 Define Tool and API Requirements

1.4 Design Memory and State Management

Quality Gate 1: Design Complete

Phase 2: IMPLEMENT - Core Agent Development

Objective

Steps

2.1 Setup Project Structure

2.2 Implement Core Agent Controller

2.3 Implement Tool Integration Framework

2.4 Implement Memory Management

Quality Gate 2: Implementation Complete

Phase 3: TRAIN - Prompt Engineering and Testing

Objective

Steps

3.1 Design Agent Prompts

3.2 Implement Testing Framework

3.3 Performance and Quality Testing

Quality Gate 3: Training Complete

Phase 4: DEPLOY - Production Deployment

Objective

Steps

4.1 Production Environment Setup

4.2 API Interface and Scaling

4.3 Monitoring and Observability

Quality Gate 4: Deployment Complete

Phase 5: OPTIMIZE - Performance and Learning

Objective

Steps

5.1 Performance Analysis and Optimization

5.2 A/B Testing and Experimentation

Quality Gate 5: Optimization Complete

Advanced Agent Patterns

Multi-Agent Coordination

Autonomous Learning Agents

Success Metrics and KPIs

Agent Performance Metrics

Technical Performance Metrics

Business Impact Metrics

Best Practices for AI Agents

Do's ✅

Don'ts ❌

Security and Compliance Considerations

Data Protection

AI Ethics and Fairness

Troubleshooting Common Issues

Issue: Agent provides inconsistent responses

Issue: Poor performance with complex queries

Issue: High API costs

See Also