Planning Your AI Agent

Before writing any code, it's essential to carefully plan your AI agent to ensure it meets your requirements and can be implemented effectively.

Defining Your Agent's Purpose and Scope

Start by clearly articulating what your agent will do and, equally important, what it won't do. This helps establish boundaries and focus your development efforts.

# AI Agent Specification

## Purpose
[One-sentence description of the agent's primary function]

## User Needs
- [Need 1: Description of a specific user need the agent addresses]
- [Need 2: Description of another user need]
- [...]

## Core Capabilities
- [Capability 1: Specific ability the agent will have]
- [Capability 2: Another specific ability]
- [...]

## Out of Scope
- [Limitation 1: What the agent explicitly will NOT do]
- [Limitation 2: Another explicit limitation]
- [...]

## Success Criteria
- [Criterion 1: How you'll know the agent is successful]
- [Criterion 2: Another success measure]
- [...]

## Key Components
- [Component 1: Major architectural component]
- [Component 2: Another major component]
- [...]

## Integration Points
- [Integration 1: External system or API the agent will use]
- [Integration 2: Another external system or API]
- [...]

## Ethical Considerations
- [Consideration 1: Potential ethical issue and mitigation]
- [Consideration 2: Another ethical consideration]
- [...]

Choosing Your Agent Architecture

Based on your agent's purpose and requirements, select an appropriate architecture that will support the necessary capabilities while remaining maintainable and extensible.

Selecting Your Tech Stack

Choose the technologies, frameworks, and services that will power your agent based on your requirements, budget, and development capabilities.

Implementing Core Components

With your architecture and tech stack selected, it's time to implement the core components that will power your AI agent.

1. Setting Up the Project Structure

A well-organized project structure makes development more efficient and helps maintain the codebase as it grows.

# Create project directory
mkdir my_ai_agent
cd my_ai_agent

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Create basic directory structure
mkdir -p src/agent src/tools src/memory src/models src/config src/utils src/api

# Create initial files
touch README.md requirements.txt .env .gitignore
touch src/__init__.py
touch src/agent/__init__.py src/tools/__init__.py src/memory/__init__.py
touch src/models/__init__.py src/config/__init__.py src/utils/__init__.py
touch src/api/__init__.py

# Initialize git repository
git init
echo "venv/" >> .gitignore
echo "__pycache__/" >> .gitignore
echo "*.pyc" >> .gitignore
echo ".env" >> .gitignore

# Create initial requirements file
cat > requirements.txt << EOF
langchain>=0.1.0
openai>=1.0.0
python-dotenv>=1.0.0
pydantic>=2.0.0
fastapi>=0.100.0
uvicorn>=0.23.0
chromadb>=0.4.0
EOF

# Install dependencies
pip install -r requirements.txt

2. Implementing the LLM Interface

The LLM interface provides a consistent way to interact with language models, abstracting away provider-specific details.

# src/models/llm.py
import os
from typing import Dict, List, Any, Optional
from dotenv import load_dotenv
from langchain.llms import OpenAI, Anthropic
from langchain.chat_models import ChatOpenAI, ChatAnthropic
from langchain.schema import HumanMessage, SystemMessage, AIMessage

# Load environment variables
load_dotenv()

class LLMInterface:
    """Interface for interacting with language models."""
    
    def __init__(self, provider: str = "openai", model: str = None, temperature: float = 0.7):
        """
        Initialize the LLM interface.
        
        Args:
            provider: The LLM provider ("openai", "anthropic", etc.)
            model: Specific model to use (if None, uses default for provider)
            temperature: Sampling temperature (0.0 to 1.0)
        """
        self.provider = provider.lower()
        self.temperature = temperature
        
        # Set default models based on provider
        if model is None:
            if self.provider == "openai":
                self.model = "gpt-4"
            elif self.provider == "anthropic":
                self.model = "claude-3-opus-20240229"
            else:
                raise ValueError(f"Unsupported provider: {provider}")
        else:
            self.model = model
        
        # Initialize the appropriate LLM
        self._initialize_llm()
    
    def _initialize_llm(self):
        """Initialize the LLM based on provider and model."""
        if self.provider == "openai":
            self.chat_model = ChatOpenAI(
                model=self.model,
                temperature=self.temperature,
                api_key=os.getenv("OPENAI_API_KEY")
            )
            self.completion_model = OpenAI(
                model=self.model,
                temperature=self.temperature,
                api_key=os.getenv("OPENAI_API_KEY")
            )
        elif self.provider == "anthropic":
            self.chat_model = ChatAnthropic(
                model=self.model,
                temperature=self.temperature,
                api_key=os.getenv("ANTHROPIC_API_KEY")
            )
            self.completion_model = Anthropic(
                model=self.model,
                temperature=self.temperature,
                api_key=os.getenv("ANTHROPIC_API_KEY")
            )
        else:
            raise ValueError(f"Unsupported provider: {self.provider}")
    
    def generate_text(self, prompt: str) -> str:
        """
        Generate text using the completion model.
        
        Args:
            prompt: The text prompt
            
        Returns:
            Generated text response
        """
        return self.completion_model.predict(prompt)
    
    def generate_chat_response(self, 
                              system_message: str,
                              user_message: str,
                              chat_history: Optional[List[Dict[str, str]]] = None) -> str:
        """
        Generate a response in a chat context.
        
        Args:
            system_message: The system instructions
            user_message: The user's message
            chat_history: Optional list of previous messages
            
        Returns:
            Generated assistant response
        """
        messages = [SystemMessage(content=system_message)]
        
        # Add chat history if provided
        if chat_history:
            for message in chat_history:
                if message["role"] == "user":
                    messages.append(HumanMessage(content=message["content"]))
                elif message["role"] == "assistant":
                    messages.append(AIMessage(content=message["content"]))
        
        # Add the current user message
        messages.append(HumanMessage(content=user_message))
        
        # Generate response
        response = self.chat_model.predict_messages(messages)
        return response.content
    
    def get_embedding(self, text: str) -> List[float]:
        """
        Get embedding vector for text.
        
        Args:
            text: The text to embed
            
        Returns:
            Embedding vector
        """
        # This is a simplified implementation
        # In a real application, you would use a dedicated embedding model
        from langchain.embeddings import OpenAIEmbeddings
        
        embeddings = OpenAIEmbeddings(
            model="text-embedding-3-small",
            api_key=os.getenv("OPENAI_API_KEY")
        )
        
        return embeddings.embed_query(text)


# Example usage
if __name__ == "__main__":
    # Create LLM interface
    llm = LLMInterface(provider="openai", model="gpt-4")
    
    # Generate text
    response = llm.generate_text("Explain the concept of AI agents in one paragraph.")
    print(f"Text generation response: {response}\n")
    
    # Generate chat response
    chat_response = llm.generate_chat_response(
        system_message="You are a helpful AI assistant that specializes in explaining technical concepts.",
        user_message="What is the difference between an AI agent and a regular chatbot?",
        chat_history=[
            {"role": "user", "content": "I'm learning about AI systems."},
            {"role": "assistant", "content": "That's great! AI systems are a fascinating field with many different approaches and applications."}
        ]
    )
    print(f"Chat response: {chat_response}")

3. Building the Tool System

The tool system enables your agent to interact with external systems and perform actions beyond just generating text.

# src/tools/base.py
from typing import Dict, List, Any, Callable, Optional
from pydantic import BaseModel, Field, create_model
import inspect
import json

class Tool:
    """Base class for agent tools."""
    
    def __init__(self, 
                name: str, 
                description: str, 
                func: Callable,
                args_schema: Optional[BaseModel] = None):
        """
        Initialize a tool.
        
        Args:
            name: Tool name
            description: Tool description
            func: Function that implements the tool
            args_schema: Pydantic model for argument validation (optional)
        """
        self.name = name
        self.description = description
        self.func = func
        
        # If no schema provided, create one from function signature
        if args_schema is None:
            self.args_schema = self._create_schema_from_function(func)
        else:
            self.args_schema = args_schema
    
    def _create_schema_from_function(self, func: Callable) -> BaseModel:
        """Create a Pydantic model from function signature."""
        sig = inspect.signature(func)
        fields = {}
        
        for name, param in sig.parameters.items():
            # Skip self parameter for methods
            if name == 'self':
                continue
                
            # Get annotation or default to Any
            annotation = param.annotation if param.annotation != inspect.Parameter.empty else Any
            
            # Get default value if available
            default = ... if param.default == inspect.Parameter.empty else param.default
            
            # Add field with description from docstring if available
            fields[name] = (annotation, Field(default, description=f"Parameter: {name}"))
        
        # Create and return the model
        return create_model(f"{func.__name__}Schema", **fields)
    
    def __call__(self, **kwargs):
        """Execute the tool with the provided arguments."""
        # Validate arguments
        validated_args = self.args_schema(**kwargs).dict()
        
        # Execute function
        return self.func(**validated_args)
    
    def get_schema(self) -> Dict[str, Any]:
        """Get the JSON schema for this tool."""
        schema_dict = self.args_schema.schema()
        
        return {
            "name": self.name,
            "description": self.description,
            "parameters": schema_dict
        }


# src/tools/web_tools.py
import requests
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field
from bs4 import BeautifulSoup
from .base import Tool

class SearchParameters(BaseModel):
    query: str = Field(..., description="The search query")
    num_results: int = Field(5, description="Number of results to return")

def search_web(query: str, num_results: int = 5) -> List[Dict[str, str]]:
    """
    Search the web for information.
    
    Args:
        query: Search query
        num_results: Number of results to return
        
    Returns:
        List of search results with title and URL
    """
    # This is a simplified implementation
    # In a real application, you would use a search API like Google, Bing, or DuckDuckGo
    
    # Simulate search results
    results = []
    for i in range(min(num_results, 10)):
        results.append({
            "title": f"Result {i+1} for: {query}",
            "url": f"https://example.com/result/{i+1}?q={query.replace(' ', '+')}"
        })
    
    return results

class WebPageParameters(BaseModel):
    url: str = Field(..., description="URL of the webpage to fetch")

def fetch_webpage(url: str) -> str:
    """
    Fetch and extract text content from a webpage.
    
    Args:
        url: URL of the webpage
        
    Returns:
        Extracted text content
    """
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        
        # Parse HTML
        soup = BeautifulSoup(response.text, 'html.parser')
        
        # Remove script and style elements
        for script in soup(["script", "style"]):
            script.extract()
        
        # Extract text
        text = soup.get_text(separator='\n')
        
        # Clean up text
        lines = (line.strip() for line in text.splitlines())
        chunks = (phrase.strip() for line in lines for phrase in line.split("  "))
        text = '\n'.join(chunk for chunk in chunks if chunk)
        
        return text
    except Exception as e:
        return f"Error fetching webpage: {str(e)}"

# Create tool instances
search_tool = Tool(
    name="search_web",
    description="Search the web for information. Use this when you need to find facts or current information.",
    func=search_web,
    args_schema=SearchParameters
)

webpage_tool = Tool(
    name="fetch_webpage",
    description="Fetch and extract text content from a webpage. Use this to get detailed information from a specific URL.",
    func=fetch_webpage,
    args_schema=WebPageParameters
)


# src/tools/utility_tools.py
import datetime
import json
import os
from typing import Dict, List, Any, Optional
from pydantic import BaseModel, Field
from .base import Tool

class CalculatorParameters(BaseModel):
    expression: str = Field(..., description="Mathematical expression to evaluate")

def calculator(expression: str) -> str:
    """
    Evaluate a mathematical expression.
    
    Args:
        expression: Mathematical expression to evaluate
        
    Returns:
        Result of the evaluation
    """
    try:
        # Use safer eval with restricted globals
        allowed_names = {
            "abs": abs,
            "float": float,
            "int": int,
            "max": max,
            "min": min,
            "pow": pow,
            "round": round,
            "sum": sum
        }
        
        # Add common math functions
        import math
        for name in dir(math):
            if not name.startswith("_"):
                allowed_names[name] = getattr(math, name)
        
        # Evaluate expression
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return f"Result: {result}"
    except Exception as e:
        return f"Error evaluating expression: {str(e)}"

class CurrentTimeParameters(BaseModel):
    timezone: Optional[str] = Field(None, description="Timezone (default: UTC)")

def get_current_time(timezone: Optional[str] = None) -> str:
    """
    Get the current date and time.
    
    Args:
        timezone: Timezone (default: UTC)
        
    Returns:
        Current date and time string
    """
    try:
        now = datetime.datetime.now(datetime.timezone.utc)
        
        if timezone:
            # This is a simplified implementation
            # In a real application, you would use pytz or similar
            return f"Current time ({timezone}): {now} (Note: timezone conversion not implemented)"
        
        return f"Current time (UTC): {now}"
    except Exception as e:
        return f"Error getting current time: {str(e)}"

# Create tool instances
calculator_tool = Tool(
    name="calculator",
    description="Evaluate mathematical expressions. Use this for calculations.",
    func=calculator,
    args_schema=CalculatorParameters
)

time_tool = Tool(
    name="get_current_time",
    description="Get the current date and time. Use this when you need to know the current time.",
    func=get_current_time,
    args_schema=CurrentTimeParameters
)


# src/tools/__init__.py
from .base import Tool
from .web_tools import search_tool, webpage_tool
from .utility_tools import calculator_tool, time_tool

# Collect all tools
available_tools = {
    "search_web": search_tool,
    "fetch_webpage": webpage_tool,
    "calculator": calculator_tool,
    "get_current_time": time_tool
}

__all__ = ["Tool", "available_tools", "search_tool", "webpage_tool", "calculator_tool", "time_tool"]

4. Implementing the Memory System

The memory system allows your agent to maintain context, remember past interactions, and store knowledge for future use.

# src/memory/base.py
from typing import Dict, List, Any, Optional
from datetime import datetime
import json
import uuid

class Memory:
    """Base class for agent memory systems."""
    
    def __init__(self):
        """Initialize the memory system."""
        pass
    
    def add(self, item: Dict[str, Any]) -> str:
        """
        Add an item to memory.
        
        Args:
            item: The item to add
            
        Returns:
            ID of the added item
        """
        raise NotImplementedError("Subclasses must implement add()")
    
    def get(self, item_id: str) -> Optional[Dict[str, Any]]:
        """
        Get an item from memory by ID.
        
        Args:
            item_id: ID of the item to retrieve
            
        Returns:
            The item if found, None otherwise
        """
        raise NotImplementedError("Subclasses must implement get()")
    
    def search(self, query: str, limit: int = 5) -> List[Dict[str, Any]]:
        """
        Search memory for relevant items.
        
        Args:
            query: Search query
            limit: Maximum number of results
            
        Returns:
            List of matching items
        """
        raise NotImplementedError("Subclasses must implement search()")
    
    def update(self, item_id: str, item: Dict[str, Any]) -> bool:
        """
        Update an item in memory.
        
        Args:
            item_id: ID of the item to update
            item: Updated item data
            
        Returns:
            True if successful, False otherwise
        """
        raise NotImplementedError("Subclasses must implement update()")
    
    def clear(self) -> None:
        """Clear all items from memory."""
        raise NotImplementedError("Subclasses must implement clear()")


# src/memory/conversation_memory.py
from typing import Dict, List, Any, Optional
from datetime import datetime
import json
import uuid
from .base import Memory

class ConversationMemory(Memory):
    """Memory system for storing conversation history."""
    
    def __init__(self, max_items: int = 100):
        """
        Initialize conversation memory.
        
        Args:
            max_items: Maximum number of items to store
        """
        super().__init__()
        self.max_items = max_items
        self.items = []
    
    def add(self, item: Dict[str, Any]) -> str:
        """
        Add a message to conversation history.
        
        Args:
            item: Message to add (must contain 'role' and 'content')
            
        Returns:
            ID of the added message
        """
        if 'role' not in item or 'content' not in item:
            raise ValueError("Item must contain 'role' and 'content' fields")
        
        # Add timestamp and ID if not present
        if 'timestamp' not in item:
            item['timestamp'] = datetime.now().isoformat()
        
        if 'id' not in item:
            item['id'] = str(uuid.uuid4())
        
        # Add to items
        self.items.append(item)
        
        # Trim if exceeding max size
        if len(self.items) > self.max_items:
            self.items = self.items[-self.max_items:]
        
        return item['id']
    
    def get(self, item_id: str) -> Optional[Dict[str, Any]]:
        """
        Get a message by ID.
        
        Args:
            item_id: ID of the message
            
        Returns:
            Message if found, None otherwise
        """
        for item in self.items:
            if item.get('id') == item_id:
                return item
        
        return None
    
    def search(self, query: str, limit: int = 5) -> List[Dict[str, Any]]:
        """
        Search conversation history.
        
        Args:
            query: Search query
            limit: Maximum number of results
            
        Returns:
            List of matching messages
        """
        # Simple text search implementation
        results = []
        query = query.lower()
        
        for item in reversed(self.items):  # Most recent first
            if query in item.get('content', '').lower():
                results.append(item)
                if len(results) >= limit:
                    break
        
        return results
    
    def update(self, item_id: str, item: Dict[str, Any]) -> bool:
        """
        Update a message.
        
        Args:
            item_id: ID of the message to update
            item: Updated message data
            
        Returns:
            True if successful, False otherwise
        """
        for i, existing_item in enumerate(self.items):
            if existing_item.get('id') == item_id:
                # Preserve ID and timestamp
                item['id'] = item_id
                if 'timestamp' in existing_item:
                    item['timestamp'] = existing_item['timestamp']
                
                self.items[i] = item
                return True
        
        return False
    
    def clear(self) -> None:
        """Clear conversation history."""
        self.items = []
    
    def get_last_n(self, n: int) -> List[Dict[str, Any]]:
        """
        Get the last N messages.
        
        Args:
            n: Number of messages to retrieve
            
        Returns:
            List of the last N messages
        """
        return self.items[-n:] if n > 0 else []
    
    def get_formatted_history(self, n: Optional[int] = None) -> List[Dict[str, str]]:
        """
        Get formatted conversation history for LLM context.
        
        Args:
            n: Optional limit on number of messages
            
        Returns:
            List of messages in format expected by LLM
        """
        history = self.items[-n:] if n is not None else self.items
        
        formatted = []
        for item in history:
            formatted.append({
                "role": item["role"],
                "content": item["content"]
            })
        
        return formatted


# src/memory/knowledge_memory.py
from typing import Dict, List, Any, Optional
import json
import uuid
from datetime import datetime
import os
from .base import Memory

try:
    import chromadb
    from chromadb.config import Settings
except ImportError:
    print("ChromaDB not installed. Run: pip install chromadb")

class KnowledgeMemory(Memory):
    """Vector-based memory system for storing knowledge."""
    
    def __init__(self, embedding_function, collection_name: str = "knowledge", persist_directory: Optional[str] = None):
        """
        Initialize knowledge memory.
        
        Args:
            embedding_function: Function to generate embeddings
            collection_name: Name of the collection
            persist_directory: Directory to persist the database (optional)
        """
        super().__init__()
        self.embedding_function = embedding_function
        
        # Initialize ChromaDB
        if persist_directory:
            self.client = chromadb.PersistentClient(path=persist_directory)
        else:
            self.client = chromadb.Client()
        
        # Get or create collection
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            embedding_function=self.embedding_function
        )
    
    def add(self, item: Dict[str, Any]) -> str:
        """
        Add a knowledge item to memory.
        
        Args:
            item: Knowledge item to add (must contain 'content')
            
        Returns:
            ID of the added item
        """
        if 'content' not in item:
            raise ValueError("Item must contain 'content' field")
        
        # Add metadata and ID if not present
        if 'metadata' not in item:
            item['metadata'] = {}
        
        if 'timestamp' not in item['metadata']:
            item['metadata']['timestamp'] = datetime.now().isoformat()
        
        if 'id' not in item:
            item['id'] = str(uuid.uuid4())
        
        # Add to collection
        self.collection.add(
            ids=[item['id']],
            documents=[item['content']],
            metadatas=[item['metadata']]
        )
        
        return item['id']
    
    def get(self, item_id: str) -> Optional[Dict[str, Any]]:
        """
        Get a knowledge item by ID.
        
        Args:
            item_id: ID of the item
            
        Returns:
            Knowledge item if found, None otherwise
        """
        try:
            result = self.collection.get(ids=[item_id])
            
            if result['ids'] and len(result['ids']) > 0:
                return {
                    'id': result['ids'][0],
                    'content': result['documents'][0],
                    'metadata': result['metadatas'][0]
                }
        except Exception:
            pass
        
        return None
    
    def search(self, query: str, limit: int = 5) -> List[Dict[str, Any]]:
        """
        Search knowledge memory for relevant items.
        
        Args:
            query: Search query
            limit: Maximum number of results
            
        Returns:
            List of matching items
        """
        results = self.collection.query(
            query_texts=[query],
            n_results=limit
        )
        
        items = []
        if results['ids'] and len(results['ids'][0]) > 0:
            for i in range(len(results['ids'][0])):
                items.append({
                    'id': results['ids'][0][i],
                    'content': results['documents'][0][i],
                    'metadata': results['metadatas'][0][i]
                })
        
        return items
    
    def update(self, item_id: str, item: Dict[str, Any]) -> bool:
        """
        Update a knowledge item.
        
        Args:
            item_id: ID of the item to update
            item: Updated item data
            
        Returns:
            True if successful, False otherwise
        """
        if 'content' not in item:
            raise ValueError("Item must contain 'content' field")
        
        # Prepare metadata
        if 'metadata' not in item:
            item['metadata'] = {}
        
        # Update in collection
        try:
            self.collection.update(
                ids=[item_id],
                documents=[item['content']],
                metadatas=[item['metadata']]
            )
            return True
        except Exception:
            return False
    
    def clear(self) -> None:
        """Clear knowledge memory."""
        self.collection.delete()


# src/memory/__init__.py
from .base import Memory
from .conversation_memory import ConversationMemory
from .knowledge_memory import KnowledgeMemory

__all__ = ["Memory", "ConversationMemory", "KnowledgeMemory"]

5. Building the Agent Core

The agent core ties together all the components and implements the main agent loop that processes user requests and generates responses.

# src/agent/base.py
from typing import Dict, List, Any, Optional
import json
import re
from ..models.llm import LLMInterface
from ..memory.conversation_memory import ConversationMemory
from ..memory.knowledge_memory import KnowledgeMemory
from ..tools.base import Tool

class Agent:
    """Base class for AI agents."""
    
    def __init__(self, 
                llm: LLMInterface,
                tools: Dict[str, Tool] = None,
                conversation_memory: Optional[ConversationMemory] = None,
                knowledge_memory: Optional[KnowledgeMemory] = None,
                system_message: str = "You are a helpful AI assistant."):
        """
        Initialize the agent.
        
        Args:
            llm: Language model interface
            tools: Dictionary of available tools
            conversation_memory: Memory for conversation history
            knowledge_memory: Memory for knowledge storage
            system_message: System message for the agent
        """
        self.llm = llm
        self.tools = tools or {}
        self.conversation_memory = conversation_memory or ConversationMemory()
        self.knowledge_memory = knowledge_memory
        self.system_message = system_message
    
    def process_message(self, user_message: str) -> str:
        """
        Process a user message and generate a response.
        
        Args:
            user_message: User's message
            
        Returns:
            Agent's response
        """
        # Add user message to conversation memory
        self.conversation_memory.add({
            "role": "user",
            "content": user_message
        })
        
        # Get conversation history
        history = self.conversation_memory.get_formatted_history()
        
        # Generate response (to be implemented by subclasses)
        response = self._generate_response(user_message, history)
        
        # Add agent response to conversation memory
        self.conversation_memory.add({
            "role": "assistant",
            "content": response
        })
        
        return response
    
    def _generate_response(self, user_message: str, history: List[Dict[str, str]]) -> str:
        """
        Generate a response to the user message.
        
        Args:
            user_message: User's message
            history: Conversation history
            
        Returns:
            Agent's response
        """
        raise NotImplementedError("Subclasses must implement _generate_response()")
    
    def get_conversation_history(self) -> List[Dict[str, Any]]:
        """
        Get the conversation history.
        
        Returns:
            List of conversation messages
        """
        return self.conversation_memory.items


# src/agent/reactive_agent.py
from typing import Dict, List, Any, Optional
import json
import re
from .base import Agent
from ..models.llm import LLMInterface
from ..memory.conversation_memory import ConversationMemory
from ..memory.knowledge_memory import KnowledgeMemory
from ..tools.base import Tool

class ReactiveAgent(Agent):
    """
    Reactive agent that uses a simple prompt-response pattern.
    This agent doesn't use planning or complex reasoning.
    """
    
    def __init__(self, 
                llm: LLMInterface,
                tools: Dict[str, Tool] = None,
                conversation_memory: Optional[ConversationMemory] = None,
                knowledge_memory: Optional[KnowledgeMemory] = None,
                system_message: str = "You are a helpful AI assistant."):
        """Initialize the reactive agent."""
        super().__init__(llm, tools, conversation_memory, knowledge_memory, system_message)
    
    def _generate_response(self, user_message: str, history: List[Dict[str, str]]) -> str:
        """
        Generate a response using the reactive pattern.
        
        Args:
            user_message: User's message
            history: Conversation history
            
        Returns:
            Agent's response
        """
        # If no tools, just generate a direct response
        if not self.tools:
            return self.llm.generate_chat_response(
                system_message=self.system_message,
                user_message=user_message,
                chat_history=history[:-1]  # Exclude the latest user message
            )
        
        # With tools, determine if a tool should be used
        tool_descriptions = "\n".join([
            f"- {name}: {tool.description}" for name, tool in self.tools.items()
        ])
        
        tool_decision_prompt = f"""
        {self.system_message}
        
        You have access to the following tools:
        {tool_descriptions}
        
        Based on the user's message, determine if you should use a tool or respond directly.
        
        User message: "{user_message}"
        
        If you need to use a tool, respond in this format:
        TOOL: [tool_name]
        ARGS: {{
            "param1": "value1",
            "param2": "value2"
        }}
        
        If you should respond directly, respond in this format:
        RESPONSE: [Your direct response to the user]
        """
        
        decision = self.llm.generate_text(tool_decision_prompt)
        
        # Parse the decision
        if "TOOL:" in decision:
            # Extract tool name and arguments
            tool_match = re.search(r"TOOL: (\w+)", decision)
            args_match = re.search(r"ARGS: ({.*})", decision, re.DOTALL)
            
            if tool_match and args_match:
                tool_name = tool_match.group(1)
                args_str = args_match.group(1)
                
                try:
                    # Parse arguments
                    args = json.loads(args_str)
                    
                    # Check if tool exists
                    if tool_name in self.tools:
                        # Execute tool
                        tool = self.tools[tool_name]
                        tool_result = tool(**args)
                        
                        # Generate response based on tool result
                        response_prompt = f"""
                        {self.system_message}
                        
                        You used the tool "{tool_name}" with arguments {args} and got this result:
                        {tool_result}
                        
                        Based on this result, provide a helpful response to the user's message:
                        "{user_message}"
                        """
                        
                        return self.llm.generate_text(response_prompt)
                    else:
                        return f"I tried to use a tool called '{tool_name}', but it's not available. I can help you in another way instead."
                except Exception as e:
                    return f"I tried to use a tool to help you, but encountered an error: {str(e)}. Let me help you differently."
        
        # If no tool was used or there was an error parsing, generate a direct response
        response_match = re.search(r"RESPONSE: (.*)", decision, re.DOTALL)
        if response_match:
            return response_match.group(1).strip()
        else:
            # Fallback if the format wasn't followed
            return self.llm.generate_chat_response(
                system_message=self.system_message,
                user_message=user_message,
                chat_history=history[:-1]  # Exclude the latest user message
            )


# src/agent/reflective_agent.py
from typing import Dict, List, Any, Optional
import json
import re
from .base import Agent
from ..models.llm import LLMInterface
from ..memory.conversation_memory import ConversationMemory
from ..memory.knowledge_memory import KnowledgeMemory
from ..tools.base import Tool

class ReflectiveAgent(Agent):
    """
    Reflective agent that uses the Reflection pattern to improve responses.
    This agent thinks about its answers before responding.
    """
    
    def __init__(self, 
                llm: LLMInterface,
                tools: Dict[str, Tool] = None,
                conversation_memory: Optional[ConversationMemory] = None,
                knowledge_memory: Optional[KnowledgeMemory] = None,
                system_message: str = "You are a helpful AI assistant.",
                reflection_rounds: int = 1):
        """
        Initialize the reflective agent.
        
        Args:
            llm: Language model interface
            tools: Dictionary of available tools
            conversation_memory: Memory for conversation history
            knowledge_memory: Memory for knowledge storage
            system_message: System message for the agent
            reflection_rounds: Number of reflection rounds to perform
        """
        super().__init__(llm, tools, conversation_memory, knowledge_memory, system_message)
        self.reflection_rounds = reflection_rounds
    
    def _generate_response(self, user_message: str, history: List[Dict[str, str]]) -> str:
        """
        Generate a response using the reflective pattern.
        
        Args:
            user_message: User's message
            history: Conversation history
            
        Returns:
            Agent's response
        """
        # If tools are available, include them in the system message
        system_message = self.system_message
        if self.tools:
            tool_descriptions = "\n".join([
                f"- {name}: {tool.description}" for name, tool in self.tools.items()
            ])
            system_message += f"\n\nYou have access to the following tools:\n{tool_descriptions}"
        
        # Initial response generation
        initial_response = self.llm.generate_chat_response(
            system_message=system_message,
            user_message=user_message,
            chat_history=history[:-1]  # Exclude the latest user message
        )
        
        current_response = initial_response
        
        # Reflection rounds
        for i in range(self.reflection_rounds):
            # Generate reflection
            reflection_prompt = f"""
            {system_message}
            
            You generated this response to the user's message:
            
            User: {user_message}
            
            Your response:
            {current_response}
            
            Reflect on your response and identify any issues or areas for improvement:
            1. Are there any factual errors or inaccuracies?
            2. Is the response complete and thorough?
            3. Is the tone appropriate and helpful?
            4. Could the explanation be clearer or more concise?
            5. Are there any missed opportunities to provide value?
            
            Provide a detailed critique of your response.
            """
            
            reflection = self.llm.generate_text(reflection_prompt)
            
            # Improve response based on reflection
            improvement_prompt = f"""
            {system_message}
            
            User message: {user_message}
            
            Your previous response:
            {current_response}
            
            Your reflection on that response:
            {reflection}
            
            Based on this reflection, provide an improved response that addresses the issues identified.
            """
            
            improved_response = self.llm.generate_text(improvement_prompt)
            current_response = improved_response
        
        # Check if we need to use tools
        if self.tools:
            # Look for tool usage patterns in the response
            tool_pattern = r"I'll use the (\w+) tool to (.*?)(?:\.|$)"
            tool_matches = re.finditer(tool_pattern, current_response)
            
            for match in tool_matches:
                tool_name = match.group(1)
                purpose = match.group(2)
                
                if tool_name in self.tools:
                    # Extract parameters from the response context
                    # This is a simplified approach; in a real implementation,
                    # you would use a more sophisticated method to extract parameters
                    param_extraction_prompt = f"""
                    I need to use the {tool_name} tool for this purpose: {purpose}
                    
                    The tool requires these parameters:
                    {json.dumps(self.tools[tool_name].args_schema.schema()["properties"])}
                    
                    Based on the user's message: "{user_message}"
                    
                    Extract the appropriate parameter values in JSON format.
                    """
                    
                    param_response = self.llm.generate_text(param_extraction_prompt)
                    
                    # Try to extract JSON from the response
                    try:
                        # Find JSON-like structure in the response
                        json_match = re.search(r"({.*})", param_response, re.DOTALL)
                        if json_match:
                            params = json.loads(json_match.group(1))
                            
                            # Execute tool
                            tool = self.tools[tool_name]
                            tool_result = tool(**params)
                            
                            # Update response with tool result
                            tool_integration_prompt = f"""
                            {system_message}
                            
                            User message: {user_message}
                            
                            You were planning to use the {tool_name} tool for this purpose: {purpose}
                            
                            The tool returned this result:
                            {tool_result}
                            
                            Your current draft response:
                            {current_response}
                            
                            Update your response to incorporate this tool result naturally.
                            The updated response should flow well and not explicitly mention "using a tool"
                            unless it's necessary for the user to understand the process.
                            """
                            
                            current_response = self.llm.generate_text(tool_integration_prompt)
                    except:
                        # If we can't parse parameters, continue with the current response
                        pass
        
        return current_response


# src/agent/planning_agent.py
from typing import Dict, List, Any, Optional
import json
import re
from .base import Agent
from ..models.llm import LLMInterface
from ..memory.conversation_memory import ConversationMemory
from ..memory.knowledge_memory import KnowledgeMemory
from ..tools.base import Tool

class PlanningAgent(Agent):
    """
    Planning agent that uses the Plan-Execute-Reflect pattern.
    This agent creates a plan before taking action.
    """
    
    def __init__(self, 
                llm: LLMInterface,
                tools: Dict[str, Tool] = None,
                conversation_memory: Optional[ConversationMemory] = None,
                knowledge_memory: Optional[KnowledgeMemory] = None,
                system_message: str = "You are a helpful AI assistant."):
        """Initialize the planning agent."""
        super().__init__(llm, tools, conversation_memory, knowledge_memory, system_message)
    
    def _generate_response(self, user_message: str, history: List[Dict[str, str]]) -> str:
        """
        Generate a response using the planning pattern.
        
        Args:
            user_message: User's message
            history: Conversation history
            
        Returns:
            Agent's response
        """
        # If no tools, use a simpler approach
        if not self.tools:
            return self.llm.generate_chat_response(
                system_message=self.system_message,
                user_message=user_message,
                chat_history=history[:-1]  # Exclude the latest user message
            )
        
        # With tools, use the planning approach
        tool_descriptions = "\n".join([
            f"- {name}: {tool.description}" for name, tool in self.tools.items()
        ])
        
        # Phase 1: Planning
        planning_prompt = f"""
        {self.system_message}
        
        You have access to the following tools:
        {tool_descriptions}
        
        User message: "{user_message}"
        
        First, create a plan to address the user's message. Consider:
        1. What is the user asking for?
        2. What information do you need to gather?
        3. Which tools would be helpful?
        4. What steps should you take?
        
        Format your plan as a numbered list of steps.
        """
        
        plan = self.llm.generate_text(planning_prompt)
        
        # Phase 2: Execution
        execution_results = []
        
        # Extract steps from the plan
        steps = re.findall(r"\d+\.\s+(.*?)(?=\n\d+\.|\n\n|$)", plan, re.DOTALL)
        
        for i, step in enumerate(steps):
            # Check if this step involves using a tool
            tool_decision_prompt = f"""
            Plan step: "{step}"
            
            Available tools:
            {tool_descriptions}
            
            Should this step use a tool? If yes, which tool and with what parameters?
            
            If a tool should be used, respond in this format:
            TOOL: [tool_name]
            ARGS: {{
                "param1": "value1",
                "param2": "value2"
            }}
            
            If no tool is needed, respond with:
            NO_TOOL
            """
            
            tool_decision = self.llm.generate_text(tool_decision_prompt)
            
            if "TOOL:" in tool_decision:
                # Extract tool name and arguments
                tool_match = re.search(r"TOOL: (\w+)", tool_decision)
                args_match = re.search(r"ARGS: ({.*})", tool_decision, re.DOTALL)
                
                if tool_match and args_match:
                    tool_name = tool_match.group(1)
                    args_str = args_match.group(1)
                    
                    try:
                        # Parse arguments
                        args = json.loads(args_str)
                        
                        # Check if tool exists
                        if tool_name in self.tools:
                            # Execute tool
                            tool = self.tools[tool_name]
                            tool_result = tool(**args)
                            
                            execution_results.append({
                                "step": i + 1,
                                "description": step,
                                "tool": tool_name,
                                "args": args,
                                "result": tool_result
                            })
                        else:
                            execution_results.append({
                                "step": i + 1,
                                "description": step,
                                "error": f"Tool '{tool_name}' not available"
                            })
                    except Exception as e:
                        execution_results.append({
                            "step": i + 1,
                            "description": step,
                            "error": f"Error executing tool: {str(e)}"
                        })
            else:
                # No tool needed for this step
                execution_results.append({
                    "step": i + 1,
                    "description": step,
                    "note": "No tool execution required"
                })
        
        # Phase 3: Reflection and Response Generation
        response_prompt = f"""
        {self.system_message}
        
        User message: "{user_message}"
        
        Your plan:
        {plan}
        
        Execution results:
        {json.dumps(execution_results, indent=2)}
        
        Based on the plan and execution results, provide a comprehensive response to the user.
        Your response should:
        1. Address the user's original request
        2. Incorporate information from tool results
        3. Be well-structured and easy to understand
        4. Not explicitly mention the planning process or tool usage unless relevant
        
        Your response:
        """
        
        return self.llm.generate_text(response_prompt)


# src/agent/__init__.py
from .base import Agent
from .reactive_agent import ReactiveAgent
from .reflective_agent import ReflectiveAgent
from .planning_agent import PlanningAgent

__all__ = ["Agent", "ReactiveAgent", "ReflectiveAgent", "PlanningAgent"]

Creating the Agent Interface

With the core components in place, it's time to create interfaces for users to interact with your agent.

1. Building a Command-Line Interface

A command-line interface provides a simple way to test and interact with your agent during development.

# src/cli.py
import os
import argparse
import dotenv
from src.models.llm import LLMInterface
from src.tools import available_tools
from src.memory.conversation_memory import ConversationMemory
from src.agent.reactive_agent import ReactiveAgent
from src.agent.reflective_agent import ReflectiveAgent
from src.agent.planning_agent import PlanningAgent

# Load environment variables
dotenv.load_dotenv()

def main():
    """Run the AI agent CLI."""
    parser = argparse.ArgumentParser(description="AI Agent CLI")
    parser.add_argument("--agent-type", type=str, default="reactive",
                        choices=["reactive", "reflective", "planning"],
                        help="Type of agent to use")
    parser.add_argument("--llm-provider", type=str, default="openai",
                        choices=["openai", "anthropic"],
                        help="LLM provider to use")
    parser.add_argument("--model", type=str, default=None,
                        help="Specific model to use (defaults to provider's default)")
    parser.add_argument("--temperature", type=float, default=0.7,
                        help="Sampling temperature (0.0 to 1.0)")
    parser.add_argument("--system-message", type=str,
                        default="You are a helpful AI assistant.",
                        help="System message for the agent")
    parser.add_argument("--no-tools", action="store_true",
                        help="Disable tools")
    
    args = parser.parse_args()
    
    # Initialize LLM
    llm = LLMInterface(
        provider=args.llm_provider,
        model=args.model,
        temperature=args.temperature
    )
    
    # Initialize memory
    conversation_memory = ConversationMemory()
    
    # Initialize tools
    tools = None if args.no_tools else available_tools
    
    # Initialize agent
    if args.agent_type == "reactive":
        agent = ReactiveAgent(
            llm=llm,
            tools=tools,
            conversation_memory=conversation_memory,
            system_message=args.system_message
        )
    elif args.agent_type == "reflective":
        agent = ReflectiveAgent(
            llm=llm,
            tools=tools,
            conversation_memory=conversation_memory,
            system_message=args.system_message,
            reflection_rounds=1
        )
    elif args.agent_type == "planning":
        agent = PlanningAgent(
            llm=llm,
            tools=tools,
            conversation_memory=conversation_memory,
            system_message=args.system_message
        )
    else:
        raise ValueError(f"Unknown agent type: {args.agent_type}")
    
    print(f"AI Agent initialized ({args.agent_type} with {args.llm_provider})")
    print("Type 'exit' or 'quit' to end the conversation")
    print("Type 'history' to view conversation history")
    print("Type 'clear' to clear conversation history")
    print("-" * 50)
    
    # Main conversation loop
    while True:
        user_input = input("\nYou: ")
        
        if user_input.lower() in ["exit", "quit"]:
            print("Goodbye!")
            break
        elif user_input.lower() == "history":
            history = agent.get_conversation_history()
            print("\nConversation History:")
            for item in history:
                role = item["role"]
                content = item["content"]
                print(f"{role.capitalize()}: {content}")
            continue
        elif user_input.lower() == "clear":
            agent.conversation_memory.clear()
            print("Conversation history cleared")
            continue
        
        # Process user input
        response = agent.process_message(user_input)
        print(f"\nAgent: {response}")

if __name__ == "__main__":
    main()

2. Creating a Web API

A web API allows your agent to be accessed by various client applications, including web and mobile interfaces.

# src/api/app.py
import os
from typing import Dict, List, Any, Optional
from fastapi import FastAPI, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import dotenv
from src.models.llm import LLMInterface
from src.tools import available_tools
from src.memory.conversation_memory import ConversationMemory
from src.agent.reactive_agent import ReactiveAgent
from src.agent.reflective_agent import ReflectiveAgent
from src.agent.planning_agent import PlanningAgent

# Load environment variables
dotenv.load_dotenv()

# Initialize FastAPI app
app = FastAPI(title="AI Agent API", description="API for interacting with AI agents")

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # In production, restrict to specific origins
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Agent instances (one per type)
agents = {}

# Request and response models
class MessageRequest(BaseModel):
    message: str
    session_id: Optional[str] = None

class MessageResponse(BaseModel):
    response: str
    session_id: str

class AgentConfigRequest(BaseModel):
    agent_type: str
    llm_provider: str = "openai"
    model: Optional[str] = None
    temperature: float = 0.7
    system_message: str = "You are a helpful AI assistant."
    enable_tools: bool = True

class AgentConfigResponse(BaseModel):
    success: bool
    message: str
    agent_type: str

# Session storage
sessions = {}

def get_or_create_agent(session_id: str, agent_type: str = "reactive"):
    """Get or create an agent for the session."""
    if session_id not in sessions:
        # Create new agent
        if agent_type not in agents:
            raise HTTPException(status_code=400, detail=f"Unknown agent type: {agent_type}")
        
        sessions[session_id] = {
            "agent": agents[agent_type],
            "conversation_memory": ConversationMemory()
        }
    
    return sessions[session_id]["agent"], sessions[session_id]["conversation_memory"]

@app.on_event("startup")
async def startup_event():
    """Initialize agents on startup."""
    # Initialize LLM
    llm = LLMInterface(provider="openai")
    
    # Initialize agents
    agents["reactive"] = ReactiveAgent(
        llm=llm,
        tools=available_tools,
        conversation_memory=None,  # Will be provided per session
        system_message="You are a helpful AI assistant."
    )
    
    agents["reflective"] = ReflectiveAgent(
        llm=llm,
        tools=available_tools,
        conversation_memory=None,  # Will be provided per session
        system_message="You are a helpful AI assistant.",
        reflection_rounds=1
    )
    
    agents["planning"] = PlanningAgent(
        llm=llm,
        tools=available_tools,
        conversation_memory=None,  # Will be provided per session
        system_message="You are a helpful AI assistant."
    )

@app.post("/message", response_model=MessageResponse)
async def process_message(request: MessageRequest):
    """Process a user message and return the agent's response."""
    # Generate session ID if not provided
    session_id = request.session_id or os.urandom(16).hex()
    
    # Get or create agent
    agent, conversation_memory = get_or_create_agent(session_id)
    
    # Set conversation memory
    agent.conversation_memory = conversation_memory
    
    # Process message
    response = agent.process_message(request.message)
    
    return MessageResponse(response=response, session_id=session_id)

@app.post("/configure", response_model=AgentConfigResponse)
async def configure_agent(request: AgentConfigRequest):
    """Configure a new agent type."""
    agent_type = request.agent_type.lower()
    
    # Initialize LLM
    llm = LLMInterface(
        provider=request.llm_provider,
        model=request.model,
        temperature=request.temperature
    )
    
    # Initialize tools
    tools = available_tools if request.enable_tools else None
    
    # Initialize agent
    try:
        if agent_type == "reactive":
            agents[agent_type] = ReactiveAgent(
                llm=llm,
                tools=tools,
                conversation_memory=None,  # Will be provided per session
                system_message=request.system_message
            )
        elif agent_type == "reflective":
            agents[agent_type] = ReflectiveAgent(
                llm=llm,
                tools=tools,
                conversation_memory=None,  # Will be provided per session
                system_message=request.system_message,
                reflection_rounds=1
            )
        elif agent_type == "planning":
            agents[agent_type] = PlanningAgent(
                llm=llm,
                tools=tools,
                conversation_memory=None,  # Will be provided per session
                system_message=request.system_message
            )
        else:
            return AgentConfigResponse(
                success=False,
                message=f"Unknown agent type: {agent_type}",
                agent_type=agent_type
            )
        
        return AgentConfigResponse(
            success=True,
            message=f"Agent {agent_type} configured successfully",
            agent_type=agent_type
        )
    except Exception as e:
        return AgentConfigResponse(
            success=False,
            message=f"Error configuring agent: {str(e)}",
            agent_type=agent_type
        )

@app.get("/sessions/{session_id}/history")
async def get_session_history(session_id: str):
    """Get conversation history for a session."""
    if session_id not in sessions:
        raise HTTPException(status_code=404, detail=f"Session {session_id} not found")
    
    conversation_memory = sessions[session_id]["conversation_memory"]
    return {"history": conversation_memory.items}

@app.delete("/sessions/{session_id}")
async def delete_session(session_id: str):
    """Delete a session."""
    if session_id not in sessions:
        raise HTTPException(status_code=404, detail=f"Session {session_id} not found")
    
    del sessions[session_id]
    return {"success": True, "message": f"Session {session_id} deleted"}

# Run with: uvicorn src.api.app:app --reload

3. Building a Simple Web Interface

A web interface provides a user-friendly way to interact with your agent.

# src/web/index.html



    
    
    
    
    


    

        
AI Agent Interface
        
        

            

                

                    
Agent Configuration
                    

                        

                            Agent Type
                            
                                Reactive Agent
                                Reflective Agent
                                Planning Agent
                            
                        
                        

                            LLM Provider
                            
                                OpenAI
                                Anthropic
                            
                        
                        

                            Temperature: 0.7
                            
                        
                        

                            System Message
                            You are a helpful AI assistant.
                        
                        

                            
                            Enable Tools
                        
                        Apply Configuration
                    
                    
                    

                    
                    

                        Clear Chat
                    
                
            
            
            

                

                    

                        

                            Hello! I'm your AI assistant. How can I help you today?
                        
                        

                            
                            
                            
                        
                    
                    
                    

                        

                            
                            Send
                        
                    
                
            
        
    
    
    

Testing and Deploying Your Agent

Before releasing your agent to users, it's important to thoroughly test it and set up a proper deployment environment.

1. Writing Tests for Your Agent

Automated tests help ensure your agent behaves as expected and catch regressions when making changes.

# tests/test_agent.py
import unittest
from unittest.mock import MagicMock, patch
import json
import os
import sys

# Add project root to path
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from src.models.llm import LLMInterface
from src.tools.base import Tool
from src.memory.conversation_memory import ConversationMemory
from src.agent.reactive_agent import ReactiveAgent

class TestReactiveAgent(unittest.TestCase):
    def setUp(self):
        # Mock LLM
        self.mock_llm = MagicMock(spec=LLMInterface)
        self.mock_llm.generate_chat_response.return_value = "This is a mock response"
        self.mock_llm.generate_text.return_value = "This is a mock text response"
        
        # Mock tool
        self.mock_tool = MagicMock(spec=Tool)
        self.mock_tool.name = "mock_tool"
        self.mock_tool.description = "A mock tool for testing"
        self.mock_tool.__call__.return_value = "Mock tool result"
        
        # Create agent
        self.agent = ReactiveAgent(
            llm=self.mock_llm,
            tools={"mock_tool": self.mock_tool},
            conversation_memory=ConversationMemory(),
            system_message="You are a test assistant"
        )
    
    def test_process_message_no_tool(self):
        # Configure mock to not use tools
        self.mock_llm.generate_text.return_value = "RESPONSE: This is a direct response"
        
        # Process message
        response = self.agent.process_message("Hello, agent!")
        
        # Check that the LLM was called correctly
        self.mock_llm.generate_text.assert_called_once()
        
        # Check response
        self.assertEqual(response, "This is a direct response")
        
        # Check conversation memory
        history = self.agent.conversation_memory.items
        self.assertEqual(len(history), 2)
        self.assertEqual(history[0]["role"], "user")
        self.assertEqual(history[0]["content"], "Hello, agent!")
        self.assertEqual(history[1]["role"], "assistant")
        self.assertEqual(history[1]["content"], "This is a direct response")
    
    def test_process_message_with_tool(self):
        # Configure mock to use a tool
        self.mock_llm.generate_text.return_value = """
        TOOL: mock_tool
        ARGS: {
            "param1": "value1",
            "param2": "value2"
        }
        """
        
        # Process message
        response = self.agent.process_message("Use the tool please")
        
        # Check that the LLM was called correctly
        self.mock_llm.generate_text.assert_called()
        
        # Check that the tool was called
        self.mock_tool.__call__.assert_called_once_with(param1="value1", param2="value2")
        
        # Check conversation memory
        history = self.agent.conversation_memory.items
        self.assertEqual(len(history), 2)
        self.assertEqual(history[0]["role"], "user")
        self.assertEqual(history[0]["content"], "Use the tool please")

class TestConversationMemory(unittest.TestCase):
    def setUp(self):
        self.memory = ConversationMemory(max_items=3)
    
    def test_add_and_get(self):
        # Add items
        id1 = self.memory.add({"role": "user", "content": "Hello"})
        id2 = self.memory.add({"role": "assistant", "content": "Hi there"})
        
        # Get items
        item1 = self.memory.get(id1)
        item2 = self.memory.get(id2)
        
        # Check items
        self.assertEqual(item1["content"], "Hello")
        self.assertEqual(item2["content"], "Hi there")
    
    def test_max_items(self):
        # Add more than max items
        self.memory.add({"role": "user", "content": "Message 1"})
        self.memory.add({"role": "assistant", "content": "Response 1"})
        self.memory.add({"role": "user", "content": "Message 2"})
        self.memory.add({"role": "assistant", "content": "Response 2"})
        
        # Check that only the last 3 items are kept
        self.assertEqual(len(self.memory.items), 3)
        self.assertEqual(self.memory.items[0]["content"], "Response 1")
        self.assertEqual(self.memory.items[1]["content"], "Message 2")
        self.assertEqual(self.memory.items[2]["content"], "Response 2")
    
    def test_clear(self):
        # Add items
        self.memory.add({"role": "user", "content": "Hello"})
        self.memory.add({"role": "assistant", "content": "Hi there"})
        
        # Clear memory
        self.memory.clear()
        
        # Check that memory is empty
        self.assertEqual(len(self.memory.items), 0)

if __name__ == "__main__":
    unittest.main()

2. Setting Up Continuous Integration

Continuous Integration (CI) helps automate testing and deployment of your agent.

# .github/workflows/ci.yml
name: CI

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

jobs:
  test:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.10'
    
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
        pip install pytest pytest-cov
    
    - name: Run tests
      run: |
        pytest tests/ --cov=src
    
    - name: Upload coverage report
      uses: codecov/codecov-action@v3
      with:
        file: ./coverage.xml
        fail_ci_if_error: true

  lint:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.10'
    
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install flake8
    
    - name: Lint with flake8
      run: |
        flake8 src/ tests/ --count --select=E9,F63,F7,F82 --show-source --statistics

3. Deploying Your Agent

There are several options for deploying your agent, depending on your requirements and resources.

Docker Deployment Example

# Dockerfile
FROM python:3.10-slim

WORKDIR /app

# Copy requirements first for better caching
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Expose port
EXPOSE 8000

# Run the application
CMD ["uvicorn", "src.api.app:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3'

services:
  ai-agent:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
    volumes:
      - ./data:/app/data
    restart: unless-stopped

Next Steps in Your AI Journey

Now that you've built a complete AI agent from scratch, you're ready to explore more advanced agent architectures and capabilities.

In the next section, we'll dive into Advanced Agentic AI Systems, where you'll learn how to build sophisticated multi-agent systems and specialized agent architectures for complex tasks.