Stage 2: Core Concepts

Stage 2: Core Concepts

Now that you understand what AI agents are and have the tools set up, let us dive into how they actually work. This stage covers the fundamental patterns and concepts that power every AI agent, from simple chatbots to complex autonomous systems.

Module 2.1: How AI Agents Think (5 hours)

Learning Objectives

By the end of this module, you will be able to:

1. Explain the ReAct pattern (Reasoning + Acting)
2. Trace an agent's decision-making process step by step
3. Identify when agents succeed and when they fail
4. Design prompts that guide agent reasoning

2.1.1 The ReAct Pattern

ReAct stands for Reasoning and Acting. It is the most influential pattern in AI agent design, introduced by researchers at Princeton and Google in 2022. The insight is simple but powerful: interleaving reasoning with action produces better results than either alone.

The Three Steps:

1. Thought: The agent reasons about what to do next. This is explicit reasoning that we can see and debug.

2. Action: The agent chooses and executes a tool. This is how agents interact with the world.

3. Observation: The agent sees the result of its action. This new information feeds back into the next thought.

2.1.2 Why ReAct Works

Traditional LLMs generate responses in one go. They cannot stop, check something, and continue. ReAct allows the agent to:

1. Break complex problems into steps: Instead of trying to solve everything at once, tackle one piece at a time

2. Ground responses in facts: Each action brings real information into the reasoning process

3. Self-correct: If an action reveals unexpected information, the agent can adjust its approach

4. Show its work: The explicit thought process makes debugging much easier

2.1.3 When Agents Fail

Understanding failure modes helps you build more robust systems.

Module 2.2: Tools and Actions (5 hours)

Learning Objectives

By the end of this module, you will be able to:

1. Define what tools are in the context of AI agents
2. Design tool schemas with clear descriptions and parameters
3. Implement input validation for tools
4. Handle tool errors gracefully

2.2.1 What Are Tools?

Tools are functions that agents can call to interact with the world. They are the bridge between thinking and doing.

Common Tool Categories:

2.2.2 Designing Good Tool Schemas

The schema tells the agent what a tool does and how to use it. A well-designed schema reduces errors and improves tool selection.

"""
Tool Schema Design
==================
How to define tools that agents can use effectively.
"""

from dataclasses import dataclass
from typing import Any, Dict, Callable


@dataclass
class Tool:
    """
    A tool that an AI agent can use.
    
    Attributes:
        name: Unique identifier (use snake_case)
        description: What the tool does (be specific!)
        function: The Python function to execute
        parameters: JSON Schema defining expected inputs
    """
    name: str
    description: str
    function: Callable
    parameters: Dict[str, Any]
    
    def to_prompt_format(self) -> str:
        """Format for inclusion in agent prompts."""
        params = ", ".join(
            f"{k}: {v.get('description', 'no description')}"
            for k, v in self.parameters.get("properties", {}).items()
        )
        return f"- {self.name}({params}): {self.description}"


# Example: Well-designed tool

def search_web(query: str, num_results: int = 5) -> dict:
    """Search the web and return results."""
    # Implementation would go here
    return {"results": []}


search_tool = Tool(
    name="search_web",
    description="Search the internet for current information. Use for facts, news, or data you do not know. Returns titles, snippets, and URLs.",
    function=search_web,
    parameters={
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "The search query. Be specific and include relevant keywords."
            },
            "num_results": {
                "type": "integer",
                "description": "Number of results to return (1-10)",
                "default": 5
            }
        },
        "required": ["query"]
    }
)

Tool Description Best Practices:

1. Be specific: "Search the internet for current information" is better than "Search"

2. Explain when to use it: "Use for facts, news, or data you do not know"

3. Describe outputs: "Returns titles, snippets, and URLs"

4. Mention limitations: If the tool only works for certain inputs, say so

2.2.3 Tool Input Validation

Never trust agent-generated inputs. Always validate.

"""
Tool Input Validation
=====================
Validate inputs before executing tools.
"""

from typing import Any, Optional, Tuple
import re


class ToolInputValidator:
    """Validates tool inputs before execution."""
    
    @staticmethod
    def validate_search_query(query: Any) -> Tuple[bool, Optional[str]]:
        """
        Validate a search query.
        
        Args:
            query: The query to validate
            
        Returns:
            Tuple of (is_valid, error_message)
        """
        # Must be a string
        if not isinstance(query, str):
            return False, "Query must be a string"
        
        # Must not be empty
        if not query.strip():
            return False, "Query cannot be empty"
        
        # Length limits
        if len(query) > 500:
            return False, "Query too long (max 500 characters)"
        
        # No potential injection patterns
        suspicious = [
            r"<script",
            r"javascript:",
            r"data:",
            r"vbscript:",
        ]
        for pattern in suspicious:
            if re.search(pattern, query, re.IGNORECASE):
                return False, "Query contains suspicious content"
        
        return True, None
    
    @staticmethod
    def validate_file_path(path: Any, allowed_dirs: list) -> Tuple[bool, Optional[str]]:
        """
        Validate a file path is within allowed directories.
        
        Args:
            path: The file path to validate
            allowed_dirs: List of allowed directory prefixes
            
        Returns:
            Tuple of (is_valid, error_message)
        """
        import os
        
        if not isinstance(path, str):
            return False, "Path must be a string"
        
        # Normalise path to prevent directory traversal
        normalised = os.path.normpath(os.path.abspath(path))
        
        # Check if within allowed directories
        for allowed in allowed_dirs:
            if normalised.startswith(os.path.abspath(allowed)):
                return True, None
        
        return False, f"Path not in allowed directories: {allowed_dirs}"

Module 2.3: Memory and Context (5 hours)

Learning Objectives

By the end of this module, you will be able to:

1. Distinguish between short-term and long-term memory
2. Implement conversation history management
3. Understand context window limitations
4. Use vector databases for semantic memory

2.3.1 Types of Agent Memory

🎯 Interactive: Memory Architecture Explorer

Explore the different types of agent memory through this interactive tool. Test your understanding of when to use short-term, long-term, and external memory for different queries.

2.3.2 Managing Conversation History

As conversations grow, they eventually exceed the context window. You need strategies to handle this.

"""
Conversation Memory Management
==============================
Strategies for maintaining conversation context.
"""

from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime


@dataclass
class Message:
    """A single message in conversation history."""
    role: str  # "user", "assistant", or "system"
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    token_count: int = 0


class ConversationMemory:
    """
    Manages conversation history with context window awareness.
    """
    
    def __init__(self, max_tokens: int = 4000, preserve_system: bool = True):
        """
        Initialise conversation memory.
        
        Args:
            max_tokens: Maximum tokens to keep in history
            preserve_system: Always keep the system message
        """
        self.max_tokens = max_tokens
        self.preserve_system = preserve_system
        self.messages: List[Message] = []
        self.system_message: Optional[Message] = None
    
    def add_message(self, role: str, content: str) -> None:
        """Add a message to history."""
        # Estimate token count (roughly 4 chars per token)
        token_count = len(content) // 4
        
        message = Message(
            role=role,
            content=content,
            token_count=token_count
        )
        
        if role == "system":
            self.system_message = message
        else:
            self.messages.append(message)
        
        # Prune if needed
        self._prune_to_fit()
    
    def _prune_to_fit(self) -> None:
        """Remove old messages to fit within token limit."""
        current_tokens = sum(m.token_count for m in self.messages)
        
        if self.system_message:
            current_tokens += self.system_message.token_count
        
        # Remove oldest messages until we fit
        while current_tokens > self.max_tokens and len(self.messages) > 2:
            removed = self.messages.pop(0)
            current_tokens -= removed.token_count
    
    def get_messages(self) -> List[dict]:
        """Get messages in format suitable for LLM."""
        result = []
        
        if self.system_message:
            result.append({
                "role": "system",
                "content": self.system_message.content
            })
        
        for msg in self.messages:
            result.append({
                "role": msg.role,
                "content": msg.content
            })
        
        return result
    
    def summarise_and_compress(self, summariser_fn) -> None:
        """
        Summarise older messages to save tokens.
        
        Args:
            summariser_fn: Function that takes messages and returns summary
        """
        if len(self.messages) < 10:
            return
        
        # Take oldest 70% of messages
        split_point = int(len(self.messages) * 0.7)
        old_messages = self.messages[:split_point]
        recent_messages = self.messages[split_point:]
        
        # Summarise old messages
        old_content = "\n".join(
            f"{m.role}: {m.content}" 
            for m in old_messages
        )
        summary = summariser_fn(old_content)
        
        # Replace with summary
        summary_message = Message(
            role="system",
            content=f"[Summary of earlier conversation: {summary}]",
            token_count=len(summary) // 4
        )
        
        self.messages = [summary_message] + recent_messages

2.3.3 Vector Databases for Semantic Memory

When you need to remember things across sessions or search through large amounts of information, vector databases are essential.

Module 2.4: Design Patterns (5 hours)

Learning Objectives

By the end of this module, you will be able to:

1. Apply the ReAct pattern to complex problems
2. Implement Plan-and-Execute patterns
3. Choose appropriate patterns for different use cases
4. Combine patterns for sophisticated behaviour

2.4.1 Pattern Overview

2.4.2 Plan-and-Execute Pattern

For complex tasks, planning before acting often works better than interleaved reasoning.

2.4.3 Reflection Pattern

The agent reviews and improves its own output before presenting it.

"""
Reflection Pattern Implementation
==================================
Agent reviews and improves its output.
"""


def generate_with_reflection(
    query: str,
    generator_fn,
    critic_fn,
    max_iterations: int = 3
) -> str:
    """
    Generate content with self-reflection loop.
    
    Args:
        query: The user's request
        generator_fn: Function that generates initial content
        critic_fn: Function that critiques and suggests improvements
        max_iterations: Maximum improvement iterations
        
    Returns:
        Final improved content
    """
    # Generate initial response
    content = generator_fn(query)
    
    for i in range(max_iterations):
        # Get critique
        critique = critic_fn(
            query=query,
            content=content
        )
        
        # Check if good enough
        if critique.get("is_satisfactory", False):
            break
        
        # Improve based on feedback
        improvements = critique.get("suggestions", [])
        content = generator_fn(
            f"{query}\n\n"
            f"Previous attempt: {content}\n\n"
            f"Improvements needed: {improvements}\n\n"
            f"Please provide an improved version."
        )
    
    return content

Module 2.5: Architecture Fundamentals (5 hours)

Learning Objectives

By the end of this module, you will be able to:

1. Design agent state management
2. Implement control flows and routing
3. Handle errors and edge cases
4. Build robust agent architectures

2.5.1 Agent State Management

State is everything your agent needs to remember to complete its task.

"""
Agent State Management
======================
Immutable state for reliable agent operation.
"""

from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
from datetime import datetime
from enum import Enum


class AgentStatus(Enum):
    """Current state of the agent."""
    IDLE = "idle"
    THINKING = "thinking"
    ACTING = "acting"
    WAITING = "waiting"
    COMPLETE = "complete"
    ERROR = "error"


@dataclass(frozen=True)
class AgentState:
    """
    Immutable agent state.
    
    Using frozen=True ensures state cannot be accidentally modified.
    Each operation creates a new state object.
    """
    messages: tuple = field(default_factory=tuple)
    status: AgentStatus = AgentStatus.IDLE
    current_step: int = 0
    max_steps: int = 10
    tool_results: tuple = field(default_factory=tuple)
    error: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def with_message(self, role: str, content: str) -> "AgentState":
        """Return new state with added message."""
        new_messages = self.messages + ({"role": role, "content": content},)
        return AgentState(
            messages=new_messages,
            status=self.status,
            current_step=self.current_step,
            max_steps=self.max_steps,
            tool_results=self.tool_results,
            error=self.error,
            metadata=self.metadata.copy()
        )
    
    def with_status(self, status: AgentStatus) -> "AgentState":
        """Return new state with updated status."""
        return AgentState(
            messages=self.messages,
            status=status,
            current_step=self.current_step,
            max_steps=self.max_steps,
            tool_results=self.tool_results,
            error=self.error,
            metadata=self.metadata.copy()
        )
    
    def with_tool_result(self, tool: str, result: Any) -> "AgentState":
        """Return new state with added tool result."""
        new_results = self.tool_results + ({"tool": tool, "result": result},)
        return AgentState(
            messages=self.messages,
            status=self.status,
            current_step=self.current_step + 1,
            max_steps=self.max_steps,
            tool_results=new_results,
            error=self.error,
            metadata=self.metadata.copy()
        )
    
    def is_complete(self) -> bool:
        """Check if agent has finished."""
        return self.status in (AgentStatus.COMPLETE, AgentStatus.ERROR)
    
    def should_continue(self) -> bool:
        """Check if agent should keep going."""
        return (
            not self.is_complete() 
            and self.current_step < self.max_steps
        )

2.5.2 Error Handling

Agents will fail. The question is how gracefully.

Stage 2 Assessment

Summary

In this stage, you have learned:

1. The ReAct pattern: How agents interleave reasoning with action

2. Tools and actions: How to design, validate, and execute agent tools

3. Memory management: Short-term, long-term, and semantic memory

4. Design patterns: When to use ReAct, Plan-and-Execute, and Reflection

5. Architecture fundamentals: State management and error handling