AI Tutorial · August 7, 2026 - Building AI Agent Workflows

Building AI Agent Automation Workflows: A Practical Guide

AI agents are no longer a futuristic concept — they are here, and they are transforming how we automate tasks, process information, and build intelligent systems. In this tutorial, we’ll walk through the practical steps of building your own AI agent workflow from scratch using Python and OpenAI’s API.

Background

What Are AI Agents and Why They Matter

An AI agent is an autonomous program that can perceive its environment, reason about goals, take actions, and learn from results. Unlike traditional scripts that follow rigid if-else logic, agents use large language models (LLMs) as their “brain” to make dynamic decisions.

Why agents matter:

• They handle unstructured tasks that rule-based systems cannot
• They adapt to changing inputs without reprogramming
• They combine multiple tools and data sources seamlessly
• They can work autonomously on complex multi-step problems

The Rise of Agentic AI

The shift from “chatbots” to “agents” represents a fundamental leap. Instead of simply generating text, modern AI agents:

• Use tools — call APIs, search the web, run code
• Maintain memory — remember context across interactions
• Plan ahead — decompose complex tasks into sub-steps
• Self-correct — retry failures with adjusted strategies

Frameworks like LangChain, CrewAI, and AutoGen have popularized these patterns, but understanding the core mechanics is essential for building robust, production-ready systems.

Step 1: Understanding AI Agent Core Concepts

Every AI agent is built on four fundamental components:

Agent = LLM + Tools + Memory + Planning

The Four Pillars

Component	Role	Example
LLM	Reasoning engine	GPT-4, Claude, Llama
Tools	Actions the agent can take	Web search, calculator, file I/O
Memory	Context and state	Conversation history, vector store
Planning	Strategy for task decomposition	ReAct, Chain-of-Thought, function calling

Common Patterns

ReAct (Reasoning + Acting): The agent alternates between thinking (“I need to find the current weather”) and acting (calling a weather API), then observing the result before the next thought.

Function Calling: The LLM outputs structured JSON to invoke predefined functions. The runtime executes the function and returns the result back to the LLM.

Tool Use: A generalization of function calling where tools can be APIs, databases, code interpreters, or even other agents.

Step 2: Environment Setup

Let’s get our environment ready. We’ll use Python 3.10+ and the OpenAI SDK.

Prerequisites

# Python 3.10+ required
python3 --version

# Install the OpenAI SDK
pip install openai

# Optional but recommended
pip install python-dotenv httpx

API Key Setup

Create a .env file in your project root:

OPENAI_API_KEY=sk-your-api-key-here

Then load it in your code:

import os
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
    raise ValueError("OPENAI_API_KEY not found in .env")

Project Structure

agent-tutorial/
├── .env
├── agent.py          # Main agent loop
├── tools.py          # Tool definitions
├── memory.py         # Memory management
├── workflow.py       # Workflow orchestration
└── requirements.txt  # Dependencies

Step 3: Building Your First Agent

Let’s build a simple agent that can search the web and answer questions using function calling.

Defining the Tools

First, we define what tools our agent can use. Each tool is a function with a JSON schema that the LLM can reference.

# tools.py
import json
import httpx
from typing import Any

def web_search(query: str) -> str:
    """Search the web for information."""
    # Simulated web search — replace with an actual API like SerpAPI or Tavily
    results = {
        "python": "Python is a high-level programming language created by Guido van Rossum.",
        "ai agents": "AI agents are autonomous programs that use LLMs to make decisions.",
        "weather": "The weather varies by location. Please specify a city."
    }
    return results.get(query.lower(), f"No results found for '{query}'.")

def calculator(expression: str) -> float:
    """Evaluate a mathematical expression."""
    return eval(expression)

# Tool registry
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "web_search",
            "description": "Search the web for information",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The search query"
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculator",
            "description": "Evaluate a mathematical expression",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "Math expression like '2 + 2' or '3 * 7'"
                    }
                },
                "required": ["expression"]
            }
        }
    }
]

# Tool dispatch map
TOOL_MAP = {
    "web_search": web_search,
    "calculator": calculator,
}

The Agent Loop

The agent loop follows the perceive → think → act → observe cycle:

# agent.py
import json
import os
from openai import OpenAI
from tools import TOOLS, TOOL_MAP

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def run_agent(user_message: str, max_turns: int = 5) -> str:
    """Run the agent loop: perceive → think → act → observe."""
    messages = [{"role": "user", "content": user_message}]
    
    for turn in range(max_turns):
        # THINK: Ask the LLM what to do
        response = client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )
        
        message = response.choices[0].message
        
        # If no tool call, the LLM has a final answer
        if not message.tool_calls:
            return message.content
        
        # ACT: Execute each tool call
        messages.append(message)
        
        for tool_call in message.tool_calls:
            function_name = tool_call.function.name
            arguments = json.loads(tool_call.function.arguments)
            
            # Execute the tool
            function_to_call = TOOL_MAP[function_name]
            result = function_to_call(**arguments)
            
            # OBSERVE: Add the result back
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": str(result),
            })
        
        # Continue the loop — the LLM sees the observations and decides next action
    
    return "Agent reached maximum turns without a final answer."

# Example usage
if __name__ == "__main__":
    result = run_agent("What is Python and what is 2 + 2?")
    print(result)

How the Agent Loop Works

1. Perceive: The agent receives the user message and its environment (tool results, memory)
2. Think: The LLM decides whether to answer directly or use a tool
3. Act: If a tool is called, the runtime executes it
4. Observe: The tool result is fed back to the LLM
5. Repeat: The cycle continues until the LLM produces a final answer

Step 4: Adding Tools

Tools are the agent’s interface to the world. Let’s look at how to define custom tools and register them.

Defining Custom Tools

Each tool needs two things:

1. A JSON schema for the LLM to understand the interface
2. A Python function that implements the logic

# More tool examples
import httpx
import json
from datetime import datetime

def get_current_time(timezone: str = "UTC") -> str:
    """Get the current time for a given timezone."""
    now = datetime.utcnow()
    return f"Current time ({timezone}): {now.isoformat()}"

def read_file_content(filepath: str) -> str:
    """Read the contents of a file."""
    try:
        with open(filepath, 'r') as f:
            return f.read()
    except FileNotFoundError:
        return f"Error: File '{filepath}' not found."
    except Exception as e:
        return f"Error reading file: {str(e)}"

def fetch_url(url: str) -> str:
    """Fetch the content of a URL."""
    try:
        response = httpx.get(url, timeout=10)
        response.raise_for_status()
        return response.text[:2000]  # Truncate for safety
    except Exception as e:
        return f"Error fetching URL: {str(e)}"

Registering Tools with the Agent

Add the new tools to the registry:

# Add to tools.py
TOOLS.extend([
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "Get the current time",
            "parameters": {
                "type": "object",
                "properties": {
                    "timezone": {
                        "type": "string",
                        "description": "Timezone string (e.g., 'UTC', 'US/Eastern')"
                    }
                }
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_file_content",
            "description": "Read a file from the local filesystem",
            "parameters": {
                "type": "object",
                "properties": {
                    "filepath": {
                        "type": "string",
                        "description": "Path to the file"
                    }
                },
                "required": ["filepath"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "fetch_url",
            "description": "Fetch content from a URL",
            "parameters": {
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "The URL to fetch"
                    }
                },
                "required": ["url"]
            }
        }
    },
])

TOOL_MAP.update({
    "get_current_time": get_current_time,
    "read_file_content": read_file_content,
    "fetch_url": fetch_url,
})

Step 5: Adding Memory

Without memory, an agent treats every interaction as a fresh start. Let’s add memory management.

Conversation History Management

The simplest approach is maintaining a context window — keeping the most recent N messages:

# memory.py
from typing import List, Dict

class ConversationMemory:
    """Manages conversation history with a sliding window."""
    
    def __init__(self, max_history: int = 10):
        self.max_history = max_history
        self.history: List[Dict] = []
    
    def add_message(self, role: str, content: str):
        self.history.append({"role": role, "content": content})
        # Trim to max_history if needed
        if len(self.history) > self.max_history:
            self.history = self.history[-self.max_history:]
    
    def get_context(self) -> List[Dict]:
        return self.history.copy()
    
    def clear(self):
        self.history = []

Persistent Memory with File Storage

For long-term memory, we can persist conversations to disk:

# memory.py (continued)
import json
import os

class PersistentMemory:
    """Memory that persists conversations to a JSON file."""
    
    def __init__(self, storage_path: str = "memory_store.json"):
        self.storage_path = storage_path
        self.conversations: Dict[str, List[Dict]] = {}
        self._load()
    
    def _load(self):
        if os.path.exists(self.storage_path):
            with open(self.storage_path, 'r') as f:
                self.conversations = json.load(f)
    
    def _save(self):
        with open(self.storage_path, 'w') as f:
            json.dump(self.conversations, f, indent=2)
    
    def add_to_conversation(self, conversation_id: str, role: str, content: str):
        if conversation_id not in self.conversations:
            self.conversations[conversation_id] = []
        self.conversations[conversation_id].append({
            "role": role,
            "content": content,
            "timestamp": str(datetime.now())
        })
        self._save()
    
    def get_conversation(self, conversation_id: str) -> List[Dict]:
        return self.conversations.get(conversation_id, [])
    
    def list_conversations(self) -> List[str]:
        return list(self.conversations.keys())

Integrating memory into the agent loop:

# agent.py (with memory)
from memory import ConversationMemory, PersistentMemory

class AgentWithMemory:
    def __init__(self, memory_type="conversation"):
        self.memory = ConversationMemory() if memory_type == "conversation" else PersistentMemory()
    
    def run(self, user_message: str) -> str:
        # Add user message to memory
        self.memory.add_message("user", user_message)
        
        # Get context from memory
        context = self.memory.get_context()
        
        # Run the agent (same loop as before, using context)
        # ...
        
        # Add assistant response to memory
        self.memory.add_message("assistant", final_answer)
        return final_answer

Step 6: Building Workflows

Now we tie everything together into multi-step workflows with conditional branching and error handling.

Multi-Step Agent Workflows

A workflow orchestrates multiple agent calls, potentially with different tools and goals for each step:

# workflow.py
from typing import List, Dict, Callable, Any
import time

class WorkflowStep:
    """A single step in a workflow."""
    def __init__(self, name: str, agent_func: Callable, 
                 max_retries: int = 3, retry_delay: float = 1.0):
        self.name = name
        self.agent_func = agent_func
        self.max_retries = max_retries
        self.retry_delay = retry_delay

class Workflow:
    """Orchestrates a sequence of agent steps with error handling."""
    
    def __init__(self, steps: List[WorkflowStep]):
        self.steps = steps
        self.context: Dict[str, Any] = {}
    
    def run(self, initial_input: str) -> Dict[str, Any]:
        results = {}
        current_input = initial_input
        
        for step in self.steps:
            print(f"\n=== Executing step: {step.name} ===")
            
            for attempt in range(step.max_retries):
                try:
                    result = step.agent_func(current_input, self.context)
                    results[step.name] = result
                    current_input = result  # Pass result to next step
                    break
                except Exception as e:
                    print(f"Step '{step.name}' failed (attempt {attempt + 1}): {e}")
                    if attempt < step.max_retries - 1:
                        time.sleep(step.retry_delay)
                    else:
                        results[step.name] = f"ERROR: {str(e)}"
        
        return results

Conditional Branching

Not all workflows are linear. Sometimes the next step depends on the previous result:

# Conditional workflow example
class ConditionalWorkflow:
    def __init__(self):
        self.steps = {}
        self.conditions = []
    
    def add_step(self, name: str, func: Callable, 
                 next_step_if_ok: str = None, 
                 next_step_if_error: str = None):
        self.steps[name] = {
            "func": func,
            "next_ok": next_step_if_ok,
            "next_error": next_step_if_error,
        }
    
    def run(self, start_step: str, initial_input: str) -> Dict:
        current_step = start_step
        data = initial_input
        results = {}
        
        while current_step:
            step = self.steps.get(current_step)
            if not step:
                break
            
            try:
                print(f"Running step: {current_step}")
                data = step["func"](data)
                results[current_step] = {"status": "ok", "result": data}
                current_step = step.get("next_ok")
            except Exception as e:
                results[current_step] = {"status": "error", "error": str(e)}
                current_step = step.get("next_error")
        
        return results

Error Handling and Retry Logic

Robust agents need proper error handling. Here’s a complete example:

# Robust agent with retry and error handling
import time
from openai import OpenAI
from openai import APIError, RateLimitError, APITimeoutError

class RobustAgent:
    def __init__(self, api_key: str, model: str = "gpt-4"):
        self.client = OpenAI(api_key=api_key)
        self.model = model
        self.max_retries = 3
    
    def call_llm(self, messages: list, tools: list = None) -> str:
        """Call the LLM with retry logic."""
        last_error = None
        
        for attempt in range(self.max_retries):
            try:
                kwargs = {"model": self.model, "messages": messages}
                if tools:
                    kwargs["tools"] = tools
                    kwargs["tool_choice"] = "auto"
                
                response = self.client.chat.completions.create(**kwargs)
                return response.choices[0].message
                
            except RateLimitError as e:
                wait = min(2 ** attempt * 5, 60)  # Exponential backoff
                print(f"Rate limited. Waiting {wait}s...")
                time.sleep(wait)
                last_error = e
                
            except APITimeoutError as e:
                wait = 2 ** attempt
                print(f"Timeout. Retrying in {wait}s...")
                time.sleep(wait)
                last_error = e
                
            except APIError as e:
                print(f"API error: {e}")
                if attempt == self.max_retries - 1:
                    raise
                time.sleep(2 ** attempt)
                last_error = e
        
        raise last_error  # Should not reach here

Putting It All Together

Here’s a complete workflow that demonstrates everything:

# Complete workflow: research assistant
def research_assistant_workflow():
    """A research assistant that searches, analyzes, and summarizes."""
    
    # Step 1: Search for information
    def search_step(query: str, context: dict) -> str:
        agent = AgentWithMemory()
        return agent.run(f"Search the web for: {query}")
    
    # Step 2: Analyze the results
    def analyze_step(raw_data: str, context: dict) -> str:
        agent = AgentWithMemory()
        prompt = f"Analyze this information and extract key insights:\n\n{raw_data}"
        return agent.run(prompt)
    
    # Step 3: Generate a summary
    def summarize_step(analysis: str, context: dict) -> str:
        agent = AgentWithMemory()
        prompt = f"Create a concise summary of these findings:\n\n{analysis}"
        return agent.run(prompt)
    
    # Define the workflow
    workflow = Workflow([
        WorkflowStep("search", search_step, max_retries=2),
        WorkflowStep("analyze", analyze_step, max_retries=2),
        WorkflowStep("summarize", summarize_step, max_retries=3),
    ])
    
    # Run it
    results = workflow.run("Latest developments in AI agent technology")
    
    print("\n=== Final Results ===")
    for step_name, result in results.items():
        print(f"\n[{step_name}]:")
        print(result)
    
    return results

if __name__ == "__main__":
    research_assistant_workflow()

Conclusion

Recap

We’ve covered the complete journey of building AI agent workflows:

1. Core concepts — LLM + Tools + Memory + Planning
2. Environment setup — Python, OpenAI SDK, project structure
3. First agent — The perceive → think → act → observe loop
4. Adding tools — Web search, calculator, file I/O, URL fetching
5. Adding memory — Conversation history, persistent storage
6. Building workflows — Multi-step orchestration, conditional branching, error handling

Key Takeaways

• Start simple: A single agent with 2-3 tools is enough for most tasks
• Test incrementally: Add one tool at a time and verify each works
• Handle errors gracefully: Networks fail, APIs rate-limit, models hallucinate
• Monitor your agents: Log every step, tool call, and decision
• Iterate on prompts: The quality of your agent depends heavily on how you describe tools and objectives

Resources

• OpenAI Function Calling Documentation
• LangChain Agent Framework
• Anthropic Tool Use Guide
• Hugging Face Agents

Ethical Considerations

As you build autonomous agents, keep these principles in mind:

• Transparency: Users should know when they are interacting with an AI agent
• Safety: Implement guardrails to prevent harmful actions
• Privacy: Never expose user data through tool calls
• Accountability: Log all agent decisions for audit
• Human oversight: Always include a human-in-the-loop for critical decisions

Building AI agent workflows is one of the most empowering skills in modern software development. Start small, experiment often, and build responsibly. The future of automation is agentic — and you now have the tools to build it.