Agentic AI in Production

from langchain.agents import Tool, AgentExecutor, initialize_agent

class DevinAgent:
    """ReAct loop: Reason → Act → Observe. Maintains state across hours."""

    def __init__(self):
        self.tools = self.initialize_tools()
        self.agent = initialize_agent(
            tools=self.tools,
            llm="gpt-4",
            agent_type="react",
            max_iterations=500,      # cap runaway loops
            early_stopping_method="generate",
        )
        self.workspace = PersistentWorkspace()
        self.checkpointer = Checkpointer(interval_seconds=300)

    def initialize_tools(self):
        return [
            Tool(name="code_editor", func=self.edit_code,
                 description="Read, write, or modify code files"),
            Tool(name="terminal", func=self.run_command,
                 description="Execute shell commands in sandbox"),
            Tool(name="browser", func=self.browse_web,
                 description="Navigate web pages, read documentation"),
            Tool(name="git", func=self.git_operation,
                 description="Git operations: commit, branch, PR"),
        ]

    def execute_task(self, task_description):
        """Execute multi-hour task with checkpointing."""
        plan = self.create_plan(task_description)

        for i, sub_task in enumerate(plan):
            print(f"[{i+1}/{len(plan)}]: {sub_task}")
            try:
                result = self.agent.run(sub_task)
                self.checkpointer.save({
                    'task': task_description,
                    'completed_subtasks': i + 1,
                    'workspace': self.workspace.state,
                    'result': result,
                })
            except Exception as e:
                print(f"Error: {e}. Self-debugging…")
                if not self.self_debug(sub_task, e):
                    self.request_human_help(sub_task, e)

        return self.workspace.state

    def self_debug(self, task, error):
        """Read logs, form hypothesis, try fixes."""
        debug_prompt = f"""
Task: {task}
Error: {error}

Analyze and attempt to fix:
1. Read relevant logs
2. Form hypothesis about root cause
3. Propose and test fix
4. Verify solution
"""
        return self.agent.run(debug_prompt)

import docker
import tempfile, os

class SandboxedExecutor:
    """Isolated environment — prevents damage to host system."""

    def __init__(self):
        self.client = docker.from_env()
        self.container = None

    def create_sandbox(self):
        self.container = self.client.containers.run(
            "python:3.11-slim",
            command="sleep infinity",   # keep container alive
            detach=True,
            mem_limit="2g",             # memory limit
            cpu_quota=50000,            # 50% of 1 core
            network_mode="bridge",      # isolated network
            remove=True,                # auto-cleanup on stop
            volumes={
                self.workspace_path: {'bind': '/workspace', 'mode': 'rw'},
            },
        )

    def execute_code(self, code, timeout=30):
        with tempfile.NamedTemporaryFile(
            mode='w', suffix='.py', delete=False,
        ) as f:
            f.write(code)
            code_file = f.name

        try:
            self.copy_to_container(code_file, '/tmp/code.py')
            exit_code, output = self.container.exec_run(
                cmd="python /tmp/code.py",
                workdir="/workspace",
                demux=True,
                stream=False,
            )
            stdout, stderr = output
            return {
                'exit_code': exit_code,
                'stdout': stdout.decode() if stdout else '',
                'stderr': stderr.decode() if stderr else '',
                'success': exit_code == 0,
            }
        except Exception as e:
            return {'exit_code': -1, 'stdout': '', 'stderr': str(e),
                    'success': False}
        finally:
            os.remove(code_file)

    def cleanup(self):
        if self.container:
            self.container.stop()

class AutoGPTAgent:
    """Self-prompting loop. Agent generates own next action."""

    def __init__(self, goal, max_iterations=100):
        self.goal = goal
        self.max_iterations = max_iterations
        self.memory = []                  # short-term
        self.vector_db = ChromaDB()       # long-term

    def run(self):
        plan = self.create_plan(self.goal)

        for iteration in range(self.max_iterations):
            context = self.get_context(plan)
            prompt = self.build_prompt(context, plan)
            response = self.call_gpt4(prompt)

            action = self.parse_action(response)
            result = self.execute_action(action)

            self.memory.append({
                'action': action, 'result': result,
                'timestamp': time.time(),
            })
            self.vector_db.add(result)

            if self.is_goal_complete(result, plan):
                return {"status": "success", "result": result}

            if len(self.memory) > 10:    # prune short-term
                self.memory = self.memory[-10:]

        return {"status": "max_iterations_reached"}

    def build_prompt(self, context, plan):
        return f"""
Goal: {self.goal}
Plan: {plan}
Current Progress: {context}

What should you do next to make progress toward the goal?
Respond with a specific action (e.g., search_web, write_code, read_file).
"""

class HybridMemorySystem:
    """Short-term for immediate context, long-term for semantic retrieval."""

    def __init__(self, chroma_collection):
        self.short_term = []
        self.long_term = chroma_collection
        self.max_short_term = 10

    def add(self, action, result):
        entry = {
            'action': action, 'result': result,
            'timestamp': time.time(),
        }
        self.short_term.append(entry)
        if len(self.short_term) > self.max_short_term:
            self.short_term.pop(0)

        self.long_term.add(
            documents=[f"{action}: {result}"],
            metadatas=[{"type": "agent_action",
                        "timestamp": entry['timestamp']}],
            ids=[f"action_{time.time()}"],
        )

    def get_context(self, current_goal):
        relevant = self.long_term.query(
            query_texts=[current_goal], n_results=5,
        )
        return {
            'recent_actions': self.short_term.copy(),
            'relevant_past':  relevant['documents'][0],
        }

class PluginSystem:
    """Extensible tool execution — community adds capabilities."""

    def __init__(self):
        self.plugins = {}
        self.load_plugins()

    def load_plugins(self):
        self.plugins['web_search']     = GoogleSearchPlugin()
        self.plugins['browse_web']     = SeleniumBrowserPlugin()
        self.plugins['write_file']     = FileSystemPlugin()
        self.plugins['execute_python'] = PythonExecutorPlugin()
        self.plugins['read_file']      = FileSystemPlugin()
        # ... 100+ community plugins

    def execute(self, action):
        plugin_name = action['command']
        args = action['args']

        if plugin_name not in self.plugins:
            return {"error": f"Unknown command: {plugin_name}"}

        try:
            result = self.plugins[plugin_name].execute(**args)
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}


class GoogleSearchPlugin:
    def execute(self, query, num_results=10):
        results = google_search(query, num_results)
        return [{"title": r.title, "url": r.url, "snippet": r.snippet}
                for r in results]