Stage 05 — AI Agents and Agentic Workflows
Building AI That Takes Action · Technical + Architecture · ⏱ 10–14 hours
Learning Objectives
By the end of this stage you will be able to:
- Explain the agent loop and how LLMs make decisions sequentially
- Build tool-using agents from scratch with Anthropic's tool use API
- Implement ReAct pattern (Reason + Act) for multi-step tasks
- Design multi-agent systems where agents collaborate
- Handle agent failures, loops, and safety guardrails
- Use LangChain/LangGraph for production agent workflows
- Understand when to use agents vs. simpler approaches
Section 1: What Is an AI Agent?
An agent is an LLM that can take actions in the world — not just respond with text, but call functions, read files, search the web, execute code, and interact with external services. Crucially, agents operate in a loop: they reason about what to do, take an action, observe the result, and decide what to do next.
The Agent Loop
┌─────────────────────────────────────┐
│ AGENT LOOP │
│ │
│ User Task ──> Think: What do I │
│ need to do next? │
│ │ │
│ ▼ │
│ Choose Action │
│ (tool call) │
│ │ │
│ ▼ │
│ Execute Action │
│ (run the tool) │
│ │ │
│ ▼ │
│ Observe Result │
│ │ │
│ ▼ │
│ Done? ──No─> Think again │
│ │ │
│ Yes │
│ │ │
│ ▼ │
│ Return Final Answer │
└─────────────────────────────────────┘This loop runs until the model decides it has enough information to answer, or until a stop condition (max iterations, error, user interrupt) is hit.
Section 2: Why Agents Are Powerful (and Dangerous)
What Agents Enable
- Complex reasoning: Multi-step research that requires synthesizing multiple sources
- Dynamic tool use: Choosing which tool to call based on intermediate results
- Long-horizon tasks: Tasks that require many steps to complete (e.g., "Research competitors and write a report")
- Parallelization: Multiple agents working on different subtasks simultaneously
Why Agents Fail
- Loops: Agent calls the same tool repeatedly, making no progress
- Hallucinated tool calls: Model invents tool parameters that don't exist
- Goal drift: Agent pursues an intermediate goal and loses sight of the original task
- Cost explosion: An unconstrained agent can make hundreds of API calls
- Security: Prompt injection attacks through tool results
When NOT to Use Agents
Agents are not the default choice. Use them only when:
- The task genuinely requires multiple steps where later steps depend on earlier results
- You don't know in advance which sequence of tools to call
- Single LLM calls with long prompts can't solve it
For most tasks, a simple prompt or a two-step chain is better.
Section 3: Building an Agent from Scratch
The Core Pattern: ReAct
ReAct (Reason + Act) prompts the model to alternately reason ("Thought: I need to find...") and act ("Action: search(...)") until it reaches a final answer.
import anthropic
import json
import re
from typing import Callable
client = anthropic.Anthropic()
# Tool definitions
tools = [
{
"name": "web_search",
"description": "Search the web for current information. Use for recent events, current prices, or facts you're uncertain about.",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query"
}
},
"required": ["query"]
}
},
{
"name": "calculator",
"description": "Evaluate a mathematical expression. Input must be a valid Python math expression.",
"input_schema": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Mathematical expression to evaluate, e.g. '2 ** 10 + 500'"
}
},
"required": ["expression"]
}
},
{
"name": "read_file",
"description": "Read the contents of a file from disk.",
"input_schema": {
"type": "object",
"properties": {
"filepath": {
"type": "string",
"description": "Relative path to the file"
}
},
"required": ["filepath"]
}
}
]
# Tool implementations
def web_search(query: str) -> str:
"""Simulated web search — replace with real API (SerpAPI, Tavily, etc.)"""
# In production: call Tavily, SerpAPI, or similar
return f"[Simulated search results for '{query}': Results would appear here from a real search API]"
def calculator(expression: str) -> str:
"""Safe math evaluation."""
# Allowlist safe operations only
allowed = set("0123456789.+-*/()** ")
if not all(c in allowed for c in expression):
return "Error: Expression contains disallowed characters"
try:
result = eval(expression, {"__builtins__": {}})
return str(result)
except Exception as e:
return f"Error: {e}"
def read_file(filepath: str) -> str:
"""Read a file, with path sanitization."""
# Sanitize to prevent directory traversal
safe_path = filepath.lstrip("/").replace("..", "")
try:
with open(safe_path, "r") as f:
return f.read()[:5000] # Limit output
except FileNotFoundError:
return f"File not found: {safe_path}"
except Exception as e:
return f"Error reading file: {e}"
tool_functions: dict[str, Callable] = {
"web_search": web_search,
"calculator": calculator,
"read_file": read_file
}
def run_agent(task: str, max_iterations: int = 10) -> str:
"""Run agent loop until task is complete or max iterations reached."""
system = """You are a capable AI assistant that can use tools to complete tasks.
Think step-by-step. Use tools when you need to look up information or perform calculations.
When you have enough information to answer completely, provide your final answer directly without using any tools."""
messages = [{"role": "user", "content": task}]
iteration = 0
print(f"Task: {task}")
print("=" * 60)
while iteration < max_iterations:
iteration += 1
print(f"\n[Iteration {iteration}]")
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=2048,
system=system,
tools=tools,
messages=messages
)
# Check stop reason
if response.stop_reason == "end_turn":
# Model is done reasoning
final_text = ""
for block in response.content:
if hasattr(block, "text"):
final_text += block.text
print(f"\nFinal Answer:\n{final_text}")
return final_text
elif response.stop_reason == "tool_use":
# Process tool calls
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
tool_name = block.name
tool_input = block.input
tool_use_id = block.id
print(f" → Calling: {tool_name}({json.dumps(tool_input, indent=None)})")
# Execute the tool
if tool_name in tool_functions:
result = tool_functions[tool_name](**tool_input)
else:
result = f"Error: Unknown tool '{tool_name}'"
print(f" ← Result: {str(result)[:200]}")
tool_results.append({
"type": "tool_result",
"tool_use_id": tool_use_id,
"content": str(result)
})
messages.append({"role": "user", "content": tool_results})
else:
print(f"Unexpected stop reason: {response.stop_reason}")
break
return "Max iterations reached. Task may be incomplete."
# Test the agent
result = run_agent("What is 2^32, and is that number larger or smaller than the number of people on Earth?")Section 4: Multi-Agent Systems
Complex tasks can be broken into subtasks handled by specialized agents.
Orchestrator-Worker Pattern
from dataclasses import dataclass
from typing import Optional
import anthropic
client = anthropic.Anthropic()
@dataclass
class AgentResult:
agent_name: str
task: str
result: str
success: bool
def create_specialist_agent(
name: str,
specialty: str,
available_tools: list[dict]
) -> Callable[[str], AgentResult]:
"""Factory function that creates a specialized agent."""
def run(task: str) -> AgentResult:
system = f"""You are a specialized AI agent: {name}.
Your specialty: {specialty}
Complete the given task using your tools. Be thorough and precise."""
try:
# Simplified agent call (single-turn for this example)
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=2048,
system=system,
tools=available_tools if available_tools else anthropic.NOT_GIVEN,
messages=[{"role": "user", "content": task}]
)
result_text = response.content[0].text if response.content else ""
return AgentResult(agent_name=name, task=task, result=result_text, success=True)
except Exception as e:
return AgentResult(agent_name=name, task=task, result=str(e), success=False)
return run
def orchestrator_agent(user_request: str) -> str:
"""Orchestrator that delegates to specialist agents."""
# Define specialist agents
researcher = create_specialist_agent(
"ResearchAgent",
"Finding and synthesizing information from multiple sources",
[tools[0]] # web_search only
)
analyst = create_specialist_agent(
"AnalysisAgent",
"Data analysis, calculations, and logical reasoning",
[tools[1]] # calculator only
)
writer = create_specialist_agent(
"WriterAgent",
"Writing clear, well-structured reports and summaries",
[] # No tools needed for writing
)
# Step 1: Orchestrator plans the task
plan_response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
system="""You are an orchestrator that breaks complex tasks into subtasks for specialist agents.
Available agents:
- ResearchAgent: Finds information (has web search)
- AnalysisAgent: Does calculations and data analysis
- WriterAgent: Writes reports and summaries
Return a JSON list of subtasks: [{"agent": "name", "task": "description"}]""",
messages=[{"role": "user", "content": f"Break this into subtasks: {user_request}"}]
)
# Parse plan (simplified — in production, use JSON mode)
plan_text = plan_response.content[0].text
# Step 2: Execute subtasks
results = []
# For demo, run a fixed sequence
research_result = researcher(f"Research information needed for: {user_request}")
results.append(research_result)
# Step 3: Synthesize results
synthesis_input = f"""Original request: {user_request}
Research findings:
{research_result.result}
Write a comprehensive response synthesizing these findings."""
final_result = writer(synthesis_input)
return final_result.result
# Example
result = orchestrator_agent("Analyze the pros and cons of Pinecone vs ChromaDB for a startup building a RAG system")
print(result)Section 5: LangChain and LangGraph
For production agents, use LangChain's ecosystem instead of raw API calls.
Installation
pip install langchain langchain-anthropic langgraph langchain-communityLangChain Agent (Simple)
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
# Define tools with decorators
@tool
def search_courses(query: str) -> str:
"""Search TechNodeX courses for relevant content."""
# Connect to your RAG system here
return f"Course results for '{query}': Python Security Stage 3 covers {query}"
@tool
def get_course_progress(course_name: str, user_id: str) -> str:
"""Get a user's progress in a specific course."""
return f"User {user_id} has completed 3/6 stages of {course_name}"
# Create agent
llm = ChatAnthropic(model="claude-sonnet-4-5")
tools = [search_courses, get_course_progress]
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful TechNodeX learning assistant."),
("human", "{input}"),
("placeholder", "{agent_scratchpad}")
])
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=5)
result = agent_executor.invoke({
"input": "I want to learn Python for security. What stage should I start at?"
})
print(result["output"])LangGraph for Complex Workflows
LangGraph builds on LangChain to create stateful, multi-step agent workflows as graphs.
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
import operator
# Define the agent state
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], operator.add]
current_step: str
research_results: list[str]
final_answer: str
# Define nodes
llm = ChatAnthropic(model="claude-sonnet-4-5")
def research_node(state: AgentState) -> dict:
"""Node: Research phase."""
messages = state["messages"]
response = llm.invoke(messages + [
HumanMessage(content="What information do you need to answer this? List 3 specific things to research.")
])
return {
"messages": [response],
"current_step": "research",
"research_results": [response.content]
}
def answer_node(state: AgentState) -> dict:
"""Node: Final answer synthesis."""
all_context = "\n".join(state["research_results"])
response = llm.invoke([
HumanMessage(content=f"Based on this research:\n{all_context}\n\nAnswer the original question comprehensively.")
])
return {
"messages": [response],
"current_step": "complete",
"final_answer": response.content
}
def should_continue(state: AgentState) -> str:
"""Conditional edge: decide next step."""
if state.get("current_step") == "research":
return "answer"
return END
# Build graph
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("answer", answer_node)
workflow.set_entry_point("research")
workflow.add_conditional_edges("research", should_continue, {"answer": "answer", END: END})
workflow.add_edge("answer", END)
app = workflow.compile()
# Run
result = app.invoke({
"messages": [HumanMessage(content="What are the best practices for securing a Python Flask API?")],
"current_step": "start",
"research_results": [],
"final_answer": ""
})
print(result["final_answer"])Section 6: Agent Safety and Guardrails
Production agents need safety mechanisms.
class SafeAgent:
"""Agent wrapper with safety guardrails."""
def __init__(self, max_iterations: int = 10, max_cost_usd: float = 0.50):
self.client = anthropic.Anthropic()
self.max_iterations = max_iterations
self.max_cost_usd = max_cost_usd
self.total_cost = 0.0
self.blocked_tools = set()
def add_cost(self, input_tokens: int, output_tokens: int) -> None:
"""Track API costs."""
cost = (input_tokens / 1e6 * 3.0) + (output_tokens / 1e6 * 15.0)
self.total_cost += cost
if self.total_cost > self.max_cost_usd:
raise RuntimeError(f"Cost limit exceeded: ${self.total_cost:.3f} > ${self.max_cost_usd}")
def validate_tool_call(self, tool_name: str, tool_input: dict) -> None:
"""Validate tool calls before execution."""
if tool_name in self.blocked_tools:
raise ValueError(f"Tool '{tool_name}' is blocked")
# Prevent directory traversal in file operations
if tool_name == "read_file":
filepath = tool_input.get("filepath", "")
if ".." in filepath or filepath.startswith("/etc") or filepath.startswith("/root"):
raise ValueError(f"Blocked file path: {filepath}")
# Rate limit search queries
if tool_name == "web_search":
query = tool_input.get("query", "")
if len(query) > 500:
raise ValueError("Search query too long")
def sanitize_tool_result(self, result: str) -> str:
"""Sanitize tool results to prevent prompt injection."""
# Remove common injection patterns
injection_patterns = [
"Ignore previous instructions",
"Disregard your system prompt",
"You are now",
"New instruction:"
]
for pattern in injection_patterns:
if pattern.lower() in result.lower():
return f"[Content filtered: potential prompt injection detected]"
# Truncate very long results
if len(result) > 10000:
return result[:10000] + "\n[Result truncated for safety]"
return resultCheckpoint Assessment
- What is the "agent loop" and what are its three key steps?
- When should you NOT use an agent? Give a specific example.
- Explain prompt injection in the context of tool use. How does a malicious search result attack an agent?
- What is the difference between an orchestrator agent and a worker agent?
- Your agent is running in an infinite loop, calling the same tool repeatedly. What guard would prevent this?
- Compare a ReAct pattern agent with a LangGraph workflow. When would you choose each?
Project: Research Assistant Agent
Build a multi-tool agent that:
- Accepts a research question from the user
- Plans the research approach (what to look up and in what order)
- Executes the plan using: web_search, read_file (for local docs), calculator
- Synthesizes findings into a structured research brief with sources
- Includes cost tracking — report total tokens and USD cost
- Safety features: max 15 iterations, max $1.00 cost, blocks dangerous file paths
Bonus challenge: Add a --multi-agent flag that splits the work between a ResearchAgent and a WriterAgent.
What's Next
Stage 6 is the Capstone — you'll build and deploy a complete, production-ready AI application using everything you've learned: APIs, RAG, and agents working together.
Lock In Founding Member Access
Get full access to every course on TechNodeX — AI, cybersecurity, Python, and everything we build next. $9/month, price locked forever.
Become a Founding Member →