What Are AI Agents?

from openai import OpenAI

class AgentBrain:
    def __init__(self, model="gpt-4"):
        self.client = OpenAI()
        self.model = model
        self.memory = []  # Conversation history
    
    def think(self, observation, goal, available_tools):
        """Reason about what to do next"""
        prompt = f"""
Goal: {goal}

Current Observation: {observation}

Available Tools: {', '.join(available_tools)}

Memory of past actions: {self.memory[-5:] if self.memory else 'None'}

What should I do next? Think step-by-step:
1. What have I accomplished so far?
2. What do I need to do to reach my goal?
3. Which tool should I use next?
4. What are the expected outcomes?

Your response:
"""
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": "You are an autonomous agent. Reason carefully and plan actions."},
                {"role": "user", "content": prompt}
            ]
        )
        
        decision = response.choices[0].message.content
        self.memory.append({
            "observation": observation,
            "decision": decision
        })
        
        return decision

# Usage
brain = AgentBrain()
observation = "User asked to book a flight to Paris"
goal = "Book a round-trip flight for the user"
tools = ["search_flights", "book_flight", "send_email"]

next_action = brain.think(observation, goal, tools)
print(next_action)

class AgentTools:
    """Define all tools the agent can use"""
    
    @staticmethod
    def search_web(query: str) -> str:
        """Search the web for information"""
        # Implementation would use Bing/Google API
        return f"Search results for: {query}"
    
    @staticmethod
    def send_email(to: str, subject: str, body: str) -> str:
        """Send an email"""
        # Implementation would use SMTP or email API
        return f"Email sent to {to}"
    
    @staticmethod
    def read_file(filepath: str) -> str:
        """Read contents of a file"""
        with open(filepath, 'r') as f:
            return f.read()
    
    @staticmethod
    def execute_code(code: str) -> str:
        """Execute Python code safely"""
        # Implementation would use sandboxed execution
        try:
            exec(code)
            return "Code executed successfully"
        except Exception as e:
            return f"Error: {str(e)}"
    
    @staticmethod
    def query_database(sql: str) -> str:
        """Query a database"""
        # Implementation would use SQLAlchemy or similar
        return "Query results..."
    
    def get_tool_descriptions(self):
        """Return descriptions of all tools for LLM"""
        tools = []
        for name, func in self.__class__.__dict__.items():
            if not name.startswith('_') and callable(func):
                tools.append({
                    "name": name,
                    "description": func.__doc__,
                    "parameters": func.__annotations__
                })
        return tools

from datetime import datetime
from collections import deque

class AgentMemory:
    def __init__(self, max_short_term=10):
        # Short-term memory: recent actions and observations
        self.short_term = deque(maxlen=max_short_term)
        
        # Long-term memory: important facts and learnings
        self.long_term = []
        
        # Episodic memory: past task completions
        self.episodes = []
        
        # Working memory: current task context
        self.working = {
            "goal": None,
            "plan": [],
            "completed_steps": [],
            "current_observation": None
        }
    
    def add_observation(self, observation, action_taken, result):
        """Store a short-term memory"""
        memory_item = {
            "timestamp": datetime.now(),
            "observation": observation,
            "action": action_taken,
            "result": result
        }
        self.short_term.append(memory_item)
    
    def save_to_long_term(self, fact, importance="high"):
        """Save important information permanently"""
        self.long_term.append({
            "timestamp": datetime.now(),
            "fact": fact,
            "importance": importance
        })
    
    def complete_episode(self, goal, success, summary):
        """Record completion of a task"""
        self.episodes.append({
            "timestamp": datetime.now(),
            "goal": goal,
            "success": success,
            "summary": summary,
            "actions_taken": len(self.short_term)
        })
    
    def get_relevant_context(self, query):
        """Retrieve relevant memories for current task"""
        # In production, use vector search here
        recent = list(self.short_term)[-5:]
        relevant_facts = [f for f in self.long_term 
                         if f['importance'] == 'high']
        return {
            "recent_actions": recent,
            "relevant_facts": relevant_facts
        }

class AgentPerception:
    """Handle all forms of perception/input"""
    
    def perceive_user_input(self, user_message):
        """Process user messages"""
        return {
            "type": "user_input",
            "content": user_message,
            "timestamp": datetime.now()
        }
    
    def perceive_environment(self):
        """Check environment state"""
        return {
            "type": "environment",
            "disk_space": "500GB free",
            "network": "connected",
            "time": datetime.now(),
            "system_load": "normal"
        }
    
    def perceive_tool_result(self, tool_name, result):
        """Process tool execution results"""
        return {
            "type": "tool_result",
            "tool": tool_name,
            "result": result,
            "success": "error" not in str(result).lower()
        }
    
    def perceive_external_event(self, event):
        """Handle external triggers (webhooks, notifications)"""
        return {
            "type": "external_event",
            "event": event,
            "requires_action": True
        }

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

class ResearchAgent:
    def __init__(self, api_key):
        self.client = openai.OpenAI(api_key=api_key)
        self.memory = []
        self.max_iterations = 10
    
    def run(self, goal: str) -> str:
        """Main agent loop"""
        print(f"🎯 Goal: {goal}\n")
        
        # Initialize
        observation = f"Starting task: {goal}"
        iterations = 0
        
        while iterations < self.max_iterations:
            # PERCEIVE
            print(f"📊 Observation: {observation}\n")
            
            # THINK
            decision = self._reason(observation, goal)
            print(f"🧠 Decision:\n{decision}\n")
            
            # Determine if task is complete
            if "TASK_COMPLETE" in decision:
                print("✅ Task completed!")
                return self._generate_final_report()
            
            # ACT
            action, params = self._parse_action(decision)
            observation = self._execute_action(action, params)
            
            # Store in memory
            self.memory.append({
                "decision": decision,
                "action": action,
                "observation": observation
            })
            
            iterations += 1
        
        return "Max iterations reached. Task incomplete."
    
    def _reason(self, observation: str, goal: str) -> str:
        """LLM-based reasoning"""
        prompt = f"""
You are an autonomous research agent. Your goal: {goal}

Current observation: {observation}

Past actions: {json.dumps(self.memory[-3:], indent=2) if self.memory else 'None'}

Available actions:
- SEARCH(query): Search web for information
- ANALYZE(text): Extract key insights from text
- SYNTHESIZE: Combine findings into report
- SAVE_REPORT(content): Save report to file
- TASK_COMPLETE: Mark task as done

Think step-by-step and decide your next action.
Format: ACTION(parameters)
"""
        
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are a systematic research agent."},
                {"role": "user", "content": prompt}
            ]
        )
        
        return response.choices[0].message.content
    
    def _parse_action(self, decision: str) -> tuple:
        """Extract action and parameters from decision"""
        # Simple parsing (production would be more robust)
        if "SEARCH(" in decision:
            query = decision.split("SEARCH(")[1].split(")")[0]
            return ("search", query)
        elif "ANALYZE(" in decision:
            return ("analyze", "")
        elif "SYNTHESIZE" in decision:
            return ("synthesize", "")
        elif "SAVE_REPORT(" in decision:
            content = decision.split("SAVE_REPORT(")[1].split(")")[0]
            return ("save_report", content)
        else:
            return ("continue", "")
    
    def _execute_action(self, action: str, params: str) -> str:
        """Execute the chosen action"""
        print(f"⚙️  Executing: {action}({params})\n")
        
        if action == "search":
            # Simulate search (production would use real search API)
            return f"Found information about {params}: [simulated search results]"
        
        elif action == "analyze":
            return "Extracted key insights: [analysis results]"
        
        elif action == "synthesize":
            return "Report synthesized successfully"
        
        elif action == "save_report":
            with open("research_report.txt", "w") as f:
                f.write(params)
            return "Report saved to research_report.txt"
        
        return "Action completed"
    
    def _generate_final_report(self) -> str:
        """Generate final summary"""
        report = "Research Report\n" + "="*50 + "\n\n"
        for item in self.memory:
            report += f"Action: {item['action']}\n"
            report += f"Result: {item['observation']}\n\n"
        return report

# Usage
agent = ResearchAgent(api_key="your-api-key")
result = agent.run("Research the latest developments in quantum computing")

class ManagerAgent:
    def __init__(self):
        self.workers = {
            "researcher": ResearchWorker(),
            "writer": WriterWorker(),
            "reviewer": ReviewWorker()
        }
    
    def run(self, task):
        # Manager decides which workers to use
        if "research" in task.lower():
            results = self.workers["researcher"].execute(task)
            draft = self.workers["writer"].execute(results)
            final = self.workers["reviewer"].execute(draft)
            return final
        # ... more logic

class ResearchWorker:
    def execute(self, task):
        # Specialized research logic
        return "research results"

class WriterWorker:
    def execute(self, research_data):
        # Specialized writing logic
        return "written draft"

class ReviewWorker:
    def execute(self, draft):
        # Specialized review logic
        return "reviewed final"

from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime

@dataclass
class AgentObservation:
    """What the agent perceives"""
    type: str  # "user_input", "tool_result", "system_event"
    content: str
    timestamp: datetime
    metadata: Dict = None

@dataclass
class AgentAction:
    """What the agent decides to do"""
    tool_name: str
    parameters: Dict
    reasoning: str

@dataclass
class AgentState:
    """Current agent state"""
    goal: str
    plan: List[str]
    completed_steps: List[str]
    current_observation: Optional[AgentObservation]
    memory: List[Dict]
    iteration_count: int = 0

import smtplib
from email.mime.text import MIMEText
from datetime import datetime, timedelta
import requests

class PersonalAssistantTools:
    def __init__(self, config):
        self.config = config
    
    def check_calendar(self, date: str = None) -> str:
        """Check calendar for a specific date"""
        if not date:
            date = datetime.now().strftime("%Y-%m-%d")
        
        # Mock implementation (use Google Calendar API in production)
        mock_events = [
            {"time": "10:00 AM", "title": "Team Standup"},
            {"time": "2:00 PM", "title": "Client Meeting"},
        ]
        
        result = f"Calendar for {date}:\n"
        for event in mock_events:
            result += f"  {event['time']} - {event['title']}\n"
        
        return result
    
    def send_email(self, to: str, subject: str, body: str) -> str:
        """Send an email"""
        try:
            msg = MIMEText(body)
            msg['Subject'] = subject
            msg['From'] = self.config.get('email')
            msg['To'] = to
            
            # Use SMTP (configure for your provider)
            # In production: uncomment and configure
            # with smtplib.SMTP('smtp.gmail.com', 587) as server:
            #     server.starttls()
            #     server.login(self.config['email'], self.config['password'])
            #     server.send_message(msg)
            
            return f"✉️ Email sent to {to} with subject: {subject}"
        
        except Exception as e:
            return f"❌ Error sending email: {str(e)}"
    
    def web_search(self, query: str) -> str:
        """Search the web"""
        # Mock implementation (use Bing/Google API in production)
        return f"🔍 Search results for '{query}':\n1. Result one\n2. Result two\n3. Result three"
    
    def take_note(self, note: str) -> str:
        """Save a note"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        
        with open("agent_notes.txt", "a") as f:
            f.write(f"[{timestamp}] {note}\n")
        
        return f"📝 Note saved: {note}"
    
    def get_weather(self, location: str) -> str:
        """Get weather for a location"""
        # Mock implementation (use weather API in production)
        return f"☀️ Weather in {location}: 72°F, Sunny"

from openai import OpenAI
import json

class PersonalAssistantBrain:
    def __init__(self, api_key, tools):
        self.client = OpenAI(api_key=api_key)
        self.tools = tools
        self.system_prompt = """You are a personal assistant agent. You help users with:
- Calendar management
- Email sending
- Web research
- Note-taking
- Weather checks

When given a task:
1. Break it down into steps
2. Use available tools systematically
3. Verify results before proceeding
4. Report back to user clearly

Available tools:
- check_calendar(date)
- send_email(to, subject, body)
- web_search(query)
- take_note(note)
- get_weather(location)

Format decisions as:
ACTION: tool_name
PARAMETERS: {"param": "value"}
REASONING: Why this action
"""
    
    def decide_action(self, observation: str, state: AgentState) -> AgentAction:
        """Decide what to do next"""
        
        context = f"""
Current Goal: {state.goal}

Plan:
{chr(10).join(f"{i+1}. {step}" for i, step in enumerate(state.plan))}

Completed:
{chr(10).join(f"✓ {step}" for step in state.completed_steps)}

Current Observation: {observation}

Recent Memory:
{json.dumps(state.memory[-3:], indent=2) if state.memory else 'None'}

What should I do next?
"""
        
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": context}
            ],
            temperature=0.1
        )
        
        decision = response.choices[0].message.content
        
        # Parse decision
        action = self._parse_decision(decision)
        return action
    
    def _parse_decision(self, decision: str) -> AgentAction:
        """Parse LLM decision into structured action"""
        lines = decision.split('\n')
        
        tool_name = None
        parameters = {}
        reasoning = ""
        
        for line in lines:
            if line.startswith("ACTION:"):
                tool_name = line.split("ACTION:")[1].strip()
            elif line.startswith("PARAMETERS:"):
                param_str = line.split("PARAMETERS:")[1].strip()
                try:
                    parameters = json.loads(param_str)
                except:
                    parameters = {"raw": param_str}
            elif line.startswith("REASONING:"):
                reasoning = line.split("REASONING:")[1].strip()
        
        return AgentAction(
            tool_name=tool_name or "continue",
            parameters=parameters,
            reasoning=reasoning
        )

class PersonalAssistantAgent:
    def __init__(self, api_key, config):
        self.tools = PersonalAssistantTools(config)
        self.brain = PersonalAssistantBrain(api_key, self.tools)
        self.max_iterations = 15
    
    def run(self, user_request: str) -> str:
        """Main agent loop"""
        print(f"\n🤖 Personal Assistant Agent")
        print(f"📋 Request: {user_request}\n")
        
        # Initialize state
        state = AgentState(
            goal=user_request,
            plan=[],
            completed_steps=[],
            current_observation=None,
            memory=[]
        )
        
        observation = f"User request: {user_request}"
        
        # Agent loop
        while state.iteration_count < self.max_iterations:
            print(f"\n--- Iteration {state.iteration_count + 1} ---")
            print(f"📊 Observation: {observation}\n")
            
            # Decide next action
            action = self.brain.decide_action(observation, state)
            
            print(f"🧠 Reasoning: {action.reasoning}")
            print(f"⚙️  Action: {action.tool_name}({action.parameters})\n")
            
            # Check if done
            if action.tool_name == "task_complete":
                print("✅ Task completed successfully!")
                return self._generate_summary(state)
            
            # Execute action
            observation = self._execute_tool(
                action.tool_name,
                action.parameters
            )
            
            # Update state
            state.completed_steps.append(action.tool_name)
            state.memory.append({
                "action": action.tool_name,
                "parameters": action.parameters,
                "result": observation
            })
            state.iteration_count += 1
            
            print(f"📥 Result: {observation}")
        
        return "⚠️ Max iterations reached"
    
    def _execute_tool(self, tool_name: str, parameters: Dict) -> str:
        """Execute a tool and return result"""
        try:
            if tool_name == "check_calendar":
                return self.tools.check_calendar(
                    parameters.get('date')
                )
            
            elif tool_name == "send_email":
                return self.tools.send_email(
                    parameters.get('to'),
                    parameters.get('subject'),
                    parameters.get('body')
                )
            
            elif tool_name == "web_search":
                return self.tools.web_search(
                    parameters.get('query')
                )
            
            elif tool_name == "take_note":
                return self.tools.take_note(
                    parameters.get('note')
                )
            
            elif tool_name == "get_weather":
                return self.tools.get_weather(
                    parameters.get('location')
                )
            
            else:
                return f"Unknown tool: {tool_name}"
        
        except Exception as e:
            return f"❌ Error: {str(e)}"
    
    def _generate_summary(self, state: AgentState) -> str:
        """Generate summary of what was accomplished"""
        summary = f"\n{'='*50}\n"
        summary += "📊 TASK SUMMARY\n"
        summary += f"{'='*50}\n\n"
        summary += f"Goal: {state.goal}\n\n"
        summary += f"Actions Taken ({len(state.completed_steps)}):\n"
        
        for i, step in enumerate(state.completed_steps, 1):
            summary += f"  {i}. {step}\n"
        
        return summary

# Usage Example
if __name__ == "__main__":
    config = {
        "email": "your-email@example.com",
        "password": "your-password"
    }
    
    agent = PersonalAssistantAgent(
        api_key="your-openai-key",
        config=config
    )
    
    # Test the agent
    result = agent.run(
        "Check my calendar for today and send an email to team@company.com "
        "summarizing my meetings"
    )
    
    print(result)

MAX_ITERATIONS = 20  # Never run unbounded
MAX_TOOL_CALLS = 50  # Limit total tool executions
TIMEOUT_SECONDS = 300  # 5 minute hard limit

if iterations >= MAX_ITERATIONS:
    logger.warning("Max iterations reached")
    return {"status": "incomplete", "reason": "iteration_limit"}

import logging
import json

logging.basicConfig(
    filename='agent_actions.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def log_agent_action(iteration, observation, action, result):
    log_entry = {
        "iteration": iteration,
        "timestamp": datetime.now().isoformat(),
        "observation": observation,
        "action": action.tool_name,
        "parameters": action.parameters,
        "result": result[:200],  # Truncate long results
        "success": "error" not in result.lower()
    }
    logging.info(json.dumps(log_entry))

SENSITIVE_TOOLS = ['send_email', 'delete_file', 'make_purchase']

def execute_tool(tool_name, params):
    if tool_name in SENSITIVE_TOOLS:
        print(f"\n⚠️  Agent wants to: {tool_name}")
        print(f"Parameters: {params}")
        approval = input("Approve? (y/n): ")
        
        if approval.lower() != 'y':
            return "Action rejected by user"
    
    return actual_tool_execution(tool_name, params)