Agentic AI represents a fundamental shift from AI as a tool that responds to prompts to AI as an autonomous system that plans, executes, and adapts. Instead of asking an AI to answer a question, you give it a goal and it figures out the steps needed to achieve it.
This guide covers the architecture patterns behind production agentic AI systems, based on our experience building these systems at Cloudpush.
What Makes AI "Agentic"?
An AI agent has four key capabilities that distinguish it from a standard chatbot:
- Planning: Decomposing complex goals into actionable steps
- Tool Use: Interacting with external systems (APIs, databases, code execution)
- Memory: Maintaining context across multiple steps and sessions
- Reflection: Evaluating its own outputs and adjusting its approach
A simple chatbot responds to one message at a time. An agent takes a goal like "analyze our Q3 sales data and prepare a board presentation" and autonomously plans the steps, queries the database, analyzes the results, creates visualizations, and drafts the presentation.
Core Architecture Patterns
Pattern 1: ReAct (Reasoning + Acting)
The simplest and most widely used agent pattern. The agent alternates between reasoning about what to do and taking action:
Thought: I need to find the Q3 revenue data. Let me query the database.
Action: sql_query("SELECT * FROM revenue WHERE quarter = 'Q3' AND year = 2025")
Observation: [Results returned: 1,247 rows with revenue data by product line]
Thought: I have the raw data. Now I need to aggregate by product line and calculate growth.
Action: python_execute("df.groupby('product_line').agg({'revenue': 'sum'}).pct_change()")
Observation: [Growth rates: Enterprise +23%, SMB +12%, Consumer -5%]
Thought: Enterprise is growing strongly, Consumer is declining. Let me create a visualization.
Action: create_chart(data, type="bar", title="Q3 Revenue Growth by Product Line")Implementation:
class ReActAgent:
def __init__(self, llm, tools, max_steps=10):
self.llm = llm
self.tools = tools
self.max_steps = max_steps
async def run(self, goal: str) -> AgentResult:
history = []
for step in range(self.max_steps):
# Reasoning step
action = await self.llm.generate(
system_prompt=self.system_prompt,
messages=history,
tools=self.tools.schemas()
)
if action.is_final_answer:
return AgentResult(answer=action.content, steps=history)
# Action step
tool_result = await self.tools.execute(action.tool_call)
history.append({"action": action, "observation": tool_result})
return AgentResult(answer="Max steps reached", steps=history)When to use ReAct: Simple, linear tasks where each step informs the next. Good for: data analysis, research queries, customer support.
Pattern 2: Plan-and-Execute
The agent creates a complete plan before executing, then follows the plan step by step:
class PlanAndExecuteAgent:
def __init__(self, planner_llm, executor_llm, tools):
self.planner = planner_llm
self.executor = executor_llm
self.tools = tools
async def run(self, goal: str) -> AgentResult:
# Planning phase
plan = await self.planner.generate_plan(goal)
results = []
for step in plan.steps:
# Execution phase
result = await self.executor.execute_step(
step=step,
tools=self.tools,
previous_results=results
)
results.append(result)
# Optional: re-plan if results are unexpected
if result.needs_replanning:
remaining_steps = await self.planner.replan(
goal, plan, results
)
plan.steps = plan.steps[:step.index+1] + remaining_steps
return self.synthesize(results)When to use Plan-and-Execute: Complex, multi-step tasks where upfront planning improves efficiency. Good for: report generation, workflow automation, project management.
Pattern 3: Hierarchical Agents
A supervisor agent manages specialized worker agents:
class SupervisorAgent:
def __init__(self):
self.workers = {
"researcher": ResearchAgent(tools=[web_search, doc_retrieval]),
"analyst": AnalysisAgent(tools=[sql_query, python_exec]),
"writer": WritingAgent(tools=[template_engine, chart_creator]),
"reviewer": ReviewAgent(tools=[fact_checker, grammar_checker])
}
async def run(self, goal: str):
# Supervisor decides which worker to invoke
while not self.is_complete():
next_action = await self.decide_next_worker(goal, self.state)
worker = self.workers[next_action.worker_name]
result = await worker.execute(next_action.instruction)
self.state.update(result)
return self.state.final_outputWhen to use Hierarchical: Complex tasks requiring diverse skills. Good for: content creation pipelines, customer onboarding, incident response.
Tool Design for Agents
Tools are the bridge between the agent's reasoning and the real world. Well-designed tools make agents reliable; poorly designed tools make them unpredictable.
Tool Design Principles
class Tool:
name: str # Clear, descriptive name
description: str # When and why to use this tool
parameters: Schema # Strongly typed parameters
returns: Schema # Strongly typed return value
async def execute(self, params) -> ToolResult:
# Validate inputs
validated = self.parameters.validate(params)
# Execute with timeout
result = await asyncio.wait_for(
self._execute(validated),
timeout=30.0
)
# Validate outputs
return self.returns.validate(result)Critical Tool Design Rules
- Descriptive names:
search_customer_databasenotsearch_db - Clear descriptions: Tell the agent WHEN to use the tool, not just what it does
- Typed parameters: Use JSON Schema or Pydantic models
- Error messages: Return actionable error messages the agent can reason about
- Idempotent where possible: Retrying a tool should be safe
- Rate limited: Prevent the agent from overwhelming external services
Memory Architecture
Working Memory
The agent's current task context. Managed through the conversation history with strategic summarization:
class WorkingMemory:
def __init__(self, max_tokens=8000):
self.messages = []
self.max_tokens = max_tokens
def add(self, message):
self.messages.append(message)
self._prune_if_needed()
def _prune_if_needed(self):
while self.token_count() > self.max_tokens:
# Summarize oldest messages
oldest = self.messages[:3]
summary = self.summarize(oldest)
self.messages = [summary] + self.messages[3:]Episodic Memory
Past task executions that the agent can learn from:
class EpisodicMemory:
async def store_episode(self, task, steps, outcome):
await self.vector_store.upsert(
text=f"Task: {task}\nOutcome: {outcome}",
metadata={
"steps": steps,
"outcome": outcome.status,
"timestamp": datetime.utcnow()
}
)
async def recall_similar(self, current_task, top_k=3):
return await self.vector_store.search(current_task, top_k=top_k)Semantic Memory
Domain knowledge and facts the agent can reference:
class SemanticMemory:
def __init__(self, knowledge_base):
self.kb = knowledge_base # Vector database with domain knowledge
async def query(self, question):
# RAG-style retrieval from the knowledge base
relevant_docs = await self.kb.search(question, top_k=5)
return relevant_docsProduction Safety for Agentic Systems
Agentic systems require additional safety measures beyond standard AI guardrails:
Execution Sandboxing
Never let agents run code or access systems without sandboxing:
class SandboxedCodeExecution:
async def execute(self, code: str) -> ExecutionResult:
container = await self.docker.create(
image="python-sandbox:latest",
mem_limit="512m",
cpu_period=100000,
cpu_quota=50000, # 50% CPU
network_disabled=True, # No network access
read_only=True,
timeout=30
)
result = await container.exec(code)
await container.remove()
return resultAction Approval Gates
Critical actions require human approval:
class ApprovalGate:
REQUIRES_APPROVAL = {
"send_email": True,
"modify_database": True,
"create_ticket": True,
"web_search": False,
"read_document": False,
}
async def check(self, tool_name, params, context):
if self.REQUIRES_APPROVAL.get(tool_name, True):
approval = await self.request_approval(tool_name, params, context)
if not approval.granted:
return ToolResult(error="Action not approved by human reviewer")
return None # ProceedBudget and Rate Limits
class AgentBudget:
def __init__(self, max_cost=5.0, max_steps=50, max_time=300):
self.max_cost = max_cost
self.max_steps = max_steps
self.max_time = max_time # seconds
self.current_cost = 0
self.current_steps = 0
self.start_time = time.time()
def check(self, step_cost):
self.current_cost += step_cost
self.current_steps += 1
elapsed = time.time() - self.start_time
if self.current_cost > self.max_cost:
raise BudgetExceeded(f"Cost limit: ${self.max_cost}")
if self.current_steps > self.max_steps:
raise StepLimitExceeded(f"Step limit: {self.max_steps}")
if elapsed > self.max_time:
raise TimeLimitExceeded(f"Time limit: {self.max_time}s")Evaluation and Testing
Agent Evaluation Framework
class AgentEvaluator:
async def evaluate(self, agent, test_cases):
results = []
for test in test_cases:
result = await agent.run(test.goal)
score = self.score(result, test.expected)
results.append({
"goal": test.goal,
"success": score.passed,
"steps_taken": len(result.steps),
"cost": result.total_cost,
"latency": result.total_time,
"correctness": score.correctness,
"efficiency": score.efficiency
})
return AgentEvalReport(results)What to Test
- Task completion: Does the agent achieve the goal?
- Efficiency: How many steps/tokens/dollars does it take?
- Safety: Does the agent stay within its approved boundaries?
- Error recovery: How does the agent handle tool failures?
- Edge cases: What happens with ambiguous, impossible, or adversarial goals?
Getting Started
Agentic AI systems are the most exciting and challenging area of applied AI right now. The key is starting with well-defined, bounded tasks and gradually expanding the agent's autonomy as you build confidence in its reliability and safety.
Contact Cloudpush to discuss how agentic AI can automate complex workflows in your business.