Advanced Training
Building
AI Agents
Design, build, and deploy autonomous AI agents. Multi-agent orchestration,
tool use, memory systems, and production deployment patterns. Build agents
that plan, reason, and act with minimal human intervention.
Course Overview
Agentic AI represents the next evolution of AI applications. Instead of single-shot prompts,
agents set goals, plan steps, use tools, and iterate until tasks are complete. This course
teaches you to build production-ready agent systems.
1
Agent Fundamentals
The ReAct pattern, tool use, and how agents reason through multi-step tasks.
ReAct
Tool Calling
Planning
2
Memory Systems
Short-term, long-term, and episodic memory. Vector stores and retrieval strategies.
Working Memory
Vector DBs
Summarization
3
Multi-Agent Systems
Orchestration patterns, agent communication, and collaborative problem solving.
CrewAI
AutoGen
Swarm
4
LangGraph Workflows
State machines for agents. Conditional edges, cycles, and human-in-the-loop.
State Graphs
Checkpointing
Streaming
5
Production Patterns
Error handling, observability, rate limiting, and graceful degradation.
LangSmith
Tracing
Guardrails
6
Real-World Project
Build a complete content generation pipeline with multi-agent orchestration.
Full Pipeline
Deployment
Monitoring
MVVM Architecture for Agents
Agent systems benefit from the Model-View-ViewModel pattern. Separate state management
from business logic and presentation for testable, maintainable agent code.
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β VIEW (API/UI) β
β - FastAPI endpoints returning ViewModels β
β - WebSocket connections for streaming β
β - React/Vue components for agent dashboards β
βββββββββββββββββββββββ¬ββββββββββββββββββββββββββββββββββββββββββββ
β Data Binding
βΌ
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β VIEWMODEL β
β - AgentStateViewModel: current step, progress, outputs β
β - Commands: StartTask, PauseAgent, ApproveStep β
β - Computed properties: can_proceed, estimated_completion β
β - Observable notifications (Observer pattern) β
βββββββββββββββββββββββ¬ββββββββββββββββββββββββββββββββββββββββββββ
β
βΌ
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β MODEL β
β - Agent business logic (LangGraph state machine) β
β - Tool implementations β
β - Memory systems (vector stores, conversation history) β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
from pydantic import BaseModel, computed_field
from enum import Enum
from typing import Callable
class AgentStatus(str, Enum):
IDLE = "idle"
PLANNING = "planning"
EXECUTING = "executing"
WAITING_APPROVAL = "waiting_approval"
COMPLETED = "completed"
ERROR = "error"
class AgentViewModel(BaseModel):
"""ViewModel: Observable state for agent UI"""
task_id: str
status: AgentStatus
current_step: int
total_steps: int
current_action: str | None = None
outputs: list[dict] = []
error_message: str | None = None
@computed_field
@property
def progress_percent(self) -> float:
if self.total_steps == 0:
return 0.0
return (self.current_step / self.total_steps) * 100
@computed_field
@property
def can_proceed(self) -> bool:
return self.status == AgentStatus.WAITING_APPROVAL
@computed_field
@property
def status_display(self) -> str:
displays = {
AgentStatus.IDLE: "Ready to start",
AgentStatus.PLANNING: "Planning steps...",
AgentStatus.EXECUTING: f"Step {self.current_step}/{self.total_steps}",
AgentStatus.WAITING_APPROVAL: "Awaiting your approval",
AgentStatus.COMPLETED: "Task completed",
AgentStatus.ERROR: f"Error: {self.error_message}"
}
return displays[self.status]
Behavioral Design Patterns
Agent systems are built on behavioral patterns. These patterns define how agents
communicate, make decisions, and manage state transitions.
State
Agent behavior changes based on internal state: planning, executing, waiting for approval, error recovery. Each state has different allowed transitions.
Command
Encapsulate agent actions as objects. Enable undo/redo, queuing, logging, and human approval workflows for sensitive operations.
Observer
Notify UI and logging systems of agent state changes. Stream progress updates, tool calls, and outputs in real-time via WebSockets.
Chain of Responsibility
Process agent requests through validation, rate limiting, authentication, and logging chains before execution.
Strategy
Swap planning algorithms, tool selection strategies, and error recovery approaches without changing agent code.
Mediator
Coordinate communication between multiple agents. The orchestrator mediates instead of agents talking directly.
from abc import ABC, abstractmethod
class AgentState(ABC):
@abstractmethod
async def execute(self, context: "AgentContext"): pass
@abstractmethod
async def handle_error(self, context: "AgentContext", error: Exception): pass
class PlanningState(AgentState):
async def execute(self, context):
plan = await context.planner.create_plan(context.task)
context.set_plan(plan)
context.transition_to(ExecutingState())
async def handle_error(self, context, error):
context.transition_to(ErrorState(str(error)))
class ExecutingState(AgentState):
async def execute(self, context):
step = context.get_next_step()
if step.requires_approval:
context.transition_to(WaitingApprovalState(step))
return
result = await context.execute_step(step)
context.record_output(result)
if context.is_complete():
context.transition_to(CompletedState())
class WaitingApprovalState(AgentState):
def __init__(self, pending_step):
self.pending_step = pending_step
async def execute(self, context):
pass
async def approve(self, context):
result = await context.execute_step(self.pending_step)
context.record_output(result)
context.transition_to(ExecutingState())
Tool System with Command Pattern
Key Concept
Every tool call in an agent system should be a Command object. This enables
logging, approval workflows, undo operations, and retry logic without
modifying tool implementations.
from abc import ABC, abstractmethod
from datetime import datetime
class ToolCommand(ABC):
"""Command pattern for agent tool calls"""
def __init__(self):
self.executed_at: datetime | None = None
self.result: Any = None
self.error: Exception | None = None
@abstractmethod
async def execute(self) -> Any:
pass
@abstractmethod
async def undo(self) -> None:
pass
@property
def requires_approval(self) -> bool:
return False
class GenerateImageCommand(ToolCommand):
def __init__(self, prompt: str, generator):
super().__init__()
self.prompt = prompt
self.generator = generator
self.image_path: str | None = None
async def execute(self):
self.executed_at = datetime.now()
self.image_path = await self.generator.generate(self.prompt)
self.result = self.image_path
return self.result
async def undo(self):
if self.image_path:
await self.generator.delete(self.image_path)
class PublishContentCommand(ToolCommand):
"""Sensitive operation - requires human approval"""
@property
def requires_approval(self) -> bool:
return True
async def execute(self):
return await self.publisher.publish(self.content)
Multi-Agent Orchestration
| Pattern |
Description |
Use Case |
| Sequential Pipeline |
Agents execute in order, each receiving output from previous |
Content pipeline: research β write β edit β publish |
| Parallel Fan-Out |
Multiple agents work simultaneously on different subtasks |
Generate images, audio, and text in parallel |
| Supervisor |
One agent delegates and reviews work from others |
Manager agent assigns tasks to specialist agents |
| Debate |
Multiple agents argue positions, synthesize consensus |
Code review with multiple perspectives |
| Reflection |
Agent reviews and critiques its own output |
Self-improving content generation |
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
class PipelineState(TypedDict):
topic: str
script: str
image_prompt: str
image_path: str
audio_path: str
errors: Annotated[list[str], operator.add]
async def script_agent(state: PipelineState) -> dict:
"""Generate script from topic"""
script = await llm.generate_script(state["topic"])
return {"script": script}
async def image_agent(state: PipelineState) -> dict:
"""Generate image from script"""
prompt = await llm.extract_visual(state["script"])
image = await comfyui.generate(prompt)
return {"image_prompt": prompt, "image_path": image}
async def voice_agent(state: PipelineState) -> dict:
"""Synthesize voice from script"""
audio = await tts.synthesize(state["script"])
return {"audio_path": audio}
workflow = StateGraph(PipelineState)
workflow.add_node("script", script_agent)
workflow.add_node("image", image_agent)
workflow.add_node("voice", voice_agent)
workflow.set_entry_point("script")
workflow.add_edge("script", "image")
workflow.add_edge("script", "voice")
workflow.add_edge("image", END)
workflow.add_edge("voice", END)
pipeline = workflow.compile()
Hands-On Projects
- Build a ReAct agent with tool use using LangChain
- Implement the State pattern for agent lifecycle management
- Create a Command-based tool system with undo and approval
- Build an AgentViewModel with observable state updates
- Design a multi-agent content pipeline with LangGraph
- Add human-in-the-loop approval for sensitive operations
- Implement the Observer pattern for real-time UI updates
- Deploy an agent with LangSmith tracing and monitoring
Ready to Build Autonomous Agents?
Master the patterns and frameworks that power production AI systems.
Continue with RAG & Knowledge Systems to give your agents access to your data.
Enroll Now