From concepts to code. Building agent components, wiring them into workflows, and understanding why the class-based pattern beats a pile of scripts.
When you set out to build an agentic system, there is a fork in the road. Neither path is wrong. The right one depends on what you are optimising for.
Frameworks are valuable for production systems and rapid iteration. But building the core logic yourself reveals exactly how agents work under the hood — how prompts shape behaviour, how data flows between components, how orchestration happens. That understanding transfers to any framework you use later.
Emerging protocols like MCP (Model Context Protocol) and A2A (Agent-to-Agent) are standardising how agents discover tools and communicate with each other. The class-based patterns here underlie those standards.
Whether you use a framework or build from scratch, every agentic system has the same three layers. The difference is who writes them: you or the framework.
The intelligence behind every agent. A well-designed system makes this swappable -- you should be able to switch from GPT-4 to Claude without rewriting agent logic. In practice: a thin wrapper that accepts a prompt and returns text.
Python code that defines how each agent behaves. This is where prompt engineering lives -- system prompts give agents personas (the role or character each one plays), and the code implements their specific capabilities.
The conductor. It decides which agents run, in what order, and how data flows between them. For deterministic workflows -- ones that run the same fixed steps every time -- this is a fixed sequence. For agentic workflows it is dynamic, decided at runtime based on agent outputs.
A clean separation of concerns makes agent systems maintainable. Three layers of code, each with a single responsibility: define, configure, wire.
Everything rests on one small wrapper -- chat() -- that sends a prompt to the model and returns the text reply. Agents never touch the API client directly. That indirection is what makes the LLM engine swappable.
from openai import OpenAI
client = OpenAI()
def chat(user_prompt, system_prompt="You are a helpful assistant.", max_tokens=200):
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
max_tokens=max_tokens,
)
return response.choices[0].message.contentA base Agent class enforces the contract: every agent has a name and a run() method. Specialist agents inherit from it and override run() with their own logic. The system prompt in each subclass is where the agent's persona lives.
class Agent:
"""Base agent -- every agent needs a name and a run() method."""
def __init__(self, name):
self.name = name
def run(self, input_data):
raise NotImplementedError
class ResearchAgent(Agent):
def run(self, query):
return chat(query, system_prompt="You are a research analyst. Gather and summarise relevant information.")
class FactCheckerAgent(Agent):
def run(self, text):
return chat(text, system_prompt="You are a fact checker. Verify claims and flag anything uncertain.")
class SummarizerAgent(Agent):
def run(self, text):
return chat(text, system_prompt="You are a summariser. Condense the verified content into a short paragraph.")Instantiation configures each agent for a specific role in this workflow. The workflow script then wires them together: each agent's output becomes the next agent's input.
# 1. Configure agents for this workflow
researcher = ResearchAgent("Research Assistant")
fact_checker = FactCheckerAgent("Fact Checker")
summarizer = SummarizerAgent("Summarizer")
# 2. Wire them together -- output of one feeds the next
query = "Agentic workflows in AI systems"
research_results = researcher.run(query)
verified_results = fact_checker.run(research_results)
final_summary = summarizer.run(verified_results)Agent definitions are reusable across workflows. The same ResearchAgent class can appear in a report-writing workflow today and an email-drafting workflow tomorrow. Instantiation configures agents for a specific job. Orchestration logic is the only part that changes between workflows.
The simplest way to understand what an LLM adds to an agent: compare a hardcoded function with an LLM-powered one handling the same job.
This is the fundamental tradeoff: deterministic functions are predictable and fast but brittle. LLM-powered agents are flexible and reasoning-capable but slower and less predictable. Real agentic systems use both -- deterministic logic for well-known paths, LLM calls for the parts that require judgment.
Here is how a three-agent pipeline looks end to end. Each agent transforms the data and passes it forward. The workflow itself is still deterministic -- the agents always run in the same order. Making it agentic means adding a decision step that can loop back or branch based on what the agents produce.
This worked example shows the pattern in a runnable form. Two agents -- a fetcher and a processor -- wired into a pipeline. No LLM yet: the point is the class structure and the early-exit on error, which applies identically when real API calls replace the simulated data.
class Agent:
def __init__(self, name):
self.name = name
def execute(self, data=None):
raise NotImplementedError
class DataFetchingAgent(Agent):
def __init__(self, name, data_source):
super().__init__(name)
self.data_source = data_source
def execute(self, user_id):
# Simulated -- a real agent would call an API or database here
users = {
"123": {"name": "Alice Wonderland", "occupation": "Dreamer", "user_id": "123"},
}
return users.get(user_id, {"error": "User not found"})
class DataProcessingAgent(Agent):
def __init__(self, name, fields_to_extract=None):
super().__init__(name)
self.fields_to_extract = fields_to_extract or ["name"]
def execute(self, fetched_data):
if "error" in fetched_data:
return {"error": fetched_data["error"]}
return {f: fetched_data.get(f, "N/A") for f in self.fields_to_extract}Notice the three things worth carrying forward: inheritance guarantees every agent exposes the same interface; configuration at init (data_source, fields_to_extract) makes agents reusable with different settings; and the error check inside DataProcessingAgent.execute() means a failure in the fetcher surfaces cleanly instead of crashing half-way through.
from workflow_agents.agent_definitions import DataFetchingAgent, DataProcessingAgent
def run_user_data_workflow(user_id):
# Instantiation -- configure each agent
fetcher = DataFetchingAgent(name="ProfileFetcher", data_source="MainUserDB")
processor = DataProcessingAgent(
name="InfoExtractor",
fields_to_extract=["name", "occupation", "user_id"],
)
# Workflow -- wire agents together
user_data = fetcher.execute(user_id=user_id)
if "error" in user_data: # early exit on failure
return {"status": "error", "detail": user_data["error"]}
report = processor.execute(fetched_data=user_data)
return {"status": "success", "data": report}
if __name__ == "__main__":
print(run_user_data_workflow(user_id="123")) # happy path
print(run_user_data_workflow(user_id="999")) # error path{'status': 'success', 'data': {'name': 'Alice Wonderland', 'occupation': 'Dreamer', 'user_id': '123'}}
{'status': 'error', 'detail': 'User not found'}This exercise is deliberately simple -- no LLM, no API calls. The point is the pattern: agent classes define capabilities, instantiation configures them, and the workflow script wires them together. Once you are comfortable with this skeleton, replacing the simulated data with a real chat() call is a one-line change per agent.