Building Your Own Framework

from typing import List, Dict, Any
import openai

class Agent:
    def __init__(self, model: str = "gpt-4", max_iterations: int = 10):
        self.model = model
        self.max_iterations = max_iterations
        self.tools: Dict[str, callable] = {}
        self.memory: List[Dict] = []
        
    def register_tool(self, name: str, func: callable, description: str):
        """Register a tool that the agent can use"""
        self.tools[name] = {
            "function": func,
            "description": description
        }
    
    def run(self, task: str) -> str:
        """Main agent loop: observe → reason → act → update"""
        self.memory.append({"role": "user", "content": task})
        
        for iteration in range(self.max_iterations):
            # Observe: Get current state
            context = self._build_context()
            
            # Reason: LLM decides next action
            decision = self._reason(context)
            
            # Act: Execute tool or return answer
            if decision["type"] == "tool_call":
                result = self._execute_tool(decision["tool"], decision["args"])
                self.memory.append({"role": "tool", "content": result})
            elif decision["type"] == "final_answer":
                return decision["answer"]
        
        return "Max iterations reached"
    
    def _reason(self, context: str) -> Dict:
        """LLM decides next action"""
        response = openai.chat.completions.create(
            model=self.model,
            messages=[{"role": "system", "content": context}]
        )
        # Parse response to determine tool call or final answer
        return self._parse_decision(response)
    
    def _execute_tool(self, tool_name: str, args: Dict) -> Any:
        """Execute a registered tool"""
        if tool_name not in self.tools:
            return f"Error: Tool '{tool_name}' not found"
        
        try:
            return self.tools[tool_name]["function"](**args)
        except Exception as e:
            return f"Error executing {tool_name}: {str(e)}"

class Memory:
    def __init__(self, max_short_term: int = 20):
        self.short_term: List[Dict] = []  # Recent conversation
        self.long_term: Dict[str, Any] = {}  # Facts, summaries
        self.max_short_term = max_short_term
    
    def add_message(self, role: str, content: str):
        """Add message to short-term memory"""
        self.short_term.append({"role": role, "content": content})
        
        # Trim if exceeds max
        if len(self.short_term) > self.max_short_term:
            self._consolidate_old_messages()
    
    def _consolidate_old_messages(self):
        """Move old messages to long-term storage"""
        # Summarize oldest messages
        old_messages = self.short_term[:5]
        summary = self._summarize(old_messages)
        
        # Store in long-term
        self.long_term[f"summary_{len(self.long_term)}"] = summary
        
        # Remove from short-term
        self.short_term = self.short_term[5:]
    
    def get_context(self) -> str:
        """Build context for LLM from memory"""
        context = []
        
        # Add long-term summaries
        for key, summary in self.long_term.items():
            context.append(f"[Previous: {summary}]")
        
        # Add recent short-term messages
        for msg in self.short_term:
            context.append(f"{msg['role']}: {msg['content']}")
        
        return "\n".join(context)