Agentic AI: Building Autonomous AI Systems That Act and Reason

from typing import List, Dict, Any
from dataclasses import dataclass
from enum import Enum

class AgentState(Enum):
    IDLE = "idle"
    THINKING = "thinking"
    ACTING = "acting"
    REFLECTING = "reflecting"

@dataclass
class Agent:
    name: str
    llm: Any  # LLM client
    tools: List[Any]
    max_iterations: int = 10
    
    def __post_init__(self):
        self.state = AgentState.IDLE
        self.memory: List[Dict] = []
        self.plan: List[str] = []
    
    async def run(self, task: str) -> str:
        """Execute a task autonomously."""
        self.plan = await self.plan_task(task)
        
        for iteration in range(self.max_iterations):
            # Think
            self.state = AgentState.THINKING
            thought = await self.think(task)
            self.memory.append({"role": "assistant", "content": thought})
            
            # Act
            self.state = AgentState.ACTING
            result = await self.act(thought)
            self.memory.append({"role": "user", "content": f"Result: {result}"})
            
            # Reflect
            self.state = AgentState.REFLECTING
            if await self.is_complete(result):
                return result
            
            # Adjust plan if needed
            self.plan = await self.adjust_plan(self.plan, result)
        
        return "Task incomplete after max iterations"
    
    async def plan_task(self, task: str) -> List[str]:
        """Create a plan to accomplish the task."""
        prompt = f"""Break down this task into steps:
        Task: {task}
        
        Provide a numbered list of steps."""
        
        response = await self.llm.complete(prompt)
        return self.parse_steps(response)
    
    async def think(self, task: str) -> str:
        """Given current context, decide what to do next."""
        prompt = f"""You are working on: {task}
        
        Current plan: {self.plan}
        
        Recent memory:
        {self.format_memory()}
        
        What should you do next? Be specific."""
        
        return await self.llm.complete(prompt)
    
    async def act(self, thought: str) -> str:
        """Execute the decided action."""
        # Determine if we need a tool
        if tool_name := self.extract_tool(thought):
            tool = self.get_tool(tool_name)
            args = self.extract_args(thought, tool)
            return await tool.execute(args)
        
        # Otherwise, just reason
        return thought
    
    async def reflect(self, result: str) -> bool:
        """Evaluate if task is complete."""
        prompt = f"""Given the result: {result}
        
        Is the original task complete? Answer yes or no and explain."""
        
        response = await self.llm.complete(prompt)
        return "yes" in response.lower()

class ReActAgent:
    """Agent that interleaves reasoning and acting."""
    
    def __init__(self, llm, tools, max_steps=10):
        self.llm = llm
        self.tools = tools
        self.max_steps = max_steps
    
    async def run(self, task: str) -> str:
        observations = []
        thoughts = []
        
        for step in range(self.max_steps):
            # Think
            thought = await self.think(task, thoughts, observations)
            thoughts.append(thought)
            
            # Decide action
            action = await self.decide_action(thought)
            
            if action["type"] == "finish":
                return action["result"]
            
            # Act
            if action["type"] == "tool":
                obs = await self.execute_tool(
                    action["tool"], 
                    action["args"]
                )
                observations.append(f"Tool: {action['tool']}, Result: {obs}")
            
            elif action["type"] == "reason":
                observations.append(f"Reasoning: {action['result']}")
        
        return "Task incomplete"
    
    async def think(self, task, thoughts, observations):
        prompt = f"""Task: {task}

Previous thoughts:
{self.format_list(thoughts)}

Observations:
{self.format_list(observations)}

Think about what to do next. Consider:
1. What have I learned from observations?
2. What information do I still need?
3. Should I use a tool or reason more?"""
        
        return await self.llm.complete(prompt)
    
    async def decide_action(self, thought: str) -> Dict:
        prompt = f"""Based on this thought: {thought}

Decide your next action. Choose from:
- tool: Use a tool (provide tool name and args)
- reason: More reasoning needed (provide your reasoning)
- finish: Task complete (provide final result)

Respond in JSON format:
{{"type": "tool|reason|finish", "tool": "...", "args": {{...}}, "result": "..."}}"""
        
        response = await self.llm.complete(prompt)
        return self.parse_json(response)

class ToolUseAgent:
    """Agent that can use tools to accomplish tasks."""
    
    def __init__(self, llm):
        self.llm = llm
        self.tools: Dict[str, Any] = {}
        self.tool_descriptions = []
    
    def register_tool(self, tool):
        """Register a tool the agent can use."""
        self.tools[tool.name] = tool
        self.tool_descriptions.append(tool.get_description())
    
    async def select_and_use_tool(self, task: str) -> str:
        """Select the best tool and use it."""
        # Get tool recommendations
        prompt = f"""Task: {task}

Available tools:
{self.format_tools()}

Which tool would be most helpful? Respond with the tool name."""
        
        tool_name = await self.llm.complete(prompt)
        
        # Get arguments
        tool = self.tools.get(tool_name.strip())
        if not tool:
            return f"Unknown tool: {tool_name}"
        
        args = await self.get_tool_arguments(tool, task)
        
        # Execute
        return await tool.execute(**args)
    
    async def get_tool_arguments(self, tool, task: str) -> Dict:
        prompt = f"""Task: {task}

Tool: {tool.name}
Description: {tool.description}
Parameters: {tool.parameters}

What are the values for these parameters?"""
        
        response = await self.llm.complete(prompt)
        return self.parse_arguments(response)

class AgentMemory:
    def __init__(self):
        self.working = []  # Current context
        self.episodic = []  # Past interactions
        self.semantic = {}   # Learned knowledge
        self.max_working = 10
    
    def add_working(self, item: str):
        """Add to working memory."""
        self.working.append(item)
        if len(self.working) > self.max_working:
            self.working.pop(0)
    
    def add_episodic(self, episode: Dict):
        """Store an interaction episode."""
        self.episodic.append({
            "task": episode["task"],
            "actions": episode["actions"],
            "result": episode["result"],
            "learnings": episode.get("learnings", [])
        })
    
    def retrieve_relevant(self, query: str, k: int = 5) -> List[str]:
        """Retrieve relevant memories."""
        # Simple retrieval - in production use embeddings
        scores = []
        for i, episode in enumerate(self.episodic):
            score = self.calculate_relevance(query, episode)
            scores.append((score, episode))
        
        scores.sort(reverse=True)
        return [ep for _, ep in scores[:k]]
    
    def store_knowledge(self, key: str, value: Any):
        """Store semantic knowledge."""
        self.semantic[key] = value
    
    def get_knowledge(self, key: str) -> Any:
        """Retrieve semantic knowledge."""
        return self.semantic.get(key)

async def chain_of_thought(agent, task):
    """Step-by-step reasoning."""
    prompt = f"""Task: {task}

Think step by step to solve this problem. Show your reasoning at each step."""

    return await agent.llm.complete(prompt)

async def tree_of_thought(agent, task, breadth=3, depth=3):
    """Explore multiple reasoning paths."""
    
    async def explore(state, depth):
        if depth == 0:
            return evaluate(state)
        
        thoughts = await generate_thoughts(state, breadth)
        
        results = []
        for thought in thoughts:
            new_state = update_state(state, thought)
            result = await explore(new_state, depth - 1)
            results.append((thought, result))
        
        return max(results, key=lambda x: x[1])
    
    return await explore({"task": task}, depth)

class RePlanner:
    """Replans periodically based on progress."""
    
    async def run(self, task: str, replan_every=3):
        plan = await self.create_initial_plan(task)
        
        for i in range(self.max_steps):
            # Execute current plan
            result = await self.execute_plan(plan)
            
            if await self.is_complete(result):
                return result
            
            # Replan periodically
            if (i + 1) % replan_every == 0:
                plan = await self.replan(task, plan, result)
        
        return "Incomplete"
    
    async def create_initial_plan(self, task):
        prompt = f"""Create a plan to: {task}
        
Format as numbered steps."""
        return await self.llm.complete(prompt)
    
    async def replan(self, task, old_plan, result):
        prompt = f"""Task: {task}

Current plan: {old_plan}

Progress: {result}

Revise the plan based on progress. What steps remain?"""
        return await self.llm.complete(prompt)

async def evaluate_agent(agent, test_cases):
    """Evaluate agent on test cases."""
    results = []
    
    for case in test_cases:
        task = case["task"]
        expected = case["expected"]
        
        result = await agent.run(task)
        
        results.append({
            "task": task,
            "result": result,
            "expected": expected,
            "success": evaluate(result, expected),
            "steps": agent.iterations,
            "tools_used": agent.tools_used
        })
    
    return summarize(results)

# Start with simple agents, add complexity as needed

# 1. Simple reflex agent
# 2. Add memory
# 3. Add planning
# 4. Add reflection
# 5. Add learning

async def safe_agent_execute(agent, task):
    try:
        return await agent.run(task)
    except ToolNotFoundError:
        return "I couldn't find the right tool for that"
    except ToolExecutionError as e:
        return f"Tool failed: {e}. Let me try a different approach."
    except MaxIterationsError:
        return "I'm taking too long. Can you break this into smaller steps?"

class GuardedAgent:
    """Agent with safety guardrails."""
    
    def __init__(self, agent, policy):
        self.agent = agent
        self.policy = policy
    
    async def run(self, task):
        # Check if task is allowed
        if not self.policy.is_allowed(task):
            return "I can't help with that request"
        
        # Run agent
        result = await self.agent.run(task)
        
        # Check output
        if not self.policy.is_safe(result):
            return "I generated content that needs review"
        
        return result