Agent Fundamentals

For years, our primary interaction with AI in conversational interfaces has been through chatbots—rule-based systems that react to user input with predefined responses. While useful for narrow tasks, these systems are fundamentally limited in their ability to handle complex, multi-step problems or adapt to unexpected scenarios.

The advent of powerful Large Language Models (LLMs) has enabled a significant paradigm shift: the rise of AI agents. These agents move beyond simple responses to become proactive, autonomous entities capable of pursuing goals, making decisions, and completing complex tasks in dynamic environments. This chapter explores the fundamental architecture that makes this transformation possible.

1. The Evolution: Beyond Reactive Chatbots

Traditional Chatbots: Rule-Bound and Reactive

Traditional chatbots operate on a simple stimulus-response model. They wait for user input, match it against predefined rules or decision trees, and return a programmed response. While this works well for FAQ systems or simple task automation, it breaks down quickly when faced with complexity.

Example: Traditional Chatbot

# Simple rule-based chatbot
class SimpleChatbot:
    def __init__(self):
        self.rules = {
            "hello": "Hi! How can I help you?",
            "weather": "I'm sorry, I can't check the weather.",
            "bye": "Goodbye! Have a great day!"
        }

    def respond(self, user_input):
        # Convert to lowercase and check for keywords
        user_input = user_input.lower()

        for keyword, response in self.rules.items():
            if keyword in user_input:
                return response

        # Default response if no match
        return "I'm sorry, I don't understand. Can you rephrase?"

# Usage
bot = SimpleChatbot()
print(bot.respond("Hello there!"))  # "Hi! How can I help you?"
print(bot.respond("What's the weather like?"))  # "I'm sorry, I can't check the weather."
print(bot.respond("Tell me a joke"))  # "I'm sorry, I don't understand. Can you rephrase?"

AI Agents: Proactive, Autonomous, and Goal-Oriented

AI agents are designed to be more like intelligent assistants or project managers. They are equipped with the ability to understand goals, plan steps, execute actions using external tools, and learn from the outcomes. They don't just respond—they act.

2. The Core Loop: Observe, Think, Act (ReAct)

The fundamental difference enabling this shift is the "Observe, Think, Act" (OTA) loop, often implemented through frameworks like ReAct (Reasoning and Acting). This iterative cycle allows agents to interact dynamically with their environment, learning and adapting as they go.

    ┌─────────────────────────────────────────────────────┐
    │                    AGENT LOOP                        │
    └─────────────────────────────────────────────────────┘

    ┌─────────────┐       ┌─────────────┐       ┌─────────────┐
    │             │       │             │       │             │
    │   OBSERVE   │──────▶│    THINK    │──────▶│     ACT     │
    │             │       │             │       │             │
    └─────────────┘       └─────────────┘       └─────────────┘
         ▲                                             │
         │                                             │
         │                 FEEDBACK                    │
         └─────────────────────────────────────────────┘

    OBSERVE: Perceive environment, analyze user input,
             process tool outputs, gather information

    THINK:   Reason about current state, decompose goals,
             plan next steps, select appropriate tools

    ACT:     Execute actions via tools, generate responses,
             update internal state, change environment
                    

Step 1: Observe

The agent perceives its environment and gathers relevant information. This step provides the agent with the current state of the world and contextual data needed for decision-making.

Step 2: Think (Reason/Plan)

Based on its observations, the agent uses its LLM "brain" to reason about the current situation and formulate a plan. This is where the agent strategizes and decides what to do next.

Step 3: Act

The agent executes the planned action. This can involve calling external tools, generating a response for the user, or updating its internal state. This step changes the environment or provides information that feeds back into the next observation.

After acting, the loop restarts with Observe, allowing the agent to evaluate the outcome of its action, learn from it, and adjust its subsequent thinking and actions. This continuous feedback mechanism is what gives AI agents their adaptability and problem-solving capabilities.

3. Implementing the ReAct Pattern

Let's build a simple agent that uses the ReAct pattern to answer questions that require accessing external information. This agent will be able to search the web and perform calculations—capabilities beyond what a standard LLM can do alone.

Example 1: Simple ReAct Agent

from openai import OpenAI
import json
import requests
from datetime import datetime

class SimpleReActAgent:
    """A basic ReAct agent with web search and calculator tools."""

    def __init__(self, api_key):
        self.client = OpenAI(api_key=api_key)
        self.model = "gpt-4"
        self.max_iterations = 5
        self.conversation_history = []

        # Define available tools
        self.tools = [
            {
                "type": "function",
                "function": {
                    "name": "web_search",
                    "description": "Search the web for current information",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {
                                "type": "string",
                                "description": "The search query"
                            }
                        },
                        "required": ["query"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "calculator",
                    "description": "Perform mathematical calculations",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "expression": {
                                "type": "string",
                                "description": "Mathematical expression to evaluate"
                            }
                        },
                        "required": ["expression"]
                    }
                }
            }
        ]

    def web_search(self, query):
        """Simulate a web search (in production, use real API like Tavily or Google)."""
        # In production: return requests.post("https://api.tavily.com/search", json={"query": query})
        return f"Search results for '{query}': [Mock result - OpenAI was founded in 2015 by Sam Altman and others]"

    def calculator(self, expression):
        """Safely evaluate mathematical expressions."""
        try:
            # Only allow basic math operations (security consideration)
            allowed_chars = set("0123456789+-*/()., ")
            if not all(c in allowed_chars for c in expression):
                return "Error: Invalid characters in expression"

            result = eval(expression)
            return f"Result: {result}"
        except Exception as e:
            return f"Error: {str(e)}"

    def execute_tool(self, tool_name, arguments):
        """Execute the specified tool with given arguments."""
        if tool_name == "web_search":
            return self.web_search(arguments["query"])
        elif tool_name == "calculator":
            return self.calculator(arguments["expression"])
        else:
            return f"Error: Unknown tool '{tool_name}'"

    def run(self, user_query):
        """Execute the ReAct loop to answer the user's query."""
        print(f"\n{'='*60}")
        print(f"USER QUERY: {user_query}")
        print(f"{'='*60}\n")

        # Initialize conversation with user query
        self.conversation_history = [
            {"role": "system", "content": "You are a helpful AI agent. Use the available tools to answer questions accurately. Think step-by-step and explain your reasoning."},
            {"role": "user", "content": user_query}
        ]

        iteration = 0
        while iteration < self.max_iterations:
            iteration += 1
            print(f"--- ITERATION {iteration} ---\n")

            # THINK: Get LLM's response (may include tool calls)
            response = self.client.chat.completions.create(
                model=self.model,
                messages=self.conversation_history,
                tools=self.tools,
                tool_choice="auto"
            )

            message = response.choices[0].message

            # Check if the agent wants to use a tool (ACT)
            if message.tool_calls:
                # Agent decided to use a tool
                self.conversation_history.append(message)

                for tool_call in message.tool_calls:
                    tool_name = tool_call.function.name
                    arguments = json.loads(tool_call.function.arguments)

                    print(f"THOUGHT: I need to use the {tool_name} tool")
                    print(f"ACTION: {tool_name}({json.dumps(arguments)})")

                    # Execute the tool (ACT)
                    result = self.execute_tool(tool_name, arguments)

                    print(f"OBSERVATION: {result}\n")

                    # Add tool result to conversation (OBSERVE)
                    self.conversation_history.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "name": tool_name,
                        "content": result
                    })
            else:
                # Agent has reached a conclusion
                print(f"FINAL ANSWER: {message.content}\n")
                return message.content

        return "Maximum iterations reached without conclusion."

# Usage
agent = SimpleReActAgent(api_key="your-api-key")
result = agent.run("When was OpenAI founded and how many years ago was that?")

============================================================
USER QUERY: When was OpenAI founded and how many years ago was that?
============================================================

--- ITERATION 1 ---

THOUGHT: I need to use the web_search tool
ACTION: web_search({"query": "when was OpenAI founded"})
OBSERVATION: Search results for 'when was OpenAI founded': [Mock result - OpenAI was founded in 2015 by Sam Altman and others]

--- ITERATION 2 ---

THOUGHT: I need to use the calculator tool
ACTION: calculator({"expression": "2025 - 2015"})
OBSERVATION: Result: 10

--- ITERATION 3 ---

FINAL ANSWER: OpenAI was founded in 2015. That was 10 years ago (as of 2025).
                    

Key Observations

4. Advanced ReAct Patterns

Self-Correction and Retry Logic

Production agents need to handle errors gracefully and retry failed operations. Let's enhance our agent with self-correction capabilities.

Example 2: Agent with Self-Correction

class ResilientReActAgent(SimpleReActAgent):
    """Enhanced agent with error handling and retry logic."""

    def __init__(self, api_key):
        super().__init__(api_key)
        self.max_retries = 3

    def execute_tool_with_retry(self, tool_name, arguments, attempt=1):
        """Execute tool with automatic retry on failure."""
        try:
            result = self.execute_tool(tool_name, arguments)

            # Check if result indicates an error
            if "Error:" in result and attempt < self.max_retries:
                print(f"⚠️  Tool execution failed (attempt {attempt}/{self.max_retries})")
                print(f"   Retrying with adjusted parameters...\n")

                # Simple retry strategy: wait and try again
                import time
                time.sleep(1 * attempt)  # Exponential backoff

                return self.execute_tool_with_retry(tool_name, arguments, attempt + 1)

            return result
        except Exception as e:
            if attempt < self.max_retries:
                print(f"⚠️  Exception during tool execution: {str(e)}")
                print(f"   Retrying (attempt {attempt + 1}/{self.max_retries})...\n")
                import time
                time.sleep(1 * attempt)
                return self.execute_tool_with_retry(tool_name, arguments, attempt + 1)
            else:
                return f"Error: Tool execution failed after {self.max_retries} attempts - {str(e)}"

    def run_with_verification(self, user_query):
        """Run agent with result verification step."""
        result = self.run(user_query)

        # Add verification step
        verification_prompt = f"""
        Please verify your answer: {result}

        Is this answer:
        1. Factually accurate based on the tools you used?
        2. Complete (fully addresses the user's question)?
        3. Clear and well-explained?

        If not, explain what needs correction.
        """

        self.conversation_history.append({
            "role": "user",
            "content": verification_prompt
        })

        verification = self.client.chat.completions.create(
            model=self.model,
            messages=self.conversation_history
        )

        verification_result = verification.choices[0].message.content

        print(f"\n{'='*60}")
        print(f"VERIFICATION: {verification_result}")
        print(f"{'='*60}\n")

        return result

# Usage
resilient_agent = ResilientReActAgent(api_key="your-api-key")
result = resilient_agent.run_with_verification("What is 15% of 850, and is that more than 100?")

Chain-of-Thought with Tool Use

We can enhance reasoning quality by explicitly prompting the agent to show its step-by-step thinking before acting.

Example 3: Chain-of-Thought ReAct Agent

class ChainOfThoughtAgent(SimpleReActAgent):
    """Agent that explicitly shows its reasoning process."""

    def __init__(self, api_key):
        super().__init__(api_key)

        # Enhanced system prompt with CoT instructions
        self.system_prompt = """You are a helpful AI agent that thinks step-by-step.

For each user query:
1. First, analyze what information you need
2. Explain your reasoning about which tools to use
3. After getting results, explain how they relate to the answer
4. Provide a clear, complete final answer

Always show your thought process before acting."""

    def run(self, user_query):
        """Execute with explicit chain-of-thought reasoning."""
        print(f"\n{'='*60}")
        print(f"USER QUERY: {user_query}")
        print(f"{'='*60}\n")

        self.conversation_history = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_query}
        ]

        iteration = 0
        while iteration < self.max_iterations:
            iteration += 1
            print(f"--- ITERATION {iteration} ---\n")

            # First, ask the agent to explain its thinking
            thinking_prompt = "Before taking action, explain your reasoning: What information do you need and why?"

            self.conversation_history.append({
                "role": "user",
                "content": thinking_prompt
            })

            # Get agent's reasoning
            thinking_response = self.client.chat.completions.create(
                model=self.model,
                messages=self.conversation_history
            )

            thought = thinking_response.choices[0].message.content
            print(f"REASONING: {thought}\n")

            self.conversation_history.append(thinking_response.choices[0].message)

            # Now get the actual response with potential tool calls
            response = self.client.chat.completions.create(
                model=self.model,
                messages=self.conversation_history,
                tools=self.tools,
                tool_choice="auto"
            )

            message = response.choices[0].message

            if message.tool_calls:
                self.conversation_history.append(message)

                for tool_call in message.tool_calls:
                    tool_name = tool_call.function.name
                    arguments = json.loads(tool_call.function.arguments)

                    print(f"ACTION: {tool_name}({json.dumps(arguments)})")
                    result = self.execute_tool(tool_name, arguments)
                    print(f"OBSERVATION: {result}\n")

                    self.conversation_history.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "name": tool_name,
                        "content": result
                    })
            else:
                print(f"FINAL ANSWER: {message.content}\n")
                return message.content

        return "Maximum iterations reached."

# Usage
cot_agent = ChainOfThoughtAgent(api_key="your-api-key")
result = cot_agent.run("If a company's revenue grew from $1.2M to $1.8M, what was the percentage growth?")

5. When to Use Agents vs Chatbots

Not every problem requires an agent. Understanding when to use autonomous agents versus simpler chatbots is crucial for building efficient, cost-effective systems.

Hybrid Approach: Router Pattern

In production, you can use a router to decide whether to use a simple chatbot or invoke a full agent based on query complexity.

Example 4: Smart Router

class SmartRouter:
    """Routes queries to either a simple chatbot or full agent."""

    def __init__(self, api_key):
        self.client = OpenAI(api_key=api_key)
        self.simple_bot = SimpleChatbot()
        self.agent = SimpleReActAgent(api_key)

    def classify_query_complexity(self, query):
        """Determine if query needs a full agent or simple response."""
        classification_prompt = f"""Classify this query's complexity:

Query: {query}

Respond with only one word:
- SIMPLE: Can be answered with basic information, no tools needed
- COMPLEX: Requires research, calculations, or multi-step reasoning

Classification:"""

        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",  # Use cheaper model for classification
            messages=[{"role": "user", "content": classification_prompt}],
            max_tokens=10,
            temperature=0
        )

        classification = response.choices[0].message.content.strip().upper()
        return classification

    def route_query(self, query):
        """Route query to appropriate handler."""
        complexity = self.classify_query_complexity(query)

        print(f"Query Classification: {complexity}\n")

        if complexity == "SIMPLE":
            print("→ Routing to simple chatbot (fast, low-cost)")
            return self.simple_bot.respond(query)
        else:
            print("→ Routing to full agent (capable, higher-cost)")
            return self.agent.run(query)

# Usage
router = SmartRouter(api_key="your-api-key")

# Simple query
router.route_query("Hello!")  # → Simple chatbot

# Complex query
router.route_query("Find the current price of Bitcoin and calculate how much 0.5 BTC is worth in USD")  # → Full agent

6. Production Best Practices

1. Set Clear Boundaries

# Define what your agent can and cannot do
agent_capabilities = {
    "can_do": [
        "Search the web for current information",
        "Perform calculations",
        "Query our internal database",
        "Send notifications to users"
    ],
    "cannot_do": [
        "Access external user accounts without permission",
        "Make financial transactions",
        "Modify production databases",
        "Share sensitive company data"
    ]
}

# Include in system prompt
system_prompt = f"""You are a helpful AI agent with these capabilities:
{agent_capabilities['can_do']}

You MUST NOT:
{agent_capabilities['cannot_do']}

If a user asks you to do something outside your capabilities, politely decline and explain."""

2. Implement Guardrails

class SafeAgent(SimpleReActAgent):
    """Agent with safety guardrails."""

    def __init__(self, api_key):
        super().__init__(api_key)
        self.forbidden_tools = ["delete_database", "transfer_money"]
        self.requires_approval = ["send_email", "post_to_social_media"]

    def execute_tool(self, tool_name, arguments):
        """Execute tool with safety checks."""
        # Check forbidden tools
        if tool_name in self.forbidden_tools:
            return f"Error: Tool '{tool_name}' is forbidden for safety reasons"

        # Check if human approval needed
        if tool_name in self.requires_approval:
            approval = self.request_human_approval(tool_name, arguments)
            if not approval:
                return "Action cancelled: Human approval not granted"

        # Proceed with execution
        return super().execute_tool(tool_name, arguments)

    def request_human_approval(self, tool_name, arguments):
        """Request human approval for sensitive actions."""
        print(f"\n{'='*60}")
        print(f"⚠️  APPROVAL REQUIRED")
        print(f"Tool: {tool_name}")
        print(f"Arguments: {json.dumps(arguments, indent=2)}")
        print(f"{'='*60}")

        approval = input("Approve this action? (yes/no): ").strip().lower()
        return approval == "yes"

3. Monitor and Log Everything

import logging
from datetime import datetime

class MonitoredAgent(SimpleReActAgent):
    """Agent with comprehensive logging."""

    def __init__(self, api_key):
        super().__init__(api_key)

        # Set up logging
        logging.basicConfig(
            filename=f'agent_log_{datetime.now().strftime("%Y%m%d")}.log',
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)

    def run(self, user_query):
        """Run agent with comprehensive logging."""
        self.logger.info(f"New query received: {user_query}")

        try:
            result = super().run(user_query)
            self.logger.info(f"Query completed successfully")
            return result
        except Exception as e:
            self.logger.error(f"Query failed: {str(e)}")
            raise

    def execute_tool(self, tool_name, arguments):
        """Execute tool with logging."""
        self.logger.info(f"Tool execution: {tool_name} with {arguments}")

        start_time = datetime.now()
        result = super().execute_tool(tool_name, arguments)
        end_time = datetime.now()

        duration = (end_time - start_time).total_seconds()
        self.logger.info(f"Tool completed in {duration:.2f}s: {result[:100]}...")

        return result

4. Handle Errors Gracefully

class RobustAgent(SimpleReActAgent):
    """Agent with comprehensive error handling."""

    def run(self, user_query):
        """Run agent with error recovery."""
        try:
            return super().run(user_query)
        except Exception as e:
            # Log the error
            print(f"Error: {str(e)}")

            # Attempt recovery
            recovery_prompt = f"""
            An error occurred while processing your request: {str(e)}

            Please provide a simplified or alternative answer based on your
            internal knowledge, without using tools.
            """

            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "You are a helpful AI assistant."},
                    {"role": "user", "content": user_query},
                    {"role": "user", "content": recovery_prompt}
                ]
            )

            return f"[Degraded Mode] {response.choices[0].message.content}"

Key Takeaways

Further Reading

• ReAct: Synergizing Reasoning and Acting in Language Models (2022)
• LangChain Agents Documentation
• OpenAI Function Calling Guide
• Anthropic Tool Use and Computer Use