Introduction: AI agents represent the next evolution of LLM applications—systems that can reason, plan, and take actions to accomplish complex tasks autonomously. Unlike simple chatbots that respond to single queries, agents maintain state, use tools, and iterate toward goals. This guide covers the architectural patterns that make agents effective: the ReAct framework for reasoning and acting, planning strategies from simple to hierarchical, memory systems for maintaining context across interactions, tool use for extending capabilities, and multi-agent systems for complex workflows. Whether you’re building a research assistant, a coding agent, or an autonomous workflow system, these patterns will help you design agents that are capable, reliable, and controllable.

ReAct Framework

from dataclasses import dataclass, field
from typing import Any, Optional, Callable
from abc import ABC, abstractmethod
from enum import Enum

class ActionType(Enum):
    """Types of agent actions."""
    
    TOOL_CALL = "tool_call"
    FINAL_ANSWER = "final_answer"
    THINK = "think"

@dataclass
class Thought:
    """Agent's reasoning step."""
    
    content: str
    step: int

@dataclass
class Action:
    """Agent action."""
    
    type: ActionType
    tool_name: str = None
    tool_input: dict = None
    answer: str = None

@dataclass
class Observation:
    """Result of an action."""
    
    content: str
    success: bool = True

@dataclass
class AgentStep:
    """A single step in agent execution."""
    
    thought: Thought
    action: Action
    observation: Observation = None

class ReActAgent:
    """ReAct (Reasoning + Acting) agent."""
    
    def __init__(
        self,
        llm_client: Any,
        tools: dict[str, Callable],
        max_steps: int = 10
    ):
        self.llm = llm_client
        self.tools = tools
        self.max_steps = max_steps
        self.history: list[AgentStep] = []
    
    async def run(self, task: str) -> str:
        """Execute task using ReAct loop."""
        
        self.history = []
        
        for step in range(self.max_steps):
            # Generate thought and action
            thought, action = await self._think_and_act(task, step)
            
            # Check for final answer
            if action.type == ActionType.FINAL_ANSWER:
                self.history.append(AgentStep(
                    thought=thought,
                    action=action
                ))
                return action.answer
            
            # Execute action
            observation = await self._execute_action(action)
            
            # Record step
            self.history.append(AgentStep(
                thought=thought,
                action=action,
                observation=observation
            ))
            
            # Check for errors
            if not observation.success:
                # Could retry or adjust strategy
                pass
        
        return "Max steps reached without finding answer."
    
    async def _think_and_act(
        self,
        task: str,
        step: int
    ) -> tuple[Thought, Action]:
        """Generate thought and decide action."""
        
        # Build prompt with history
        prompt = self._build_prompt(task)
        
        response = await self.llm.complete(prompt)
        
        # Parse response
        thought, action = self._parse_response(response.content, step)
        
        return thought, action
    
    def _build_prompt(self, task: str) -> str:
        """Build ReAct prompt."""
        
        tools_desc = "\n".join(
            f"- {name}: {func.__doc__ or 'No description'}"
            for name, func in self.tools.items()
        )
        
        history_text = ""
        for step in self.history:
            history_text += f"\nThought: {step.thought.content}"
            if step.action.type == ActionType.TOOL_CALL:
                history_text += f"\nAction: {step.action.tool_name}({step.action.tool_input})"
                history_text += f"\nObservation: {step.observation.content}"
        
        return f"""You are an AI assistant that solves tasks by thinking step-by-step and using tools.

Available tools:
{tools_desc}

Task: {task}

{history_text}

Think about what to do next, then either use a tool or provide the final answer.

Format your response as:
Thought: [your reasoning]
Action: [tool_name(input)] OR Final Answer: [your answer]"""
    
    def _parse_response(self, response: str, step: int) -> tuple[Thought, Action]:
        """Parse LLM response into thought and action."""
        
        lines = response.strip().split('\n')
        
        thought_content = ""
        action = None
        
        for line in lines:
            if line.startswith("Thought:"):
                thought_content = line[8:].strip()
            elif line.startswith("Action:"):
                action_text = line[7:].strip()
                action = self._parse_action(action_text)
            elif line.startswith("Final Answer:"):
                answer = line[13:].strip()
                action = Action(type=ActionType.FINAL_ANSWER, answer=answer)
        
        thought = Thought(content=thought_content, step=step)
        
        if action is None:
            action = Action(type=ActionType.THINK)
        
        return thought, action
    
    def _parse_action(self, action_text: str) -> Action:
        """Parse action text into Action object."""
        
        import re
        
        # Match tool_name(input)
        match = re.match(r'(\w+)\((.*)\)', action_text)
        
        if match:
            tool_name = match.group(1)
            tool_input = match.group(2)
            
            # Try to parse as JSON or dict
            try:
                import json
                input_dict = json.loads(tool_input)
            except json.JSONDecodeError:
                input_dict = {"input": tool_input}
            
            return Action(
                type=ActionType.TOOL_CALL,
                tool_name=tool_name,
                tool_input=input_dict
            )
        
        return Action(type=ActionType.THINK)
    
    async def _execute_action(self, action: Action) -> Observation:
        """Execute tool action."""
        
        if action.type != ActionType.TOOL_CALL:
            return Observation(content="No action to execute", success=True)
        
        if action.tool_name not in self.tools:
            return Observation(
                content=f"Unknown tool: {action.tool_name}",
                success=False
            )
        
        try:
            tool = self.tools[action.tool_name]
            result = tool(**action.tool_input)
            
            # Handle async tools
            if hasattr(result, '__await__'):
                result = await result
            
            return Observation(content=str(result), success=True)
        except Exception as e:
            return Observation(content=f"Error: {str(e)}", success=False)

Planning Strategies

from dataclasses import dataclass, field
from typing import Any, Optional
from abc import ABC, abstractmethod
from enum import Enum

class PlanStatus(Enum):
    """Status of a plan step."""
    
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class PlanStep:
    """A step in a plan."""
    
    id: str
    description: str
    status: PlanStatus = PlanStatus.PENDING
    result: str = None
    dependencies: list[str] = field(default_factory=list)

@dataclass
class Plan:
    """A plan for completing a task."""
    
    goal: str
    steps: list[PlanStep]
    current_step: int = 0

class Planner(ABC):
    """Abstract planner."""
    
    @abstractmethod
    async def create_plan(self, task: str) -> Plan:
        """Create a plan for the task."""
        pass
    
    @abstractmethod
    async def replan(self, plan: Plan, feedback: str) -> Plan:
        """Revise plan based on feedback."""
        pass

class SimplePlanner(Planner):
    """Simple sequential planner."""
    
    def __init__(self, llm_client: Any):
        self.llm = llm_client
    
    async def create_plan(self, task: str) -> Plan:
        """Create sequential plan."""
        
        prompt = f"""Create a step-by-step plan to accomplish this task.
Each step should be a single, actionable item.

Task: {task}

Plan (numbered list):"""
        
        response = await self.llm.complete(prompt)
        
        steps = []
        for i, line in enumerate(response.content.split('\n')):
            line = line.strip().lstrip('0123456789.-) ')
            if line:
                steps.append(PlanStep(
                    id=f"step_{i}",
                    description=line
                ))
        
        return Plan(goal=task, steps=steps)
    
    async def replan(self, plan: Plan, feedback: str) -> Plan:
        """Revise plan based on feedback."""
        
        completed = [s for s in plan.steps if s.status == PlanStatus.COMPLETED]
        remaining = [s for s in plan.steps if s.status == PlanStatus.PENDING]
        
        prompt = f"""Revise this plan based on the feedback.

Original goal: {plan.goal}

Completed steps:
{chr(10).join(s.description for s in completed)}

Remaining steps:
{chr(10).join(s.description for s in remaining)}

Feedback: {feedback}

Revised remaining steps (numbered list):"""
        
        response = await self.llm.complete(prompt)
        
        new_steps = list(completed)
        for i, line in enumerate(response.content.split('\n')):
            line = line.strip().lstrip('0123456789.-) ')
            if line:
                new_steps.append(PlanStep(
                    id=f"step_{len(completed) + i}",
                    description=line
                ))
        
        return Plan(goal=plan.goal, steps=new_steps, current_step=len(completed))

class HierarchicalPlanner(Planner):
    """Hierarchical task decomposition planner."""
    
    def __init__(self, llm_client: Any, max_depth: int = 3):
        self.llm = llm_client
        self.max_depth = max_depth
    
    async def create_plan(self, task: str) -> Plan:
        """Create hierarchical plan."""
        
        # First level decomposition
        high_level = await self._decompose(task, depth=0)
        
        # Recursively decompose complex steps
        all_steps = []
        for step in high_level:
            if await self._is_complex(step.description):
                substeps = await self._decompose(step.description, depth=1)
                all_steps.extend(substeps)
            else:
                all_steps.append(step)
        
        return Plan(goal=task, steps=all_steps)
    
    async def _decompose(self, task: str, depth: int) -> list[PlanStep]:
        """Decompose task into subtasks."""
        
        if depth >= self.max_depth:
            return [PlanStep(id=f"step_0", description=task)]
        
        prompt = f"""Break down this task into 3-5 subtasks.
Each subtask should be specific and actionable.

Task: {task}

Subtasks (numbered list):"""
        
        response = await self.llm.complete(prompt)
        
        steps = []
        for i, line in enumerate(response.content.split('\n')):
            line = line.strip().lstrip('0123456789.-) ')
            if line:
                steps.append(PlanStep(
                    id=f"step_{depth}_{i}",
                    description=line
                ))
        
        return steps
    
    async def _is_complex(self, task: str) -> bool:
        """Check if task needs further decomposition."""
        
        prompt = f"""Is this task complex enough to need breaking down into subtasks?
Answer YES or NO.

Task: {task}

Answer:"""
        
        response = await self.llm.complete(prompt)
        return "yes" in response.content.lower()
    
    async def replan(self, plan: Plan, feedback: str) -> Plan:
        """Revise hierarchical plan."""
        
        # Similar to simple planner but maintains hierarchy
        return await SimplePlanner(self.llm).replan(plan, feedback)

class DAGPlanner(Planner):
    """Plan with dependencies (DAG)."""
    
    def __init__(self, llm_client: Any):
        self.llm = llm_client
    
    async def create_plan(self, task: str) -> Plan:
        """Create plan with dependencies."""
        
        prompt = f"""Create a plan with dependencies for this task.
Some steps may depend on others being completed first.

Task: {task}

For each step, specify:
- Step description
- Dependencies (which steps must be done first)

Format:
1. [description] | depends on: none
2. [description] | depends on: 1
3. [description] | depends on: 1, 2

Plan:"""
        
        response = await self.llm.complete(prompt)
        
        steps = []
        for line in response.content.split('\n'):
            if '|' in line:
                parts = line.split('|')
                desc = parts[0].strip().lstrip('0123456789.-) ')
                
                deps = []
                if len(parts) > 1 and 'depends on:' in parts[1]:
                    dep_text = parts[1].split('depends on:')[1].strip()
                    if dep_text.lower() != 'none':
                        deps = [f"step_{int(d.strip()) - 1}" for d in dep_text.split(',') if d.strip().isdigit()]
                
                steps.append(PlanStep(
                    id=f"step_{len(steps)}",
                    description=desc,
                    dependencies=deps
                ))
        
        return Plan(goal=task, steps=steps)
    
    def get_executable_steps(self, plan: Plan) -> list[PlanStep]:
        """Get steps that can be executed (dependencies met)."""
        
        completed_ids = {s.id for s in plan.steps if s.status == PlanStatus.COMPLETED}
        
        executable = []
        for step in plan.steps:
            if step.status == PlanStatus.PENDING:
                if all(dep in completed_ids for dep in step.dependencies):
                    executable.append(step)
        
        return executable
    
    async def replan(self, plan: Plan, feedback: str) -> Plan:
        """Revise DAG plan."""
        return await self.create_plan(f"{plan.goal}\n\nPrevious attempt feedback: {feedback}")

Agent Memory Systems

from dataclasses import dataclass, field
from typing import Any, Optional
from abc import ABC, abstractmethod
from datetime import datetime
import json

@dataclass
class MemoryEntry:
    """A memory entry."""
    
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    importance: float = 0.5
    metadata: dict = field(default_factory=dict)

class AgentMemory(ABC):
    """Abstract agent memory."""
    
    @abstractmethod
    def add(self, content: str, **kwargs) -> None:
        """Add to memory."""
        pass
    
    @abstractmethod
    def retrieve(self, query: str, k: int = 5) -> list[MemoryEntry]:
        """Retrieve relevant memories."""
        pass
    
    @abstractmethod
    def clear(self) -> None:
        """Clear memory."""
        pass

class WorkingMemory(AgentMemory):
    """Short-term working memory."""
    
    def __init__(self, max_entries: int = 10):
        self.max_entries = max_entries
        self.entries: list[MemoryEntry] = []
    
    def add(self, content: str, **kwargs) -> None:
        """Add to working memory."""
        
        entry = MemoryEntry(
            content=content,
            importance=kwargs.get('importance', 0.5),
            metadata=kwargs.get('metadata', {})
        )
        
        self.entries.append(entry)
        
        # Trim if over capacity
        if len(self.entries) > self.max_entries:
            # Remove least important
            self.entries.sort(key=lambda e: e.importance, reverse=True)
            self.entries = self.entries[:self.max_entries]
    
    def retrieve(self, query: str, k: int = 5) -> list[MemoryEntry]:
        """Get recent memories."""
        
        # Return most recent
        return sorted(self.entries, key=lambda e: e.timestamp, reverse=True)[:k]
    
    def clear(self) -> None:
        self.entries = []

class EpisodicMemory(AgentMemory):
    """Memory of past episodes/experiences."""
    
    def __init__(self, embedding_model: Any, vector_store: Any):
        self.embedder = embedding_model
        self.store = vector_store
        self.entries: list[MemoryEntry] = []
    
    def add(self, content: str, **kwargs) -> None:
        """Add episode to memory."""
        
        entry = MemoryEntry(
            content=content,
            importance=kwargs.get('importance', 0.5),
            metadata=kwargs.get('metadata', {})
        )
        
        # Store with embedding
        embedding = self.embedder.embed(content)
        self.store.add(
            id=str(len(self.entries)),
            embedding=embedding,
            metadata={
                "content": content,
                "timestamp": entry.timestamp.isoformat(),
                "importance": entry.importance
            }
        )
        
        self.entries.append(entry)
    
    def retrieve(self, query: str, k: int = 5) -> list[MemoryEntry]:
        """Retrieve relevant episodes."""
        
        query_embedding = self.embedder.embed(query)
        results = self.store.search(query_embedding, k=k)
        
        entries = []
        for r in results:
            entries.append(MemoryEntry(
                content=r.metadata["content"],
                timestamp=datetime.fromisoformat(r.metadata["timestamp"]),
                importance=r.metadata["importance"]
            ))
        
        return entries
    
    def clear(self) -> None:
        self.entries = []
        self.store.clear()

class SemanticMemory(AgentMemory):
    """Long-term knowledge memory."""
    
    def __init__(self, embedding_model: Any):
        self.embedder = embedding_model
        self.knowledge: dict[str, MemoryEntry] = {}
        self.embeddings: dict[str, Any] = {}
    
    def add(self, content: str, key: str = None, **kwargs) -> None:
        """Add knowledge to memory."""
        
        if key is None:
            key = content[:50]
        
        entry = MemoryEntry(
            content=content,
            importance=kwargs.get('importance', 0.5),
            metadata=kwargs.get('metadata', {})
        )
        
        self.knowledge[key] = entry
        self.embeddings[key] = self.embedder.embed(content)
    
    def retrieve(self, query: str, k: int = 5) -> list[MemoryEntry]:
        """Retrieve relevant knowledge."""
        
        import numpy as np
        
        query_embedding = self.embedder.embed(query)
        
        # Calculate similarities
        similarities = []
        for key, emb in self.embeddings.items():
            sim = np.dot(query_embedding, emb)
            similarities.append((key, sim))
        
        # Sort by similarity
        similarities.sort(key=lambda x: x[1], reverse=True)
        
        return [self.knowledge[key] for key, _ in similarities[:k]]
    
    def clear(self) -> None:
        self.knowledge = {}
        self.embeddings = {}

class ReflectionMemory(AgentMemory):
    """Memory with reflection and synthesis."""
    
    def __init__(self, llm_client: Any, base_memory: AgentMemory):
        self.llm = llm_client
        self.base = base_memory
        self.reflections: list[MemoryEntry] = []
    
    def add(self, content: str, **kwargs) -> None:
        """Add and potentially reflect."""
        
        self.base.add(content, **kwargs)
        
        # Periodically reflect
        if len(self.base.entries) % 10 == 0:
            asyncio.create_task(self._reflect())
    
    async def _reflect(self) -> None:
        """Generate reflection on recent memories."""
        
        recent = self.base.retrieve("", k=10)
        
        prompt = f"""Reflect on these recent experiences and extract key insights.

Experiences:
{chr(10).join(e.content for e in recent)}

Key insights (2-3 sentences):"""
        
        response = await self.llm.complete(prompt)
        
        self.reflections.append(MemoryEntry(
            content=response.content,
            importance=0.8,
            metadata={"type": "reflection"}
        ))
    
    def retrieve(self, query: str, k: int = 5) -> list[MemoryEntry]:
        """Retrieve memories and relevant reflections."""
        
        base_results = self.base.retrieve(query, k=k-1)
        
        # Add most relevant reflection
        if self.reflections:
            base_results.append(self.reflections[-1])
        
        return base_results
    
    def clear(self) -> None:
        self.base.clear()
        self.reflections = []

class CompositeMemory(AgentMemory):
    """Combine multiple memory systems."""
    
    def __init__(
        self,
        working: WorkingMemory,
        episodic: EpisodicMemory,
        semantic: SemanticMemory
    ):
        self.working = working
        self.episodic = episodic
        self.semantic = semantic
    
    def add(self, content: str, memory_type: str = "working", **kwargs) -> None:
        """Add to appropriate memory."""
        
        if memory_type == "working":
            self.working.add(content, **kwargs)
        elif memory_type == "episodic":
            self.episodic.add(content, **kwargs)
        elif memory_type == "semantic":
            self.semantic.add(content, **kwargs)
        else:
            # Add to all
            self.working.add(content, **kwargs)
            self.episodic.add(content, **kwargs)
    
    def retrieve(self, query: str, k: int = 5) -> list[MemoryEntry]:
        """Retrieve from all memories."""
        
        results = []
        
        # Get from each memory type
        results.extend(self.working.retrieve(query, k=2))
        results.extend(self.episodic.retrieve(query, k=2))
        results.extend(self.semantic.retrieve(query, k=2))
        
        # Sort by importance and recency
        results.sort(key=lambda e: (e.importance, e.timestamp), reverse=True)
        
        return results[:k]
    
    def clear(self) -> None:
        self.working.clear()
        self.episodic.clear()
        self.semantic.clear()

Multi-Agent Systems

from dataclasses import dataclass, field
from typing import Any, Optional, Callable
from abc import ABC, abstractmethod
from enum import Enum
import asyncio

@dataclass
class Message:
    """Message between agents."""
    
    sender: str
    recipient: str
    content: str
    message_type: str = "text"
    metadata: dict = field(default_factory=dict)

class Agent(ABC):
    """Abstract agent in multi-agent system."""
    
    def __init__(self, name: str, llm_client: Any):
        self.name = name
        self.llm = llm_client
        self.inbox: asyncio.Queue = asyncio.Queue()
    
    @abstractmethod
    async def process(self, message: Message) -> Optional[Message]:
        """Process incoming message."""
        pass
    
    async def send(self, recipient: str, content: str, **kwargs) -> Message:
        """Create message to send."""
        
        return Message(
            sender=self.name,
            recipient=recipient,
            content=content,
            **kwargs
        )

class SpecialistAgent(Agent):
    """Agent with specific expertise."""
    
    def __init__(
        self,
        name: str,
        llm_client: Any,
        expertise: str,
        system_prompt: str = None
    ):
        super().__init__(name, llm_client)
        self.expertise = expertise
        self.system_prompt = system_prompt or f"You are an expert in {expertise}."
    
    async def process(self, message: Message) -> Optional[Message]:
        """Process with expertise."""
        
        prompt = f"""{self.system_prompt}

Message from {message.sender}:
{message.content}

Your response:"""
        
        response = await self.llm.complete(prompt)
        
        return await self.send(
            recipient=message.sender,
            content=response.content
        )

class OrchestratorAgent(Agent):
    """Agent that coordinates other agents."""
    
    def __init__(
        self,
        name: str,
        llm_client: Any,
        agents: dict[str, Agent]
    ):
        super().__init__(name, llm_client)
        self.agents = agents
    
    async def process(self, message: Message) -> Optional[Message]:
        """Orchestrate task across agents."""
        
        # Decide which agent(s) to involve
        agent_names = list(self.agents.keys())
        
        prompt = f"""You are coordinating a team of specialists: {', '.join(agent_names)}

Task: {message.content}

Which specialist(s) should handle this? Respond with agent names separated by commas:"""
        
        response = await self.llm.complete(prompt)
        
        selected = [a.strip() for a in response.content.split(',')]
        
        # Delegate to selected agents
        results = []
        for agent_name in selected:
            if agent_name in self.agents:
                agent = self.agents[agent_name]
                sub_message = Message(
                    sender=self.name,
                    recipient=agent_name,
                    content=message.content
                )
                result = await agent.process(sub_message)
                if result:
                    results.append(f"{agent_name}: {result.content}")
        
        # Synthesize results
        synthesis_prompt = f"""Synthesize these specialist responses into a final answer.

Task: {message.content}

Responses:
{chr(10).join(results)}

Final answer:"""
        
        final = await self.llm.complete(synthesis_prompt)
        
        return await self.send(
            recipient=message.sender,
            content=final.content
        )

class DebateSystem:
    """Multi-agent debate for better reasoning."""
    
    def __init__(
        self,
        llm_client: Any,
        num_agents: int = 3,
        num_rounds: int = 2
    ):
        self.llm = llm_client
        self.num_agents = num_agents
        self.num_rounds = num_rounds
    
    async def debate(self, question: str) -> str:
        """Run debate and synthesize answer."""
        
        # Initial responses
        responses = []
        for i in range(self.num_agents):
            prompt = f"""You are Agent {i+1}. Answer this question with your best reasoning.

Question: {question}

Your answer:"""
            
            response = await self.llm.complete(prompt)
            responses.append(response.content)
        
        # Debate rounds
        for round_num in range(self.num_rounds):
            new_responses = []
            
            for i in range(self.num_agents):
                other_responses = [r for j, r in enumerate(responses) if j != i]
                
                prompt = f"""You are Agent {i+1}. Consider other agents' responses and refine your answer.

Question: {question}

Your previous answer: {responses[i]}

Other agents' answers:
{chr(10).join(f"Agent {j+1}: {r}" for j, r in enumerate(other_responses))}

Your refined answer (consider their points but maintain your reasoning):"""
                
                response = await self.llm.complete(prompt)
                new_responses.append(response.content)
            
            responses = new_responses
        
        # Final synthesis
        synthesis_prompt = f"""Synthesize these debated responses into a final answer.

Question: {question}

Final responses after debate:
{chr(10).join(f"Agent {i+1}: {r}" for i, r in enumerate(responses))}

Best synthesized answer:"""
        
        final = await self.llm.complete(synthesis_prompt)
        return final.content

class CrewSystem:
    """Crew of agents working together."""
    
    def __init__(self):
        self.agents: dict[str, Agent] = {}
        self.message_bus: asyncio.Queue = asyncio.Queue()
    
    def add_agent(self, agent: Agent):
        """Add agent to crew."""
        self.agents[agent.name] = agent
    
    async def run_task(self, task: str, coordinator: str) -> str:
        """Run task with coordinator."""
        
        if coordinator not in self.agents:
            raise ValueError(f"Unknown coordinator: {coordinator}")
        
        # Send task to coordinator
        message = Message(
            sender="user",
            recipient=coordinator,
            content=task
        )
        
        result = await self.agents[coordinator].process(message)
        
        return result.content if result else "No result"
    
    async def run_pipeline(
        self,
        task: str,
        pipeline: list[str]
    ) -> str:
        """Run task through agent pipeline."""
        
        current_content = task
        
        for agent_name in pipeline:
            if agent_name not in self.agents:
                continue
            
            message = Message(
                sender="pipeline",
                recipient=agent_name,
                content=current_content
            )
            
            result = await self.agents[agent_name].process(message)
            
            if result:
                current_content = result.content
        
        return current_content

Production Agent Service

from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Optional, Any
import asyncio
import uuid

app = FastAPI()

class TaskRequest(BaseModel):
    task: str
    agent_type: str = "react"
    max_steps: int = 10

class TaskResponse(BaseModel):
    task_id: str
    status: str
    result: Optional[str] = None
    steps: list[dict] = []

# Task storage
tasks: dict[str, dict] = {}

# Mock agent for demo
class MockAgent:
    async def run(self, task: str) -> str:
        await asyncio.sleep(1)
        return f"Completed task: {task}"

agent = MockAgent()

async def run_agent_task(task_id: str, task: str, max_steps: int):
    """Run agent task in background."""
    
    tasks[task_id]["status"] = "running"
    
    try:
        result = await agent.run(task)
        tasks[task_id]["status"] = "completed"
        tasks[task_id]["result"] = result
    except Exception as e:
        tasks[task_id]["status"] = "failed"
        tasks[task_id]["error"] = str(e)

@app.post("/v1/tasks")
async def create_task(
    request: TaskRequest,
    background_tasks: BackgroundTasks
) -> TaskResponse:
    """Create agent task."""
    
    task_id = str(uuid.uuid4())
    
    tasks[task_id] = {
        "status": "pending",
        "task": request.task,
        "result": None,
        "steps": []
    }
    
    background_tasks.add_task(
        run_agent_task,
        task_id,
        request.task,
        request.max_steps
    )
    
    return TaskResponse(
        task_id=task_id,
        status="pending"
    )

@app.get("/v1/tasks/{task_id}")
async def get_task(task_id: str) -> TaskResponse:
    """Get task status."""
    
    if task_id not in tasks:
        raise HTTPException(status_code=404, detail="Task not found")
    
    task = tasks[task_id]
    
    return TaskResponse(
        task_id=task_id,
        status=task["status"],
        result=task.get("result"),
        steps=task.get("steps", [])
    )

@app.delete("/v1/tasks/{task_id}")
async def cancel_task(task_id: str) -> dict:
    """Cancel task."""
    
    if task_id not in tasks:
        raise HTTPException(status_code=404, detail="Task not found")
    
    tasks[task_id]["status"] = "cancelled"
    
    return {"status": "cancelled"}

@app.get("/v1/tasks")
async def list_tasks() -> list[TaskResponse]:
    """List all tasks."""
    
    return [
        TaskResponse(
            task_id=tid,
            status=task["status"],
            result=task.get("result")
        )
        for tid, task in tasks.items()
    ]

@app.get("/health")
async def health():
    return {"status": "healthy"}

References

• ReAct Paper: https://arxiv.org/abs/2210.03629
• LangChain Agents: https://python.langchain.com/docs/modules/agents/
• AutoGPT: https://github.com/Significant-Gravitas/AutoGPT
• CrewAI: https://github.com/joaomdmoura/crewAI

Conclusion

Agent architectures transform LLMs from reactive responders into proactive problem solvers. The ReAct framework provides the foundation—interleaving reasoning and action creates more reliable behavior than either alone. Planning is essential for complex tasks; start with simple sequential plans and add hierarchy or dependencies as needed. Memory systems determine what agents can remember and learn; working memory for immediate context, episodic memory for past experiences, and semantic memory for accumulated knowledge. Tool use extends agent capabilities beyond text generation; design tools with clear interfaces and good error messages. Multi-agent systems enable specialization and collaboration; orchestrators coordinate specialists, debates improve reasoning through diverse perspectives, and pipelines enable sequential processing. In production, implement proper task management with status tracking, cancellation, and timeout handling. Monitor agent behavior closely—autonomous systems can fail in unexpected ways. Start with constrained agents that have limited tools and clear boundaries, then expand capabilities as you understand their behavior. The key insight is that agents are systems, not just prompts—architecture decisions about planning, memory, and coordination matter as much as the underlying LLM.