Agentic AI: Paradigms and Design Patterns for Intelligent Autonomous Systems
Introduction
Artificial intelligence is undergoing a fundamental transformation: we are transitioning from passive algorithms requiring constant supervision to autonomous agents capable of perceiving, reasoning, and acting independently in complex environments. This revolution has a name: Agentic AI.
In this article, we will explore the conceptual foundations of Agentic AI, its underlying architectures, and especially the design patterns that enable the construction of robust, scalable, and adaptive agentic systems.
Agentic AI Overview
Agentic AI refers to artificial intelligence systems designed as autonomous agents endowed with three fundamental capabilities:
- Perception: Ability to observe and understand their environment through sensors or data streams
- Cognition: Aptitude to reason, plan, and make decisions based on objectives
- Action: Power to execute actions to modify their environment or accomplish tasks
The Pillars of Agentic AI
1. Operational Autonomy
Agentic agents operate largely independently, minimizing the need for continuous human intervention. They can handle unexpected situations and make real-time decisions.
2. Goal Orientation
Unlike simple reactive systems, agentic agents pursue explicit goals. They plan their actions based on short-term and long-term objectives.
3. Contextual Adaptability
These systems dynamically adjust their behavior based on environmental changes, evolving constraints, and new available information.
4. Learning Capability
Through machine learning techniques (particularly reinforcement learning), agentic agents improve their performance over time.
The Importance of Agentic AI
Transformative Use Cases
- Autonomous Vehicles: Real-time navigation in complex urban environments
- Industrial Automation: Collaborative robots adapting to production lines
- Intelligent Assistants: Virtual agents managing calendars, communications, and workflows
- Algorithmic Finance: Adaptive trading systems optimizing portfolios
- Connected Healthcare: Diagnostic agents assisting physicians in medical data analysis
Key Advantages
- Increased Efficiency: Drastic reduction of repetitive manual interventions
- Scalability: Simultaneous management of multiple parallel tasks
- Resilience: Adaptation to failures and degraded conditions
- Human Augmentation: Human-machine collaboration for more informed decisions
Foundational Design Patterns
Agentic patterns constitute proven architectural solutions for solving recurring problems in autonomous agent design. The five patterns below cover core agent behaviors — from reflexive responses to collaborative multi-agent coordination:
1. Reactive Pattern
Principle: Immediate responses to environmental stimuli without complex planning.
Characteristics:
- Minimal latency between perception and action
- Stimulus-response architecture
- Absence of complete world model
Example: Anti-collision system of a drone detecting an obstacle and performing an instantaneous evasive maneuver.
Applications: Real-time robotics, video games, critical systems requiring ultra-fast reactions.
1
2
3
4
5
6
7
8
9
10
11
12
# Pseudo-code for a reflex agent
class ReflexAgent:
def perceive(self, environment):
return environment.get_current_state()
def decide(self, state):
if state.obstacle_detected:
return "EVADE"
return "CONTINUE"
def act(self, action):
self.execute(action)
2. Goal-Oriented Pattern
Principle: Deliberative planning oriented towards achieving specific objectives.
Characteristics:
- Explicit representation of goals
- Planning algorithms (A*, MCTS, etc.)
- Action evaluation according to their contribution to objectives
Example: Delivery robot optimizing its route to minimize energy consumption while meeting deadlines.
Applications: Logistics, trajectory planning, strategic recommendation systems.
1
2
3
4
5
6
7
8
9
10
11
12
# Pseudo-code for a goal-oriented agent
class GoalOrientedAgent:
def __init__(self, goal):
self.goal = goal
self.planner = PathPlanner()
def plan(self, current_state):
return self.planner.find_path(current_state, self.goal)
def execute_plan(self, plan):
for action in plan:
self.perform(action)
3. Hierarchical Pattern
Principle: Decomposition of complex tasks into hierarchically organized subtasks.
Characteristics:
- Multi-level architecture (strategic, tactical, operational)
- Specialized agents by abstraction layer
- Inter-level communication
Example: Personal assistant simultaneously managing agenda planning, travel reservations, and email prioritization.
Applications: Enterprise management systems, workflow orchestration, multi-function virtual assistants.
4. Learning-Based Pattern (Adaptive Pattern)
Principle: Continuous behavior improvement through learning from experience.
Characteristics:
- Use of ML techniques (reinforcement learning, supervised learning)
- Exploration vs exploitation mechanisms
- Dynamic update of decision policies
Example: Stock trading agent refining its strategies via deep reinforcement learning (DRL).
Applications: Algorithmic finance, content personalization, adaptive video game AI.
1
2
3
4
5
6
7
8
9
10
11
12
# Pseudo-code for an adaptive agent
class AdaptiveAgent:
def __init__(self):
self.policy_network = NeuralNetwork()
self.experience_replay = []
def act(self, state):
return self.policy_network.predict(state)
def learn(self, state, action, reward, next_state):
self.experience_replay.append((state, action, reward, next_state))
self.policy_network.train(self.experience_replay)
5. Collaborative Pattern
Principle: Cooperation between multiple agents or between agents and humans to solve complex problems.
Characteristics:
- Inter-agent communication protocols
- Coordination and negotiation mechanisms
- Knowledge and objective sharing
Example: Swarm of drones collaborating to map a disaster area after a natural catastrophe.
Applications: Multi-agent systems, swarm robotics, collaborative medical diagnosis.
Reference Architecture for Agentic AI
A typical agentic architecture comprises the following components:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
┌─────────────────────────────────────────────┐
│ PERCEPTION LAYER │
│ (Sensors, APIs, Data Streams) │
└──────────────┬──────────────────────────────┘
│
┌──────────────▼──────────────────────────────┐
│ COGNITION LAYER │
│ ┌─────────────────────────────────────┐ │
│ │ World Model │ │
│ │ (State Representation) │ │
│ └─────────────────────────────────────┘ │
│ ┌─────────────────────────────────────┐ │
│ │ Reasoning Engine │ │
│ │ (Planning, Decision Making) │ │
│ └─────────────────────────────────────┘ │
│ ┌─────────────────────────────────────┐ │
│ │ Memory & Learning │ │
│ │ (Experience Replay, Knowledge Base)│ │
│ └─────────────────────────────────────┘ │
└──────────────┬──────────────────────────────┘
│
┌──────────────▼──────────────────────────────┐
│ ACTION LAYER │
│ (Actuators, API Calls, Outputs) │
└─────────────────────────────────────────────┘
Methodology for Developing Agentic Agents
Step 1: Domain Analysis
- Precisely define the agent’s objectives
- Identify environmental constraints
- Map the space of possible actions
Step 2: Pattern Selection
- Choose pattern(s) adapted to task complexity
- Consider latency and precision requirements
- Evaluate learning and adaptability needs
Step 3: Hybrid Architecture
- Combine multiple patterns if necessary (e.g., reactive + goal-oriented)
- Define interfaces between components
- Design fallback and robustness mechanisms
Step 4: Iterative Implementation
- Prototype with a subset of functionalities
- Test in simulated environments
- Deploy progressively with continuous monitoring
Step 5: Continuous Optimization
- Analyze performance metrics
- Refine models and strategies
- Adapt to feedback and experience
Production Implementation Patterns
The following patterns provide production-grade implementations using LangChain, LangGraph, and OpenTelemetry. Each addresses real engineering concerns: retries, cost budgets, human oversight gates, and distributed tracing.
6. ReAct Pattern — Reasoning + Acting
The ReAct (Reasoning + Acting) pattern interleaves chain-of-thought reasoning with tool invocations, enabling agents to reason about their next step before acting:
1
2
3
4
5
6
7
8
Thought: I need to find the current stock price of AAPL.
Action: search(query="AAPL stock price today")
Observation: AAPL is trading at $192.34
Thought: Now I have the price. I should calculate the percentage change from $180.
Action: calculator(expression="(192.34 - 180) / 180 * 100")
Observation: 6.856
Thought: AAPL is up 6.86% from $180.
Answer: AAPL is currently trading at $192.34, up 6.86% from the $180 reference price.
ReAct Agent with LangChain
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain_core.prompts import PromptTemplate
@tool
def search_web(query: str) -> str:
"""Search the web for current information."""
# In production: integrate with Tavily, SerpAPI, or Bing Search
return f"Search results for: {query}"
@tool
def run_python(code: str) -> str:
"""Execute Python code and return the output. Only use for math/data operations."""
import io, contextlib
stdout = io.StringIO()
try:
with contextlib.redirect_stdout(stdout):
exec(code, {"__builtins__": {}}) # sandboxed
return stdout.getvalue() or "Executed successfully (no output)"
except Exception as e:
return f"Error: {e}"
@tool
def read_file(path: str) -> str:
"""Read a text file. Only safe paths allowed."""
import pathlib
safe_base = pathlib.Path("/data/sandbox")
target = (safe_base / path).resolve()
if not str(target).startswith(str(safe_base)):
return "Error: Access denied — path outside sandbox"
return target.read_text() if target.exists() else f"File not found: {path}"
REACT_TEMPLATE = """You are a helpful assistant with access to tools. Use them to answer accurately.
Available tools:
{tools}
Format:
Thought: [reasoning about what to do next]
Action: [tool_name]
Action Input: [input to the tool]
Observation: [tool result]
... (repeat as needed)
Thought: I now have enough information to answer.
Final Answer: [your complete answer]
Question: {input}
{agent_scratchpad}"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [search_web, run_python, read_file]
prompt = PromptTemplate.from_template(REACT_TEMPLATE)
agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=10,
handle_parsing_errors=True,
return_intermediate_steps=True,
)
result = executor.invoke({"input": "What is 15% of the GDP of France (2023 estimate)?"})
print(result["output"])
7. Plan-and-Execute Pattern
For complex multi-step tasks, agents first generate a full plan, then execute each step:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel
from typing import List
import asyncio
class ExecutionPlan(BaseModel):
steps: List[str]
reasoning: str
class PlanAndExecuteAgent:
def __init__(self, tools: list, model: str = "gpt-4o-mini"):
self.llm = ChatOpenAI(model=model, temperature=0)
self.tools = {t.name: t for t in tools}
def plan(self, objective: str) -> ExecutionPlan:
prompt = f"""You are a planning agent. Create a step-by-step plan to accomplish this objective.
Objective: {objective}
Available tools: {list(self.tools.keys())}
Create a numbered list of concrete, actionable steps. Each step should use one specific tool.
Return JSON: steps"""
response = self.llm.invoke(prompt)
import json
data = json.loads(response.content)
return ExecutionPlan(**data)
def execute_step(self, step: str, context: dict) -> str:
"""Execute a single plan step, injecting results from previous steps."""
exec_prompt = f"""Execute this step using the available tools.
Context from previous steps:
{json.dumps(context, indent=2)}
Step to execute: {step}
Use the appropriate tool and return the result."""
# Simplified: in production, this would call the actual tool
return f"Executed: {step} → Result: [simulated output]"
def run(self, objective: str) -> dict:
plan = self.plan(objective)
context = {}
results = []
for i, step in enumerate(plan.steps):
result = self.execute_step(step, context)
context[f"step_{i+1}_result"] = result
results.append({"step": step, "result": result})
print(f"✓ Step {i+1}: {step[:60]}...")
# Final synthesis
synthesis_prompt = f"""
Synthesize these execution results into a complete answer for: {objective}
Steps and results:
{json.dumps(results, indent=2)}
Provide a comprehensive, actionable answer."""
final = self.llm.invoke(synthesis_prompt)
return {"plan": plan.steps, "steps": results, "answer": final.content}
8. Multi-Agent Architecture
Multiple specialized agents collaborate, each owning a domain:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
from langchain_openai import ChatOpenAI
from dataclasses import dataclass
from typing import Callable
@dataclass
class Agent:
name: str
description: str
system_prompt: str
tools: list
llm: ChatOpenAI
class MultiAgentSystem:
"""Orchestrator that routes tasks to specialist agents."""
def __init__(self, agents: list[Agent], orchestrator_llm: ChatOpenAI):
self.agents = {a.name: a for a in agents}
self.orchestrator = orchestrator_llm
def route(self, task: str) -> str:
"""Decide which agent should handle this task."""
agent_descriptions = "\n".join(
f"- {name}: {agent.description}"
for name, agent in self.agents.items()
)
prompt = f"""Route this task to the most appropriate agent.
Available agents:
{agent_descriptions}
Task: {task}
Return only the agent name, nothing else."""
return self.orchestrator.invoke(prompt).content.strip()
def execute(self, task: str) -> dict:
agent_name = self.route(task)
agent = self.agents.get(agent_name)
if not agent:
return {"error": f"Unknown agent: {agent_name}"}
# Execute task with chosen agent
messages = [
{"role": "system", "content": agent.system_prompt},
{"role": "user", "content": task},
]
result = agent.llm.invoke(messages)
return {"agent": agent_name, "result": result.content}
# Example: Research + Writing + Review agent pipeline
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
researcher = Agent(
name="researcher",
description="Searches for factual information, data, and evidence",
system_prompt="You are a research specialist. Find accurate, up-to-date information. Always cite sources.",
tools=[search_web],
llm=llm,
)
writer = Agent(
name="writer",
description="Writes clear, engaging content based on provided information",
system_prompt="You are an expert writer. Produce professional, well-structured content.",
tools=[],
llm=ChatOpenAI(model="gpt-4o-mini", temperature=0.7),
)
reviewer = Agent(
name="reviewer",
description="Reviews content for accuracy, clarity, and quality",
system_prompt="You are a senior editor. Identify errors, unclear passages, and suggest improvements.",
tools=[],
llm=llm,
)
system = MultiAgentSystem(
agents=[researcher, writer, reviewer],
orchestrator_llm=llm,
)
result = system.execute("Research and write a 3-paragraph summary of quantum computing advances in 2024.")
print(f"Routed to: {result['agent']}")
print(result['result'])
9. LangGraph for Stateful Agentic Workflows
LangGraph is the production standard for building stateful, cyclic agent workflows with explicit control flow:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
retry_count: int
final_answer: str | None
# Tools
@tool
def search(query: str) -> str:
"""Search the web."""
return f"Results for: {query}"
@tool
def calculate(expr: str) -> str:
"""Evaluate a math expression."""
return str(eval(expr, {"__builtins__": {}}))
tools = [search, calculate]
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)
tool_node = ToolNode(tools)
# Nodes
def agent_node(state: AgentState) -> AgentState:
response = llm.invoke(state["messages"])
return {"messages": [response]}
def should_continue(state: AgentState) -> str:
messages = state["messages"]
last_message = messages[-1]
retry_count = state.get("retry_count", 0)
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
if retry_count >= 5:
return "error" # prevent infinite loops
return "tools"
return END
# Build graph
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {
"tools": "tools",
"error": END,
END: END,
})
graph.add_edge("tools", "agent")
app = graph.compile()
# Invoke
result = app.invoke({
"messages": [{"role": "user", "content": "What is 15% of 2340, and who invented Python?"}],
"retry_count": 0,
"final_answer": None,
})
print(result["messages"][-1].content)
10. Error Handling and Resilience
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import time
class ResilientAgentState(TypedDict):
task: str
current_step: str
results: list
error: str | None
retry_count: int
max_retries: int
completed: bool
def safe_tool_executor(state: ResilientAgentState) -> ResilientAgentState:
"""Tool execution with retry logic and graceful degradation."""
try:
# Simulate tool call
result = call_external_tool(state["task"])
return {
"results": state["results"] + [result],
"error": None,
"retry_count": 0, # reset on success
}
except TimeoutError:
retry = state["retry_count"] + 1
if retry <= state["max_retries"]:
time.sleep(2 ** retry) # exponential backoff
return {"retry_count": retry, "error": f"timeout_retry_{retry}"}
return {"error": "tool_timeout_exhausted", "completed": True}
except PermissionError as e:
return {"error": f"permission_denied: {e}", "completed": True}
except Exception as e:
return {"error": f"unexpected: {e}", "completed": True}
def retry_router(state: ResilientAgentState) -> str:
if state.get("completed"):
return END
if state.get("error") and "timeout_retry" in state["error"]:
return "tool_executor" # retry
if state.get("error"):
return "error_handler" # terminal failure → escalate
return "synthesizer" # success → generate answer
def error_handler(state: ResilientAgentState) -> ResilientAgentState:
print(f"Agent failed: {state['error']}. Escalating to human review.")
return {"completed": True}
def synthesizer(state: ResilientAgentState) -> ResilientAgentState:
answer = llm.invoke(f"Synthesize these results: {state['results']}")
return {"final_answer": answer.content, "completed": True}
11. Human-in-the-Loop Checkpoints
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END
class HITLState(TypedDict):
request: str
draft_action: str | None
human_approved: bool | None
final_result: str | None
def plan_action(state: HITLState) -> HITLState:
"""Agent plans the action but does not execute it yet."""
draft = llm.invoke(f"Plan an action for: {state['request']}").content
return {"draft_action": draft}
def human_approval_gate(state: HITLState) -> str:
"""Pause here for human review."""
if state.get("human_approved") is True:
return "execute"
if state.get("human_approved") is False:
return "reject"
return "await_approval" # not yet decided → pause
def execute_action(state: HITLState) -> HITLState:
result = actually_execute(state["draft_action"])
return {"final_result": result}
def reject_action(state: HITLState) -> HITLState:
return {"final_result": "Action rejected by human reviewer."}
memory = MemorySaver()
graph = StateGraph(HITLState)
graph.add_node("plan", plan_action)
graph.add_node("execute", execute_action)
graph.add_node("reject", reject_action)
graph.set_entry_point("plan")
graph.add_conditional_edges("plan", human_approval_gate, {
"execute": "execute",
"reject": "reject",
"await_approval": END, # pause and return to caller
})
graph.add_edge("execute", END)
graph.add_edge("reject", END)
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "approval-flow-001"}}
# Step 1: Run until approval needed
state = app.invoke({"request": "Delete all records from Q1 2024", "human_approved": None}, config)
print("Proposed action:", state["draft_action"])
# Step 2: Human reviews and approves/rejects
snapshot = app.get_state(config)
app.update_state(config, {"human_approved": False}) # human says no
# Step 3: Resume
final = app.invoke(None, config)
print("Result:", final["final_result"]) # → "Action rejected by human reviewer."
12. Observability and Agent Tracing
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from opentelemetry import trace
import time
tracer = trace.get_tracer("agentic_ai")
class TracedAgentExecutor:
"""Wraps an agent executor with distributed tracing."""
def __init__(self, agent, tools: list):
self.agent = agent
self.tools = {t.name: t for t in tools}
def run(self, task: str, session_id: str) -> dict:
with tracer.start_as_current_span("agent.run") as root_span:
root_span.set_attributes({
"agent.task": task[:200],
"agent.session_id": session_id,
})
steps = []
t_start = time.perf_counter()
# Each tool call gets its own span
for step in self.agent.iter(task):
if hasattr(step, "tool"):
with tracer.start_as_current_span(f"tool.{step.tool}") as tool_span:
tool_span.set_attributes({
"tool.name": step.tool,
"tool.input": str(step.tool_input)[:500],
})
observation = self.tools[step.tool].run(step.tool_input)
tool_span.set_attribute("tool.output", str(observation)[:500])
steps.append({"tool": step.tool, "result": observation})
elapsed = (time.perf_counter() - t_start) * 1000
root_span.set_attributes({
"agent.steps_count": len(steps),
"agent.latency_ms": elapsed,
})
return {"steps": steps, "latency_ms": elapsed}
Challenges and Ethical Considerations
Technical Challenges
- Design Complexity: Managing the combinatorial explosion of possible states in multi-step workflows
- Loop Prevention: Agents can enter infinite reasoning loops without proper loop detection and retry limits
- Tool Safety: Every tool invocation is a potential side effect — validate before execution
- Explainability: Multi-step agent traces must be auditable for debugging and compliance
- Scalability: Parallel agent execution requires careful resource budgeting
Ethical and Safety Issues
- Responsibility: Who is accountable when an autonomous agent causes harm?
- Bias Propagation: Biased tools or retrieval can amplify bias at each agent step
- Transparency: Users must know they are interacting with an autonomous system
- Human Control: High-stakes operations should always have a human-in-the-loop gate
- Data Privacy: Agent tools often access sensitive systems — permissions must be scoped narrowly
The Future of Agentic AI
Agentic AI is poised to revolutionize numerous sectors:
- Advanced Conversational Agents: Assistants capable of managing complex multi-step workflows across days or weeks
- Scientific Research: Agents autonomously designing and running computational experiments
- Software Engineering: End-to-end coding agents that write, test, and deploy code
- Enterprise Automation: Agents handling multi-system workflows (CRM, ERP, support ticketing)
Emerging Trends
- LLM-Based Agents: Reasoning models (o1, o3, DeepSeek-R1) dramatically improve multi-hop reasoning
- Memory Systems: Long-term episodic and semantic memory enabling persistent agent identities
- Computer Use: Agents that directly manipulate UIs and browsers (Claude Computer Use, OpenAI Operator)
- Multi-Agent Collaboration: Hierarchical agent organizations with specialist and generalist roles
- Evaluation Frameworks: Standardized benchmarks (GAIA, AgentBench, SWE-bench) for measuring real agentic capability
Agentic AI Evaluation Framework
Evaluating agentic systems requires different metrics than single-turn LLM evaluation:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
from dataclasses import dataclass
from typing import Optional
@dataclass
class AgentEvalCase:
task: str
expected_outcome: str
required_tools: list[str] # tools that MUST be used
forbidden_tools: list[str] # tools that MUST NOT be used
max_steps: int = 10
time_budget_s: float = 30.0
@dataclass
class AgentEvalResult:
case_id: str
success: bool
steps_taken: int
tools_used: list[str]
time_elapsed_s: float
task_completed: bool
required_tools_used: bool
forbidden_tools_used: bool
final_answer: str
class AgentEvaluator:
def __init__(self, agent, judge_llm):
self.agent = agent
self.judge = judge_llm
def evaluate(self, case: AgentEvalCase, case_id: str = "0") -> AgentEvalResult:
import time
t0 = time.perf_counter()
result = self.agent.invoke({
"messages": [{"role": "user", "content": case.task}],
"retry_count": 0,
})
elapsed = time.perf_counter() - t0
steps = len([m for m in result.get("messages", []) if hasattr(m, "tool_calls")])
tools_used = [tc.function.name for m in result.get("messages", [])
if hasattr(m, "tool_calls") and m.tool_calls
for tc in m.tool_calls]
final_answer = result["messages"][-1].content if result.get("messages") else ""
# Judge task completion
judge_prompt = f"""Did this agent successfully complete the task?
Task: {case.task}
Expected outcome: {case.expected_outcome}
Agent answer: {final_answer[:500]}
Respond: correct or incorrect"""
completed = "correct" in self.judge.invoke(judge_prompt).content.lower()
return AgentEvalResult(
case_id=case_id,
success=completed and not any(t in tools_used for t in case.forbidden_tools),
steps_taken=steps,
tools_used=tools_used,
time_elapsed_s=round(elapsed, 2),
task_completed=completed,
required_tools_used=all(t in tools_used for t in case.required_tools),
forbidden_tools_used=any(t in tools_used for t in case.forbidden_tools),
final_answer=final_answer[:200],
)
def benchmark(self, cases: list[AgentEvalCase]) -> dict:
results = [self.evaluate(c, str(i)) for i, c in enumerate(cases)]
return {
"total": len(results),
"success_rate": sum(r.success for r in results) / len(results),
"task_completion": sum(r.task_completed for r in results) / len(results),
"mean_steps": sum(r.steps_taken for r in results) / len(results),
"mean_time_s": sum(r.time_elapsed_s for r in results) / len(results),
"required_tool_rate": sum(r.required_tools_used for r in results) / len(results),
}
Agent Benchmarks
| Benchmark | Focus | Agent Capability Tested |
|---|---|---|
| GAIA | General assistant tasks | Tool use, multi-step reasoning |
| SWE-bench | Software engineering | Code editing, debugging |
| WebArena | Web navigation | Browser control, form completion |
| AgentBench | Multi-domain | Database, OS, game, web tasks |
| τ-bench | Tool agent | API tool calling accuracy |
| HumanEval-Agent | Coding agents | Test-driven development |
State-of-the-art agents (GPT-4o + tools) achieve ~70% on GAIA Level 1 tasks and ~20-30% on SWE-bench, showing there is still significant headroom for improvement in complex multi-step reasoning and planning.
Agent Reliability Patterns
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
# Pattern 1: Timeout wrapper for async agent execution
import asyncio
from functools import wraps
def with_timeout(seconds: float):
def decorator(fn):
@wraps(fn)
async def wrapper(*args, **kwargs):
try:
return await asyncio.wait_for(
asyncio.ensure_future(fn(*args, **kwargs)),
timeout=seconds
)
except asyncio.TimeoutError:
return {"error": f"Agent timed out after {seconds}s", "completed": False}
return wrapper
return decorator
# Pattern 2: Cost guard for agent runs
class CostGuard:
def __init__(self, max_cost_usd: float = 0.10):
self.max_cost = max_cost_usd
self.spent = 0.0
def record(self, tokens: int, cost_per_1k: float = 0.0025):
self.spent += tokens * cost_per_1k / 1000
if self.spent > self.max_cost:
raise RuntimeError(
f"Agent budget exceeded: ${self.spent:.4f} > ${self.max_cost}"
)
def remaining(self) -> float:
return max(0.0, self.max_cost - self.spent)
# Pattern 3: Idempotent tool execution
import hashlib, json
_executed_tools: dict[str, str] = {} # call_hash → result
def idempotent_tool(tool_fn):
"""Ensure identical tool calls return cached results (safe for retries)."""
@wraps(tool_fn)
def wrapper(*args, **kwargs):
call_hash = hashlib.md5(json.dumps({"args": args, "kwargs": kwargs}).encode()).hexdigest()
if call_hash in _executed_tools:
return _executed_tools[call_hash] # replay cached result
result = tool_fn(*args, **kwargs)
_executed_tools[call_hash] = result
return result
return wrapper
@idempotent_tool
def delete_record(record_id: str) -> str:
"""Safe for retry: idempotent delete."""
return f"Deleted {record_id}"
Conclusion
Agentic AI represents a major paradigm shift in our conception of artificial intelligence. By transitioning from passive tools to autonomous partners capable of reasoning, learning, and acting, we are building systems of unprecedented sophistication.
The twelve design patterns in this article — from Reactive and Goal-Oriented to ReAct, LangGraph, and Human-in-the-Loop — provide the vocabulary and structure to build agentic systems that are powerful without being unpredictable. Explicit control flow, comprehensive observability, and benchmark-driven evaluation are what transform an impressive demo into a production-ready product.
Core Properties of Agentic Systems
- Autonomy — The system acts without step-by-step human direction
- Persistence — State is maintained across multiple steps and sessions
- Goal-directedness — The system works toward a defined objective, adapting its approach as needed
Key Engineering Disciplines
- Explicit control flow — Use LangGraph graphs, not ad-hoc loops
- Bounded autonomy — Enforce iteration limits, cost caps, and time budgets
- Human oversight gates — Required for all irreversible operations
- Full observability — Every step traced, every tool call logged
- Benchmark-driven evaluation — Measure task success rate, not just output quality
Robust error handling and human oversight are not optional — they are the difference between a useful autonomous system and an unpredictable one. The future belongs to systems capable not only of executing tasks but of understanding contexts, anticipating needs, and intelligently collaborating with humans — while remaining safe, auditable, and controllable.
