# Create a virtual environment python -m venv ai-agent-env source ai-agent-env/bin/activate # On Windows: ai-agent-env\Scripts\activate # Install required packages -weight: 500;">pip -weight: 500;">install openai langchain python-dotenv # Create a virtual environment python -m venv ai-agent-env source ai-agent-env/bin/activate # On Windows: ai-agent-env\Scripts\activate # Install required packages -weight: 500;">pip -weight: 500;">install openai langchain python-dotenv # Create a virtual environment python -m venv ai-agent-env source ai-agent-env/bin/activate # On Windows: ai-agent-env\Scripts\activate # Install required packages -weight: 500;">pip -weight: 500;">install openai langchain python-dotenv from openai import OpenAI from typing import List, Dict, Any from dataclasses import dataclass from enum import Enum class AgentAction(Enum): """Available actions the agent can take""" THINK = "think" SEARCH = "search" EXECUTE = "execute" RESPOND = "respond" @dataclass class Tool: """Represents a tool the agent can use""" name: str description: str function: callable class AutonomousAgent: def __init__(self, api_key: str, system_prompt: str = None): self.client = OpenAI(api_key=api_key) self.tools: Dict[str, Tool] = {} self.conversation_history = [] self.system_prompt = system_prompt or """You are an autonomous agent that can: - Think: Analyze the current situation - Search: Look up information - Execute: Perform actions using available tools - Respond: Provide final answers to the user For each task, determine the best sequence of actions.""" def register_tool(self, tool: Tool): """Register a new tool the agent can use""" self.tools[tool.name] = tool def think(self, prompt: str) -> str: """Use the LLM to reason about the next action""" messages = [ {"role": "system", "content": self.system_prompt}, *self.conversation_history, {"role": "user", "content": prompt} ] response = self.client.chat.completions.create( model="gpt-4", messages=messages, temperature=0.7 ) return response.choices[0].message.content def run(self, task: str, max_iterations: int = 10): """Execute a task autonomously""" self.conversation_history.append({"role": "user", "content": task}) for iteration in range(max_iterations): # Agent decides what to do next thought = self.think(f"What should I do next for: {task}") print(f"[Iteration {iteration + 1}] {thought}") # Check if task is complete if "complete" in thought.lower() or "done" in thought.lower(): return thought # Here you would implement tool execution logic # For now, we demonstrate the thinking process return "Task could not be completed within iteration limit" from openai import OpenAI from typing import List, Dict, Any from dataclasses import dataclass from enum import Enum class AgentAction(Enum): """Available actions the agent can take""" THINK = "think" SEARCH = "search" EXECUTE = "execute" RESPOND = "respond" @dataclass class Tool: """Represents a tool the agent can use""" name: str description: str function: callable class AutonomousAgent: def __init__(self, api_key: str, system_prompt: str = None): self.client = OpenAI(api_key=api_key) self.tools: Dict[str, Tool] = {} self.conversation_history = [] self.system_prompt = system_prompt or """You are an autonomous agent that can: - Think: Analyze the current situation - Search: Look up information - Execute: Perform actions using available tools - Respond: Provide final answers to the user For each task, determine the best sequence of actions.""" def register_tool(self, tool: Tool): """Register a new tool the agent can use""" self.tools[tool.name] = tool def think(self, prompt: str) -> str: """Use the LLM to reason about the next action""" messages = [ {"role": "system", "content": self.system_prompt}, *self.conversation_history, {"role": "user", "content": prompt} ] response = self.client.chat.completions.create( model="gpt-4", messages=messages, temperature=0.7 ) return response.choices[0].message.content def run(self, task: str, max_iterations: int = 10): """Execute a task autonomously""" self.conversation_history.append({"role": "user", "content": task}) for iteration in range(max_iterations): # Agent decides what to do next thought = self.think(f"What should I do next for: {task}") print(f"[Iteration {iteration + 1}] {thought}") # Check if task is complete if "complete" in thought.lower() or "done" in thought.lower(): return thought # Here you would implement tool execution logic # For now, we demonstrate the thinking process return "Task could not be completed within iteration limit" from openai import OpenAI from typing import List, Dict, Any from dataclasses import dataclass from enum import Enum class AgentAction(Enum): """Available actions the agent can take""" THINK = "think" SEARCH = "search" EXECUTE = "execute" RESPOND = "respond" @dataclass class Tool: """Represents a tool the agent can use""" name: str description: str function: callable class AutonomousAgent: def __init__(self, api_key: str, system_prompt: str = None): self.client = OpenAI(api_key=api_key) self.tools: Dict[str, Tool] = {} self.conversation_history = [] self.system_prompt = system_prompt or """You are an autonomous agent that can: - Think: Analyze the current situation - Search: Look up information - Execute: Perform actions using available tools - Respond: Provide final answers to the user For each task, determine the best sequence of actions.""" def register_tool(self, tool: Tool): """Register a new tool the agent can use""" self.tools[tool.name] = tool def think(self, prompt: str) -> str: """Use the LLM to reason about the next action""" messages = [ {"role": "system", "content": self.system_prompt}, *self.conversation_history, {"role": "user", "content": prompt} ] response = self.client.chat.completions.create( model="gpt-4", messages=messages, temperature=0.7 ) return response.choices[0].message.content def run(self, task: str, max_iterations: int = 10): """Execute a task autonomously""" self.conversation_history.append({"role": "user", "content": task}) for iteration in range(max_iterations): # Agent decides what to do next thought = self.think(f"What should I do next for: {task}") print(f"[Iteration {iteration + 1}] {thought}") # Check if task is complete if "complete" in thought.lower() or "done" in thought.lower(): return thought # Here you would implement tool execution logic # For now, we demonstrate the thinking process return "Task could not be completed within iteration limit" import json from datetime import datetime class WebSearchTool: """Example tool for web searching""" def __init__(self): self.name = "web_search" self.description = "Search the web for information" def execute(self, query: str) -> str: # In production, use actual search API return f"Results for: {query}" class CalculatorTool: """Example tool for calculations""" def __init__(self): self.name = "calculate" self.description = "Perform mathematical calculations" def execute(self, expression: str) -> str: try: # SECURITY: In production, use safe evaluation result = eval(expression, {"__builtins__": {}}, {}) return str(result) except Exception as e: return f"Error: {str(e)}" class FileManagerTool: """Example tool for file operations""" def __init__(self): self.name = "file_manager" self.description = "Read, write, or manipulate files" self.base_path = "./agent_workspace/" def execute(self, action: str, filename: str, content: str = None) -> str: import os os.makedirs(self.base_path, exist_ok=True) filepath = os.path.join(self.base_path, filename) if action == "read": with open(filepath, 'r') as f: return f.read() elif action == "write": with open(filepath, 'w') as f: f.write(content or "") return f"Written to {filename}" return "Unknown action" import json from datetime import datetime class WebSearchTool: """Example tool for web searching""" def __init__(self): self.name = "web_search" self.description = "Search the web for information" def execute(self, query: str) -> str: # In production, use actual search API return f"Results for: {query}" class CalculatorTool: """Example tool for calculations""" def __init__(self): self.name = "calculate" self.description = "Perform mathematical calculations" def execute(self, expression: str) -> str: try: # SECURITY: In production, use safe evaluation result = eval(expression, {"__builtins__": {}}, {}) return str(result) except Exception as e: return f"Error: {str(e)}" class FileManagerTool: """Example tool for file operations""" def __init__(self): self.name = "file_manager" self.description = "Read, write, or manipulate files" self.base_path = "./agent_workspace/" def execute(self, action: str, filename: str, content: str = None) -> str: import os os.makedirs(self.base_path, exist_ok=True) filepath = os.path.join(self.base_path, filename) if action == "read": with open(filepath, 'r') as f: return f.read() elif action == "write": with open(filepath, 'w') as f: f.write(content or "") return f"Written to {filename}" return "Unknown action" import json from datetime import datetime class WebSearchTool: """Example tool for web searching""" def __init__(self): self.name = "web_search" self.description = "Search the web for information" def execute(self, query: str) -> str: # In production, use actual search API return f"Results for: {query}" class CalculatorTool: """Example tool for calculations""" def __init__(self): self.name = "calculate" self.description = "Perform mathematical calculations" def execute(self, expression: str) -> str: try: # SECURITY: In production, use safe evaluation result = eval(expression, {"__builtins__": {}}, {}) return str(result) except Exception as e: return f"Error: {str(e)}" class FileManagerTool: """Example tool for file operations""" def __init__(self): self.name = "file_manager" self.description = "Read, write, or manipulate files" self.base_path = "./agent_workspace/" def execute(self, action: str, filename: str, content: str = None) -> str: import os os.makedirs(self.base_path, exist_ok=True) filepath = os.path.join(self.base_path, filename) if action == "read": with open(filepath, 'r') as f: return f.read() elif action == "write": with open(filepath, 'w') as f: f.write(content or "") return f"Written to {filename}" return "Unknown action" from collections import deque class AgentMemory: def __init__(self, max_short_term: int = 10, max_long_term: int = 100): self.short_term = deque(maxlen=max_short_term) self.long_term = [] self.important_memories = [] def add(self, experience: dict): """Add a new experience to memory""" self.short_term.append({ "timestamp": datetime.now().isoformat(), "content": experience }) # Transfer important memories to long-term if len(self.short_term) >= self.short_term.maxlen: self.long_term.extend(list(self.short_term)) def get_relevant(self, query: str, limit: int = 5) -> List[dict]: """Retrieve relevant memories (simplified)""" recent = list(self.short_term)[-limit:] return recent def mark_important(self, memory_index: int, reason: str): """Mark a memory as important for long-term retention""" if memory_index < len(self.short_term): important = self.short_term[memory_index] important["importance_reason"] = reason self.important_memories.append(important) from collections import deque class AgentMemory: def __init__(self, max_short_term: int = 10, max_long_term: int = 100): self.short_term = deque(maxlen=max_short_term) self.long_term = [] self.important_memories = [] def add(self, experience: dict): """Add a new experience to memory""" self.short_term.append({ "timestamp": datetime.now().isoformat(), "content": experience }) # Transfer important memories to long-term if len(self.short_term) >= self.short_term.maxlen: self.long_term.extend(list(self.short_term)) def get_relevant(self, query: str, limit: int = 5) -> List[dict]: """Retrieve relevant memories (simplified)""" recent = list(self.short_term)[-limit:] return recent def mark_important(self, memory_index: int, reason: str): """Mark a memory as important for long-term retention""" if memory_index < len(self.short_term): important = self.short_term[memory_index] important["importance_reason"] = reason self.important_memories.append(important) from collections import deque class AgentMemory: def __init__(self, max_short_term: int = 10, max_long_term: int = 100): self.short_term = deque(maxlen=max_short_term) self.long_term = [] self.important_memories = [] def add(self, experience: dict): """Add a new experience to memory""" self.short_term.append({ "timestamp": datetime.now().isoformat(), "content": experience }) # Transfer important memories to long-term if len(self.short_term) >= self.short_term.maxlen: self.long_term.extend(list(self.short_term)) def get_relevant(self, query: str, limit: int = 5) -> List[dict]: """Retrieve relevant memories (simplified)""" recent = list(self.short_term)[-limit:] return recent def mark_important(self, memory_index: int, reason: str): """Mark a memory as important for long-term retention""" if memory_index < len(self.short_term): important = self.short_term[memory_index] important["importance_reason"] = reason self.important_memories.append(important) class ReActAgent: """Agent using Reasoning + Acting pattern""" def __init__(self, api_key: str): self.client = OpenAI(api_key=api_key) self.tools = {} self.examples = [ { "task": "What's 15 + 27?", "thought": "I need to calculate 15 + 27", "action": "calculate", "action_input": "15 + 27", "observation": "42", "final_thought": "The answer is 42" } ] def run(self, task: str, max_steps: int = 15): """Execute task using ReAct pattern""" steps = [] for step in range(max_steps): # 1. Thought: Analyze the situation thought_prompt = self._build_thought_prompt(task, steps) thought = self._get_completion(thought_prompt) steps.append({"step": step + 1, "thought": thought}) # Check if we're done if self._is_complete(thought): return self._get_final_response(steps) # 2. Action: Decide what to do action_prompt = f"Based on: {thought}\nWhat action should I take?" action = self._get_completion(action_prompt).strip() steps[-1]["action"] = action # 3. Execute action and observe observation = self._execute_action(action) steps[-1]["observation"] = observation print(f"Step {step + 1}: {thought[:50]}... → {action}") return "Max steps reached" def _build_thought_prompt(self, task: str, steps: List[dict]) -> str: prompt = f"Task: {task}\n\n" prompt += "Previous steps:\n" for s in steps: prompt += f"- Thought: {s.get('thought', '')}\n" if 'action' in s: prompt += f" Action: {s['action']}\n" if 'observation' in s: prompt += f" Result: {s['observation']}\n" prompt += "\nWhat should I do next? Think step by step." return prompt def _get_completion(self, prompt: str) -> str: response = self.client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": prompt}], temperature=0.7 ) return response.choices[0].message.content def _execute_action(self, action: str) -> str: # Parse and execute the action # This is a simplified version return "Action executed successfully" def _is_complete(self, thought: str) -> bool: complete_indicators = ["final answer", "complete", "finished", "done"] return any(indicator in thought.lower() for indicator in complete_indicators) def _get_final_response(self, steps: List[dict]) -> str: return steps[-1].get("thought", "Task completed") class ReActAgent: """Agent using Reasoning + Acting pattern""" def __init__(self, api_key: str): self.client = OpenAI(api_key=api_key) self.tools = {} self.examples = [ { "task": "What's 15 + 27?", "thought": "I need to calculate 15 + 27", "action": "calculate", "action_input": "15 + 27", "observation": "42", "final_thought": "The answer is 42" } ] def run(self, task: str, max_steps: int = 15): """Execute task using ReAct pattern""" steps = [] for step in range(max_steps): # 1. Thought: Analyze the situation thought_prompt = self._build_thought_prompt(task, steps) thought = self._get_completion(thought_prompt) steps.append({"step": step + 1, "thought": thought}) # Check if we're done if self._is_complete(thought): return self._get_final_response(steps) # 2. Action: Decide what to do action_prompt = f"Based on: {thought}\nWhat action should I take?" action = self._get_completion(action_prompt).strip() steps[-1]["action"] = action # 3. Execute action and observe observation = self._execute_action(action) steps[-1]["observation"] = observation print(f"Step {step + 1}: {thought[:50]}... → {action}") return "Max steps reached" def _build_thought_prompt(self, task: str, steps: List[dict]) -> str: prompt = f"Task: {task}\n\n" prompt += "Previous steps:\n" for s in steps: prompt += f"- Thought: {s.get('thought', '')}\n" if 'action' in s: prompt += f" Action: {s['action']}\n" if 'observation' in s: prompt += f" Result: {s['observation']}\n" prompt += "\nWhat should I do next? Think step by step." return prompt def _get_completion(self, prompt: str) -> str: response = self.client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": prompt}], temperature=0.7 ) return response.choices[0].message.content def _execute_action(self, action: str) -> str: # Parse and execute the action # This is a simplified version return "Action executed successfully" def _is_complete(self, thought: str) -> bool: complete_indicators = ["final answer", "complete", "finished", "done"] return any(indicator in thought.lower() for indicator in complete_indicators) def _get_final_response(self, steps: List[dict]) -> str: return steps[-1].get("thought", "Task completed") class ReActAgent: """Agent using Reasoning + Acting pattern""" def __init__(self, api_key: str): self.client = OpenAI(api_key=api_key) self.tools = {} self.examples = [ { "task": "What's 15 + 27?", "thought": "I need to calculate 15 + 27", "action": "calculate", "action_input": "15 + 27", "observation": "42", "final_thought": "The answer is 42" } ] def run(self, task: str, max_steps: int = 15): """Execute task using ReAct pattern""" steps = [] for step in range(max_steps): # 1. Thought: Analyze the situation thought_prompt = self._build_thought_prompt(task, steps) thought = self._get_completion(thought_prompt) steps.append({"step": step + 1, "thought": thought}) # Check if we're done if self._is_complete(thought): return self._get_final_response(steps) # 2. Action: Decide what to do action_prompt = f"Based on: {thought}\nWhat action should I take?" action = self._get_completion(action_prompt).strip() steps[-1]["action"] = action # 3. Execute action and observe observation = self._execute_action(action) steps[-1]["observation"] = observation print(f"Step {step + 1}: {thought[:50]}... → {action}") return "Max steps reached" def _build_thought_prompt(self, task: str, steps: List[dict]) -> str: prompt = f"Task: {task}\n\n" prompt += "Previous steps:\n" for s in steps: prompt += f"- Thought: {s.get('thought', '')}\n" if 'action' in s: prompt += f" Action: {s['action']}\n" if 'observation' in s: prompt += f" Result: {s['observation']}\n" prompt += "\nWhat should I do next? Think step by step." return prompt def _get_completion(self, prompt: str) -> str: response = self.client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": prompt}], temperature=0.7 ) return response.choices[0].message.content def _execute_action(self, action: str) -> str: # Parse and execute the action # This is a simplified version return "Action executed successfully" def _is_complete(self, thought: str) -> bool: complete_indicators = ["final answer", "complete", "finished", "done"] return any(indicator in thought.lower() for indicator in complete_indicators) def _get_final_response(self, steps: List[dict]) -> str: return steps[-1].get("thought", "Task completed") class Guardrails: """Add safety guardrails to your agent""" def __init__(self): self.allowed_domains = ["general", "productivity"] self.blocked_patterns = ["harmful", "illegal", "malicious"] def validate_request(self, request: str) -> tuple[bool, str]: """Validate if request is allowed""" request_lower = request.lower() for blocked in self.blocked_patterns: if blocked in request_lower: return False, f"Request contains blocked content: {blocked}" return True, "Request allowed" class Guardrails: """Add safety guardrails to your agent""" def __init__(self): self.allowed_domains = ["general", "productivity"] self.blocked_patterns = ["harmful", "illegal", "malicious"] def validate_request(self, request: str) -> tuple[bool, str]: """Validate if request is allowed""" request_lower = request.lower() for blocked in self.blocked_patterns: if blocked in request_lower: return False, f"Request contains blocked content: {blocked}" return True, "Request allowed" class Guardrails: """Add safety guardrails to your agent""" def __init__(self): self.allowed_domains = ["general", "productivity"] self.blocked_patterns = ["harmful", "illegal", "malicious"] def validate_request(self, request: str) -> tuple[bool, str]: """Validate if request is allowed""" request_lower = request.lower() for blocked in self.blocked_patterns: if blocked in request_lower: return False, f"Request contains blocked content: {blocked}" return True, "Request allowed" import logging from functools import wraps def agent_error_handler(func): """Decorator for agent error handling""" @wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: logging.error(f"Agent error: {str(e)}") return { "-weight: 500;">status": "error", "message": str(e), "fallback_action": "Report error to user" } return wrapper import logging from functools import wraps def agent_error_handler(func): """Decorator for agent error handling""" @wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: logging.error(f"Agent error: {str(e)}") return { "-weight: 500;">status": "error", "message": str(e), "fallback_action": "Report error to user" } return wrapper import logging from functools import wraps def agent_error_handler(func): """Decorator for agent error handling""" @wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: logging.error(f"Agent error: {str(e)}") return { "-weight: 500;">status": "error", "message": str(e), "fallback_action": "Report error to user" } return wrapper import time from threading import Lock class RateLimiter: def __init__(self, max_calls: int, time_window: int): self.max_calls = max_calls self.time_window = time_window self.calls = [] self.lock = Lock() def allow_request(self) -> bool: with self.lock: now = time.time() self.calls = [t for t in self.calls if now - t < self.time_window] if len(self.calls) < self.max_calls: self.calls.append(now) return True return False import time from threading import Lock class RateLimiter: def __init__(self, max_calls: int, time_window: int): self.max_calls = max_calls self.time_window = time_window self.calls = [] self.lock = Lock() def allow_request(self) -> bool: with self.lock: now = time.time() self.calls = [t for t in self.calls if now - t < self.time_window] if len(self.calls) < self.max_calls: self.calls.append(now) return True return False import time from threading import Lock class RateLimiter: def __init__(self, max_calls: int, time_window: int): self.max_calls = max_calls self.time_window = time_window self.calls = [] self.lock = Lock() def allow_request(self) -> bool: with self.lock: now = time.time() self.calls = [t for t in self.calls if now - t < self.time_window] if len(self.calls) < self.max_calls: self.calls.append(now) return True return False from dotenv import load_dotenv import os load_dotenv() # Load from .env file api_key = os.getenv("OPENAI_API_KEY") agent = AutonomousAgent(api_key=api_key) from dotenv import load_dotenv import os load_dotenv() # Load from .env file api_key = os.getenv("OPENAI_API_KEY") agent = AutonomousAgent(api_key=api_key) from dotenv import load_dotenv import os load_dotenv() # Load from .env file api_key = os.getenv("OPENAI_API_KEY") agent = AutonomousAgent(api_key=api_key) FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN -weight: 500;">pip -weight: 500;">install --no-cache-dir -r requirements.txt COPY . . ENV OPENAI_API_KEY=${OPENAI_API_KEY} CMD ["python", "agent.py"] FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN -weight: 500;">pip -weight: 500;">install --no-cache-dir -r requirements.txt COPY . . ENV OPENAI_API_KEY=${OPENAI_API_KEY} CMD ["python", "agent.py"] FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN -weight: 500;">pip -weight: 500;">install --no-cache-dir -r requirements.txt COPY . . ENV OPENAI_API_KEY=${OPENAI_API_KEY} CMD ["python", "agent.py"] import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger("ai-agent") logger.info("Agent initialized successfully") import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger("ai-agent") logger.info("Agent initialized successfully") import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger("ai-agent") logger.info("Agent initialized successfully") class AgentTeam: def __init__(self): self.agents = {} def add_agent(self, role: str, agent: AutonomousAgent): self.agents[role] = agent def coordinate(self, task: str) -> str: """Coordinate multiple agents to solve a task""" # Simple coordination: route to appropriate agent # In production, use more sophisticated routing for role, agent in self.agents.items(): if role in task.lower(): return agent.run(task) return "No suitable agent found for task" class AgentTeam: def __init__(self): self.agents = {} def add_agent(self, role: str, agent: AutonomousAgent): self.agents[role] = agent def coordinate(self, task: str) -> str: """Coordinate multiple agents to solve a task""" # Simple coordination: route to appropriate agent # In production, use more sophisticated routing for role, agent in self.agents.items(): if role in task.lower(): return agent.run(task) return "No suitable agent found for task" class AgentTeam: def __init__(self): self.agents = {} def add_agent(self, role: str, agent: AutonomousAgent): self.agents[role] = agent def coordinate(self, task: str) -> str: """Coordinate multiple agents to solve a task""" # Simple coordination: route to appropriate agent # In production, use more sophisticated routing for role, agent in self.agents.items(): if role in task.lower(): return agent.run(task) return "No suitable agent found for task" - Agent Core: The main decision-making engine - Tools: Actions the agent can perform - Memory: Stores context and history - Planning: Breaks down complex tasks - Reflection: Evaluates actions and learns - Start with a clear architecture - Define the core components early - Add tools strategically - Extend capabilities as needed - Implement memory - Maintain context across interactions - Add guardrails - Safety should be built-in from the -weight: 500;">start - Monitor everything - You can't improve what you don't measure