MODULE 3 - CHAPTER 2 ⏱️ 50 min read πŸ“– 4,200 words

Agent Architecture & Frameworks

Understanding the agent stack and mastering LangGraph, CrewAI, and AutoGPT

While the previous chapter introduced the conceptual foundation of AI agents, this chapter dives into the practical architecture and frameworks used to build production-grade agent systems. We'll explore the layered architecture that powers modern agents and compare the three leading frameworks dominating the landscape in 2025.

What You'll Learn

  • The five-layer agent architecture stack
  • Memory systems: Short-term vs long-term storage
  • LangGraph: Graph-based stateful workflows
  • CrewAI: Role-based collaborative teams
  • AutoGPT and autonomous execution
  • Framework selection criteria for your projects

1. The AI Agent Stack: Core Components

An AI agent is not a monolithic entityβ€”it's a sophisticated system built from several interconnected components that work in harmony. Understanding this "agent stack" is crucial for designing, building, and debugging effective AI agents.

The Five-Layer Agent Architecture

    β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
    β”‚  LAYER 5: USER INTERFACE                           β”‚
    β”‚  β€’ Chat UI  β€’ API Endpoints  β€’ Integrations        β”‚
    β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
                            β–²
                            β”‚
    β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
    β”‚  LAYER 4: ORCHESTRATION & PLANNING                 β”‚
    β”‚  β€’ Goal Decomposition  β€’ Task Sequencing           β”‚
    β”‚  β€’ ReAct Loop  β€’ Chain-of-Thought  β€’ Tree of Thoughtβ”‚
    β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
                            β–²
                            β”‚
    β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
    β”‚  LAYER 3: LLM ENGINE (The "Brain")                 β”‚
    β”‚  β€’ GPT-4  β€’ Claude 3.5  β€’ Gemini 1.5              β”‚
    β”‚  β€’ Reasoning  β€’ Language Generation  β€’ Tool Callingβ”‚
    β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
                            β–²
                            β”‚
    β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
    β”‚  LAYER 2: MEMORY & STATE MANAGEMENT                β”‚
    β”‚  β€’ Short-term (context window)                     β”‚
    β”‚  β€’ Long-term (vector databases, knowledge graphs)  β”‚
    β”‚  β€’ Conversation history  β€’ Learned preferences    β”‚
    β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
                            β–²
                            β”‚
    β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
    β”‚  LAYER 1: TOOLS & ACTIONS                          β”‚
    β”‚  β€’ Web Search  β€’ Code Execution  β€’ Database Queriesβ”‚
    β”‚  β€’ APIs  β€’ File Operations  β€’ External Services    β”‚
    β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
                    

Layer 1: Tools & Actions

Tools are the agent's hands and eyesβ€”they extend capabilities beyond internal knowledge. Every production agent needs a carefully curated toolkit.

Tool Categories
  • Information Retrieval: Web search (Tavily, Google), database queries, file reading
  • Computation: Code execution, mathematical calculations, data analysis
  • Communication: Email, Slack, SMS, push notifications
  • Integration: External APIs (CRM, calendar, payment systems)
  • Creation: File writing, image generation, report creation

Example 1: Building a Comprehensive Tool Library

from typing import Dict, Any, Callable
import requests
import subprocess
import sqlite3
from datetime import datetime

class AgentToolkit:
    """Comprehensive tool library for AI agents."""

    def __init__(self):
        self.tools = {
            "web_search": self.web_search,
            "calculator": self.calculator,
            "execute_code": self.execute_code,
            "query_database": self.query_database,
            "send_email": self.send_email,
            "get_current_time": self.get_current_time,
            "read_file": self.read_file,
            "write_file": self.write_file
        }

    def get_tool_definitions(self):
        """Return tool definitions for LLM function calling."""
        return [
            {
                "type": "function",
                "function": {
                    "name": "web_search",
                    "description": "Search the web for current information",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "description": "Search query"},
                            "num_results": {"type": "integer", "description": "Number of results (default 5)"}
                        },
                        "required": ["query"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "calculator",
                    "description": "Perform mathematical calculations safely",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "expression": {"type": "string", "description": "Math expression to evaluate"}
                        },
                        "required": ["expression"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "execute_code",
                    "description": "Execute Python code in a sandboxed environment",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "code": {"type": "string", "description": "Python code to execute"},
                            "timeout": {"type": "integer", "description": "Timeout in seconds (default 10)"}
                        },
                        "required": ["code"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "query_database",
                    "description": "Query SQLite database with SQL",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "sql": {"type": "string", "description": "SQL query to execute"}
                        },
                        "required": ["sql"]
                    }
                }
            }
        ]

    def web_search(self, query: str, num_results: int = 5) -> str:
        """Search the web using Tavily API."""
        try:
            # In production: use real API
            # response = requests.post(
            #     "https://api.tavily.com/search",
            #     json={"query": query, "max_results": num_results}
            # )
            return f"[Web Search Results for '{query}']: Mock results - implement with Tavily API"
        except Exception as e:
            return f"Search error: {str(e)}"

    def calculator(self, expression: str) -> str:
        """Safely evaluate mathematical expressions."""
        try:
            # Security: Only allow mathematical operations
            import ast
            import operator

            operators = {
                ast.Add: operator.add,
                ast.Sub: operator.sub,
                ast.Mult: operator.mul,
                ast.Div: operator.truediv,
                ast.Pow: operator.pow,
                ast.Mod: operator.mod
            }

            def eval_expr(node):
                if isinstance(node, ast.Num):
                    return node.n
                elif isinstance(node, ast.BinOp):
                    return operators[type(node.op)](eval_expr(node.left), eval_expr(node.right))
                elif isinstance(node, ast.UnaryOp):
                    return operators[type(node.op)](eval_expr(node.operand))
                else:
                    raise TypeError(node)

            result = eval_expr(ast.parse(expression, mode='eval').body)
            return f"Result: {result}"
        except Exception as e:
            return f"Calculation error: {str(e)}"

    def execute_code(self, code: str, timeout: int = 10) -> str:
        """Execute Python code in sandboxed environment."""
        try:
            # In production: use Docker or secure sandbox
            result = subprocess.run(
                ["python", "-c", code],
                capture_output=True,
                text=True,
                timeout=timeout
            )

            if result.returncode == 0:
                return f"Output:\n{result.stdout}"
            else:
                return f"Error:\n{result.stderr}"
        except subprocess.TimeoutExpired:
            return "Error: Code execution timeout"
        except Exception as e:
            return f"Execution error: {str(e)}"

    def query_database(self, sql: str) -> str:
        """Execute SQL query on SQLite database."""
        try:
            conn = sqlite3.connect('agent_database.db')
            cursor = conn.cursor()
            cursor.execute(sql)

            if sql.strip().upper().startswith('SELECT'):
                results = cursor.fetchall()
                conn.close()
                return f"Results: {results}"
            else:
                conn.commit()
                conn.close()
                return f"Query executed successfully"
        except Exception as e:
            return f"Database error: {str(e)}"

    def send_email(self, to: str, subject: str, body: str) -> str:
        """Send email via SMTP."""
        try:
            # In production: use SendGrid or AWS SES
            return f"Email sent to {to}: {subject}"
        except Exception as e:
            return f"Email error: {str(e)}"

    def get_current_time(self) -> str:
        """Get current date and time."""
        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    def read_file(self, filepath: str) -> str:
        """Read file contents."""
        try:
            with open(filepath, 'r') as f:
                content = f.read()
            return f"File content:\n{content}"
        except Exception as e:
            return f"File read error: {str(e)}"

    def write_file(self, filepath: str, content: str) -> str:
        """Write content to file."""
        try:
            with open(filepath, 'w') as f:
                f.write(content)
            return f"File written successfully: {filepath}"
        except Exception as e:
            return f"File write error: {str(e)}"

    def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> str:
        """Execute specified tool with arguments."""
        if tool_name not in self.tools:
            return f"Error: Unknown tool '{tool_name}'"

        tool_function = self.tools[tool_name]
        return tool_function(**arguments)

# Usage
toolkit = AgentToolkit()
print(toolkit.execute_tool("calculator", {"expression": "123 * 456"}))
print(toolkit.execute_tool("get_current_time", {}))

Layer 2: Memory & State Management

Memory is what transforms a stateless LLM into an agent that can learn, remember, and adapt over time. Modern agents employ multiple types of memory for different purposes.

Memory Types
  • Short-Term Memory (Working Memory): Holds immediate context for the current interaction (conversation history, recent observations)
  • Long-Term Memory (Persistent Memory): Stores information across sessions (user preferences, learned facts, past experiences)
  • Episodic Memory: Recalls specific events ("Last time user asked about X, they needed Y")
  • Semantic Memory: Factual knowledge accumulated over time
  • Procedural Memory: Learned skills and patterns

Example 2: Multi-Tier Memory System

from typing import List, Dict, Any
import chromadb
from datetime import datetime
import json

class AgentMemory:
    """Multi-tier memory system for AI agents."""

    def __init__(self, user_id: str):
        self.user_id = user_id

        # Short-term memory (in-memory)
        self.short_term = []
        self.max_short_term = 20  # Last 20 messages

        # Long-term memory (vector database)
        self.chroma_client = chromadb.Client()
        self.long_term = self.chroma_client.get_or_create_collection(
            name=f"agent_memory_{user_id}"
        )

        # Semantic facts (key-value store)
        self.facts = {}

    def add_to_short_term(self, role: str, content: str):
        """Add message to short-term memory."""
        self.short_term.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        })

        # Keep only recent messages
        if len(self.short_term) > self.max_short_term:
            # Move oldest to long-term before removing
            oldest = self.short_term.pop(0)
            self.add_to_long_term(
                content=oldest["content"],
                metadata={"role": oldest["role"], "timestamp": oldest["timestamp"]}
            )

    def get_short_term(self) -> List[Dict[str, str]]:
        """Retrieve short-term memory (recent conversation)."""
        return self.short_term

    def add_to_long_term(self, content: str, metadata: Dict[str, Any]):
        """Add to long-term memory (vector database)."""
        doc_id = f"{self.user_id}_{datetime.now().timestamp()}"

        self.long_term.add(
            documents=[content],
            metadatas=[metadata],
            ids=[doc_id]
        )

    def query_long_term(self, query: str, n_results: int = 5) -> List[Dict[str, Any]]:
        """Semantic search over long-term memory."""
        results = self.long_term.query(
            query_texts=[query],
            n_results=n_results
        )

        return results

    def add_fact(self, key: str, value: Any):
        """Store semantic fact."""
        self.facts[key] = {
            "value": value,
            "timestamp": datetime.now().isoformat()
        }

    def get_fact(self, key: str) -> Any:
        """Retrieve semantic fact."""
        return self.facts.get(key, {}).get("value")

    def get_all_facts(self) -> Dict[str, Any]:
        """Retrieve all stored facts."""
        return self.facts

    def get_full_context(self, current_query: str) -> str:
        """Build comprehensive context for LLM."""
        context_parts = []

        # Add relevant long-term memories
        context_parts.append("=== RELEVANT PAST INTERACTIONS ===")
        long_term_results = self.query_long_term(current_query, n_results=3)

        if long_term_results["documents"]:
            for i, doc in enumerate(long_term_results["documents"][0]):
                context_parts.append(f"{i+1}. {doc}")
        else:
            context_parts.append("(No relevant past interactions)")

        # Add known facts
        context_parts.append("\n=== KNOWN FACTS ABOUT USER ===")
        if self.facts:
            for key, data in self.facts.items():
                context_parts.append(f"- {key}: {data['value']}")
        else:
            context_parts.append("(No facts stored yet)")

        # Add short-term memory
        context_parts.append("\n=== CURRENT CONVERSATION ===")
        for msg in self.short_term:
            context_parts.append(f"{msg['role']}: {msg['content']}")

        return "\n".join(context_parts)

    def save_to_disk(self, filepath: str):
        """Persist memory to disk."""
        memory_state = {
            "user_id": self.user_id,
            "short_term": self.short_term,
            "facts": self.facts
        }

        with open(filepath, 'w') as f:
            json.dump(memory_state, f, indent=2)

    def load_from_disk(self, filepath: str):
        """Load memory from disk."""
        with open(filepath, 'r') as f:
            memory_state = json.load(f)

        self.short_term = memory_state["short_term"]
        self.facts = memory_state["facts"]

# Usage
memory = AgentMemory(user_id="user123")

# Add to short-term
memory.add_to_short_term("user", "I love Python programming")
memory.add_to_short_term("assistant", "Great! What aspects of Python do you enjoy most?")

# Store a fact
memory.add_fact("preferred_language", "Python")
memory.add_fact("skill_level", "Advanced")

# Query long-term memory
results = memory.query_long_term("What does the user like?")

# Get full context
context = memory.get_full_context("Tell me about my preferences")
print(context)

Layer 3: LLM Engine (The Brain)

The LLM serves as the agent's cognitive engine, responsible for interpretation, reasoning, planning, and language generation. In 2025, the top choices are GPT-4, Claude 3.5 Sonnet, and Gemini 1.5 Pro.

GPT-4 (OpenAI)

  • βœ“ Excellent function calling
  • βœ“ Strong code generation
  • βœ“ 128K context window
  • βœ“ Vision capabilities
  • ⚠ Higher cost

Claude 3.5 Sonnet (Anthropic)

  • βœ“ Superior reasoning
  • βœ“ 200K context window
  • βœ“ Computer use (control UI)
  • βœ“ Strong safety
  • ⚠ Function calling newer

Gemini 1.5 Pro (Google)

  • βœ“ 1M context window
  • βœ“ Multimodal (video, audio)
  • βœ“ Fast inference
  • βœ“ Lower cost
  • ⚠ Newer to market

Layer 4: Orchestration & Planning

This layer coordinates the agent's behavior, implementing the ReAct loop, managing task sequences, and handling complex planning strategies like Chain-of-Thought and Tree of Thought.

Layer 5: User Interface

The final layer handles interaction with users through chat UIs, API endpoints, voice interfaces, or integrations with existing platforms (Slack, Discord, web apps).

2. LangGraph: Graph-Based Stateful Workflows

LangGraph, an extension of the LangChain ecosystem, is designed for building stateful, multi-actor applications using a graph-based architecture. It provides explicit control over execution flow, making it ideal for complex, non-linear workflows with dynamic branching and robust state management.

Core Concepts

LangGraph Architecture
  • Nodes: Represent agents, tools, or processing steps
  • Edges: Define transitions between nodes (conditional or unconditional)
  • State: Shared data passed between nodes, can be checkpointed
  • Graphs: The overall workflow structure connecting nodes via edges

Example 3: LangGraph Agent with Self-Correction

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor, ToolInvocation
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from typing import TypedDict, Annotated, Sequence
import operator

# Define tools
@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    return f"Search results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Calculate mathematical expressions."""
    try:
        result = eval(expression)
        return f"Result: {result}"
    except Exception as e:
        return f"Error: {str(e)}"

# Define state
class AgentState(TypedDict):
    messages: Annotated[Sequence[dict], operator.add]
    user_query: str
    final_answer: str
    iteration: int
    max_iterations: int

# Initialize LLM and tools
llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [web_search, calculator]
tool_executor = ToolExecutor(tools)

# Bind tools to LLM
llm_with_tools = llm.bind_tools(tools)

# Define agent node
def agent_node(state: AgentState):
    """Agent reasoning node."""
    messages = state["messages"]
    iteration = state["iteration"]

    # Get LLM response
    response = llm_with_tools.invoke(messages)

    # Add response to messages
    return {
        "messages": [response],
        "iteration": iteration + 1
    }

# Define tool execution node
def tool_node(state: AgentState):
    """Execute tools requested by agent."""
    messages = state["messages"]
    last_message = messages[-1]

    tool_results = []

    if hasattr(last_message, "tool_calls"):
        for tool_call in last_message.tool_calls:
            # Execute tool
            result = tool_executor.invoke(
                ToolInvocation(
                    tool=tool_call["name"],
                    tool_input=tool_call["args"]
                )
            )

            # Create tool message
            tool_message = {
                "role": "tool",
                "content": str(result),
                "name": tool_call["name"]
            }
            tool_results.append(tool_message)

    return {"messages": tool_results}

# Define verification node
def verification_node(state: AgentState):
    """Verify the agent's answer before finalizing."""
    messages = state["messages"]
    last_message = messages[-1]

    verification_prompt = f"""
    Verify this answer is correct and complete:
    {last_message.content}

    Is it:
    1. Factually accurate?
    2. Complete?
    3. Clearly explained?

    Respond with: VERIFIED or NEEDS_REVISION
    """

    verification_response = llm.invoke([
        {"role": "user", "content": verification_prompt}
    ])

    if "VERIFIED" in verification_response.content:
        return {
            "final_answer": last_message.content,
            "messages": [{"role": "assistant", "content": "Answer verified and finalized"}]
        }
    else:
        return {
            "messages": [{"role": "user", "content": "Please revise your answer based on verification feedback"}]
        }

# Define routing functions
def should_continue(state: AgentState):
    """Decide whether to continue or end."""
    messages = state["messages"]
    last_message = messages[-1]

    # Check iteration limit
    if state["iteration"] >= state["max_iterations"]:
        return "end"

    # Check if tool calls are present
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"

    # If no tool calls, verify answer
    return "verify"

def should_retry(state: AgentState):
    """Decide whether to retry after verification."""
    if state.get("final_answer"):
        return "end"
    else:
        return "agent"

# Build the graph
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.add_node("verify", verification_node)

# Add edges
workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        "verify": "verify",
        "end": END
    }
)

workflow.add_edge("tools", "agent")

workflow.add_conditional_edges(
    "verify",
    should_retry,
    {
        "agent": "agent",
        "end": END
    }
)

# Compile the graph
app = workflow.compile()

# Execute
initial_state = {
    "messages": [{"role": "user", "content": "Search for the population of Tokyo and calculate 10% of that number"}],
    "user_query": "Search for the population of Tokyo and calculate 10% of that number",
    "final_answer": "",
    "iteration": 0,
    "max_iterations": 5
}

final_state = app.invoke(initial_state)
print(f"Final Answer: {final_state['final_answer']}")

LangGraph Strengths

  • Explicit Control Flow: Define exactly how your agent should behave
  • Checkpointing: Save and restore state at any point
  • Human-in-the-Loop: Easy to add approval nodes
  • Self-Correction: Build retry and verification loops
  • Visualization: See your workflow as a graph

When to Use LangGraph

  • Complex workflows with multiple decision points
  • Systems requiring human approval at specific steps
  • Self-correcting agents that retry on failure
  • Advanced RAG pipelines with multiple retrieval steps
  • Financial or medical systems requiring audit trails

3. CrewAI: Role-Based Collaborative Intelligence

CrewAI is a standalone framework focused on orchestrating role-playing, autonomous AI agents that collaborate dynamically. It emphasizes a "team-based" approach where specialized agents with defined roles, goals, and backstories work together to achieve a common objective.

Core Concepts

CrewAI Architecture
  • Agents: Autonomous entities with roles, goals, and backstories
  • Tasks: Specific jobs assigned to agents
  • Crew: The team of agents working together
  • Process: Sequential, parallel, or hierarchical execution

Example 4: CrewAI Research Team

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
from crewai_tools import SerperDevTool, WebsiteSearchTool

# Initialize LLM
llm = ChatOpenAI(model="gpt-4", temperature=0.7)

# Initialize tools
search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()

# Define agents
researcher = Agent(
    role="Senior Research Analyst",
    goal="Uncover cutting-edge developments in AI and data science",
    backstory="""You're a seasoned researcher with a PhD in Computer Science.
    You're known for your ability to find and synthesize information from
    multiple sources to create comprehensive reports.""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool, web_tool],
    llm=llm
)

analyst = Agent(
    role="Data Analyst",
    goal="Analyze research findings and extract key insights",
    backstory="""You're an expert at analyzing complex information and
    identifying patterns and trends. You have a talent for turning raw
    data into actionable insights.""",
    verbose=True,
    allow_delegation=False,
    llm=llm
)

writer = Agent(
    role="Technical Writer",
    goal="Create clear, engaging reports from research findings",
    backstory="""You're a skilled technical writer who can take complex
    information and make it accessible to a broad audience. You're known
    for your clear, concise writing style.""",
    verbose=True,
    allow_delegation=False,
    llm=llm
)

# Define tasks
research_task = Task(
    description="""Research the latest developments in autonomous AI agents.
    Focus on:
    1. Recent breakthroughs and innovations
    2. Key companies and researchers in the space
    3. Practical applications and use cases
    4. Challenges and limitations

    Provide a comprehensive overview with sources.""",
    agent=researcher,
    expected_output="A detailed research report with sources and citations"
)

analysis_task = Task(
    description="""Analyze the research findings and identify:
    1. Key trends and patterns
    2. Most promising developments
    3. Potential risks and concerns
    4. Recommendations for practitioners

    Create a structured analysis with clear conclusions.""",
    agent=analyst,
    expected_output="An analytical report with insights and recommendations"
)

writing_task = Task(
    description="""Create a final report that:
    1. Synthesizes the research and analysis
    2. Presents information in a clear, engaging manner
    3. Includes executive summary
    4. Provides actionable recommendations

    The report should be suitable for technical leaders and decision-makers.""",
    agent=writer,
    expected_output="A polished final report (1500-2000 words)"
)

# Create the crew
research_crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,  # Tasks execute in order
    verbose=True
)

# Execute the crew
result = research_crew.kickoff()

print("="*60)
print("FINAL REPORT:")
print("="*60)
print(result)

Example 5: CrewAI with Hierarchical Process

from crewai import Agent, Task, Crew, Process

# Define specialist agents
coder = Agent(
    role="Senior Python Developer",
    goal="Write clean, efficient Python code",
    backstory="Expert Python developer with 10+ years experience",
    verbose=True,
    llm=llm
)

tester = Agent(
    role="QA Engineer",
    goal="Test code thoroughly and identify bugs",
    backstory="Meticulous QA engineer who never misses a bug",
    verbose=True,
    llm=llm
)

reviewer = Agent(
    role="Code Reviewer",
    goal="Review code for best practices and improvements",
    backstory="Senior architect focused on code quality",
    verbose=True,
    llm=llm
)

# Define manager agent
manager = Agent(
    role="Engineering Manager",
    goal="Coordinate the team to deliver high-quality software",
    backstory="""Experienced engineering manager who excels at delegating
    tasks and ensuring quality standards are met.""",
    verbose=True,
    allow_delegation=True,  # Can delegate to other agents
    llm=llm
)

# Define tasks
coding_task = Task(
    description="Write a Python function to calculate Fibonacci numbers efficiently",
    agent=coder
)

testing_task = Task(
    description="Write comprehensive unit tests for the Fibonacci function",
    agent=tester
)

review_task = Task(
    description="Review the code and tests, provide improvement suggestions",
    agent=reviewer
)

# Create crew with hierarchical process
dev_crew = Crew(
    agents=[manager, coder, tester, reviewer],
    tasks=[coding_task, testing_task, review_task],
    process=Process.hierarchical,  # Manager coordinates
    manager_llm=llm,
    verbose=True
)

# Execute
result = dev_crew.kickoff()
print(result)

CrewAI Strengths

  • Role-Based Design: Intuitive team metaphor
  • Autonomous Collaboration: Agents work together without explicit control flow
  • Specialization: Each agent excels at their specific role
  • Multiple Execution Modes: Sequential, parallel, or hierarchical
  • Enterprise Features: Visual builders and monitoring tools

When to Use CrewAI

  • Simulating expert teams (researchers, analysts, writers)
  • Content creation workflows
  • Automated reporting and analysis
  • Software development teams (coding, testing, reviewing)
  • Customer service with specialized agents

4. Framework Comparison: LangGraph vs CrewAI

Feature/Aspect LangGraph CrewAI
Core Paradigm Graph-based state machine Role-based collaborative team
Control Flow Explicit, deterministic edges/loops Autonomous delegation
State Management Shared state with checkpointing Task results as context
Agent Definition Nodes in graph Roles, goals, backstories
Complexity Handling Excellent for non-linear workflows Excellent for specialized teamwork
Human-in-the-Loop Strong support with breakpoints Supported via task definitions
Ease of Use Steeper learning curve More intuitive for common patterns
Performance Focus Robustness, control, persistence Speed, efficiency, collaboration
Integration Deeply integrated with LangChain Standalone, LangChain tools compatible
Best For Complex logic, loops, branching Team simulation, role specialization

Decision Framework

Choose LangGraph when:

  • You need granular control over execution flow
  • Your workflow has complex conditional logic
  • You require robust state persistence and checkpointing
  • Human approval at specific points is critical
  • You're building self-correcting systems with retry logic

Choose CrewAI when:

  • You want to simulate a team of specialized experts
  • Your tasks benefit from role-based collaboration
  • You prefer declarative agent definitions
  • You're building content creation or analysis pipelines
  • You want faster development for common patterns

5. AutoGPT and Fully Autonomous Execution

AutoGPT represents a different paradigm: fully autonomous agents that can pursue goals with minimal human intervention. While LangGraph and CrewAI provide structured frameworks, AutoGPT-style agents operate with maximum autonomy.

Example 6: Simple AutoGPT-Style Agent

from openai import OpenAI
import json
from typing import List, Dict

class AutonomousAgent:
    """AutoGPT-inspired autonomous agent."""

    def __init__(self, api_key: str, goal: str):
        self.client = OpenAI(api_key=api_key)
        self.goal = goal
        self.memory = []
        self.max_iterations = 10
        self.completed_tasks = []

    def think(self) -> Dict[str, any]:
        """Agent thinks about what to do next."""
        system_prompt = f"""You are an autonomous AI agent pursuing this goal:
{self.goal}

Your memory of what you've done so far:
{json.dumps(self.memory, indent=2)}

Tasks you've completed:
{json.dumps(self.completed_tasks, indent=2)}

Based on your goal and progress, decide what to do next.

Respond in JSON format:
{{
    "reasoning": "Your thought process",
    "action": "search_web" | "execute_code" | "write_file" | "complete",
    "action_input": {{}},
    "is_goal_achieved": true/false
}}
"""

        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": system_prompt}],
            response_format={"type": "json_object"}
        )

        decision = json.loads(response.choices[0].message.content)
        return decision

    def act(self, decision: Dict[str, any]) -> str:
        """Execute the decided action."""
        action = decision["action"]
        action_input = decision["action_input"]

        if action == "search_web":
            result = f"Web search results for: {action_input.get('query', '')}"
        elif action == "execute_code":
            # In production: use proper sandboxing
            result = "Code executed successfully"
        elif action == "write_file":
            result = f"File written: {action_input.get('filename', '')}"
        elif action == "complete":
            result = "Goal achieved!"
        else:
            result = f"Unknown action: {action}"

        return result

    def run(self):
        """Run the autonomous agent loop."""
        print(f"🎯 Goal: {self.goal}\n")

        for iteration in range(self.max_iterations):
            print(f"--- Iteration {iteration + 1} ---")

            # Think
            decision = self.think()
            print(f"πŸ’­ Reasoning: {decision['reasoning']}")
            print(f"🎬 Action: {decision['action']}")

            # Check if goal achieved
            if decision.get("is_goal_achieved") or decision["action"] == "complete":
                print("\nβœ… Goal achieved!")
                break

            # Act
            result = self.act(decision)
            print(f"πŸ“Š Result: {result}\n")

            # Update memory
            self.memory.append({
                "iteration": iteration + 1,
                "decision": decision,
                "result": result
            })

            self.completed_tasks.append(decision["action"])

        print(f"\nπŸ“ Summary:")
        print(f"Total iterations: {len(self.memory)}")
        print(f"Actions taken: {', '.join(self.completed_tasks)}")

# Usage
agent = AutonomousAgent(
    api_key="your-api-key",
    goal="Research the top 3 AI frameworks in 2025 and create a comparison report"
)
agent.run()
Warning: Fully autonomous agents can be unpredictable and expensive. Always implement cost limits, action safeguards, and human oversight for production systems.

Key Takeaways

  • Agent Stack: Modern agents are layered systems (Tools, Memory, LLM, Orchestration, UI)
  • Memory Systems: Combine short-term (conversation) and long-term (vector DB) memory for continuity
  • LangGraph: Best for complex workflows requiring explicit control and state management
  • CrewAI: Best for role-based teams and specialized collaboration patterns
  • Framework Selection: Choose based on your control requirements and team metaphor fit
  • Production Ready: Both frameworks support enterprise features like monitoring and deployment

Further Reading