Agent-based architectures are becoming increasingly popular in modern software design, especially in AI-driven systems. In this article, we’ll walk through a simple yet powerful example of how to design and implement an agentic workflow in Python, using modular components and clear separation of responsibilities.
What Is an Agentic Workflow?
An agentic workflow is a system where multiple specialized components (agents) collaborate to accomplish a task. Each agent has a defined role, and data flows between them in a structured way.
Think of it as a team:
One agent gathers information
Another processes it
Others could analyze, validate, or store it
Step 1: Defining Agent Blueprints
We begin by creating a module (agent_definitions.py) that contains our agent classes. These classes act as reusable blueprints.
Base Agent Class
At the core, we define a generic Agent class:
It assigns a name to each agent
It enforces the implementation of an execute() method
This ensures consistency across all agents.
Specialized Agents
We then create two concrete agents:
DataFetchingAgent
Responsible for retrieving data
Simulates fetching user information from a data source
Returns either valid user data or an error
DataProcessingAgent
Takes fetched data as input
Extracts specific fields such as name and occupation
Handles errors gracefully
This separation of concerns makes the system flexible and easy to extend.
Step 2: Instantiating Agents
In main_workflow.py, we import our agent classes and create specific instances:
A fetcher configured with a data source
A processor configured with fields to extract
Each instance is tailored for its role, demonstrating how agents can be customized dynamically.
Step 3: Implementing the Workflow
The workflow is defined as a sequence of steps inside a function.
First, fetch data. The fetching agent is executed with a user_id. If successful, the data moves to the next step. If it fails, the workflow stops early.
Second, process data. The processing agent extracts meaningful information from the fetched data.
Key Concept: Data Flow Between Agents
The most important idea here is data passing.
Fetcher → Processor
The output of one agent becomes the input of the next. This chaining is what makes agentic workflows powerful and scalable.
Error Handling
A simple but effective error-handling mechanism is included:
If fetching fails, stop the workflow
Prevent unnecessary processing
Return a clear status message
This pattern is essential in real-world systems.
Project Structure
To run this example, organize your project like this:
your_project_directory/
├── workflow_agents/
│ ├── __init__.py
│ └── agent_definitions.py
└── main_workflow.py
Then execute:
python main_workflow.py
Example Runs
Successful case
Input: user_id = "123"
Output: Processed user information (name, occupation, user_id)
Failure case
Input: user_id = "456"
Output: Error message (“User not found”)
Why This Pattern Matters
This simple example demonstrates powerful design principles:
Modularity: each agent does one thing well
Reusability: agents can be reused in different workflows
Scalability: you can easily add more agents
Maintainability: clear separation of logic
Agent-based workflows are a foundational concept in modern AI systems and distributed architectures. Even this minimal example shows how you can break down complex tasks, build flexible pipelines, and create systems that are easy to extend.
Agent Library/Module (e.g., agent_definitions.py****)
This file will house the class definitions for our various types of
agents. These classes act as blueprints.
# workflow_agents/agent_definitions.py
class Agent:
"""
A base class for our agents.
While not strictly necessary for simple cases,
it can be useful for defining common interfaces or utilities.
"""
def __init__(self, name):
self.name = name
print(f"Agent '{self.name}' initialized.")
def execute(self, data=None):
"""
A generic method to execute the agent's task.
Specific agents will override this.
"""
raise NotImplementedError("Each agent must implement the
'execute' method.")
class DataFetchingAgent(Agent):
"""
An agent specialized in fetching data.
For this example, it will simulate fetching user data.
"""
def __init__(self, name, data_source):
super().__init__(name)
self.data_source = data_source
print(f"DataFetchingAgent will fetch from: {self.data_source}")
def execute(self, user_id):
"""
Simulates fetching data for a given user_id.
In a real scenario, this might involve an API call or database query.
"""
print(f"'{self.name}' is fetching data for user_id: {user_id}
from {self.data_source}...")
# Simulate data fetching
if user_id == "123":
return {"user_id": "123", "name": "Alice Wonderland",
"occupation": "Dreamer"}
else:
return {"user_id": user_id, "error": "User not found"}
class DataProcessingAgent(Agent):
"""
An agent specialized in processing data.
For this example, it will extract key information from the fetched data.
"""
def __init__(self, name, fields_to_extract=None):
super().__init__(name)
self.fields_to_extract = fields_to_extract if
fields_to_extract else ["name", "occupation"]
print(f"DataProcessingAgent will extract fields:
{self.fields_to_extract}")
def execute(self, fetched_data):
"""
Processes the fetched data to extract specified fields.
"""
print(f"'{self.name}' is processing data: {fetched_data}")
if "error" in fetched_data:
return {"processed_info": None, "error": fetched_data["error"]}
processed_info = {}
for field in self.fields_to_extract:
processed_info[field] = fetched_data.get(field, "N/A")
return {"processed_info": processed_info,
"original_data_keys": list(fetched_data.keys())}
agent_definitions.py:
We have a base Agent class with an __init__ method to give each agent
a name and an execute method that specific agents must implement.
DataFetchingAgent inherits from Agent. Its execute method simulates
fetching data for a user.
DataProcessingAgent also inherits from Agent. Its execute method takes
the data fetched by the previous agent and extracts specific fields.
These classes serve as the blueprints defining the agents' attributes
and methods.
Agent Instantiation and Workflow Implementation Logic (e.g., main_workflow.py)
This script will import the agent classes, create specific instances
(objects) of them, and then define the actual workflow by coordinating
these instances.
# main_workflow.py
# Import agent classes from our library
from workflow_agents.agent_definitions import DataFetchingAgent,
DataProcessingAgent
def run_user_data_workflow(user_id_to_process):
"""
This function implements the agentic workflow.
"""
print(f"\n--- Starting User Data Workflow for User ID:
{user_id_to_process} ---")
# === 2. Agent Instantiation Logic ===
# Create specific instances (objects) of agent classes and configure them.
# Each agent can be customized for its specific role or task.
fetcher = DataFetchingAgent(name="UserProfileFetcher",
data_source="MainUserDatabase")
processor = DataProcessingAgent(name="UserInfoExtractor",
fields_to_extract=["name", "occupation", "user_id"])
print("--- Agents Instantiated ---")
# === 3. Workflow Implementation Logic ===
# Define the sequence of tasks and how data is passed between agents.
# Step 1: Fetch user data
print("\nStep 1: Fetching Data...")
fetched_user_data = fetcher.execute(user_id=user_id_to_process)
print(f"Fetcher Output: {fetched_user_data}")
if "error" in fetched_user_data and fetched_user_data["error"]:
print(f"Workflow stopped due to error in fetching:
{fetched_user_data['error']}")
return {"status": "Error", "details": fetched_user_data['error']}
# Step 2: Process fetched data
print("\nStep 2: Processing Data...")
processed_data_report = processor.execute(fetched_data=fetched_user_data)
print(f"Processor Output: {processed_data_report}")
print("--- Workflow Completed ---")
return {"status": "Success", "data": processed_data_report}
if __name__ == "__main__":
# Example 1: Successful run
result1 = run_user_data_workflow("123")
print(f"\nFinal Result for User 123: {result1}\n")
# Example 2: User not found
result2 = run_user_data_workflow("456")
print(f"Final Result for User 456: {result2}")
main_workflow.py:
Import: It first imports the agent classes defined in
workflow_agents.agent_definitions.
Agent Instantiation Logic:
Inside run_user_data_workflow, we create instances: Workspaceer =
DataFetchingAgent(...) and processor = DataProcessingAgent(...).
Each instance is configured upon creation (e.g., data_source for the
fetcher, fields_to_extract for the processor). This is where each
agent could be customized to its specific role or task.
Workflow Implementation Logic:
The workflow is defined by the sequence of calls:
Workspaceer.execute() is called first.
Its output, Workspaceed_user_data, is then passed as input to
processor.execute().
This demonstrates defining the sequence of tasks and how data is
passed between the instantiated agents.
Basic error handling is included to show how a workflow might react if
a step fails.
The if __name__ == "__main__": block shows how to trigger the workflow
with different inputs.
To make this run, you would structure your files like this:
your_project_directory/
├── workflow_agents/
│ ├── __init__.py # This can be an empty file
│ └── agent_definitions.py
└── main_workflow.py