Agentic AI Systems: The Future of Autonomous AI in 2025

from typing import List, Dict, Any
import anthropic
import json

class AgentPlanner:
    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)

    def create_plan(self, goal: str, available_tools: List[Dict], context: Dict[str, Any]) -> Dict:
        """
        Create a step-by-step plan using Claude's advanced reasoning.

        Returns plan with steps, dependencies, and fallback strategies.
        """
        tools_description = "\n".join([
            f"- {tool['name']}: {tool['description']} (cost: {tool.get('cost_tokens', 0)} tokens)"
            for tool in available_tools
        ])

        prompt = f"""You are an AI planning agent. Create a detailed, efficient plan to achieve this goal.

Goal: {goal}

Available Tools:
{tools_description}

Current Context:
{json.dumps(context, indent=2)}

Create a plan with these requirements:
1. Break goal into 3-7 concrete steps
2. For each step, specify:
   - Action description
   - Tool to use
   - Expected input/output
   - Success criteria
   - Fallback if step fails
   - Estimated cost (tokens)
3. Identify dependencies between steps
4. Estimate total cost and time
5. Consider edge cases and error scenarios

Return JSON format:
{{
    "steps": [
        {{
            "id": 1,
            "action": "description",
            "tool": "tool_name",
            "input": {{}},
            "expected_output": "description",
            "success_criteria": "how to verify",
            "fallback": "alternative approach",
            "depends_on": [],
            "cost_estimate": 100
        }}
    ],
    "total_cost_estimate": 500,
    "total_time_estimate_seconds": 30,
    "risk_factors": ["factor1", "factor2"]
}}"""

        message = self.client.messages.create(
            model="claude-sonnet-4-5-20250929",
            max_tokens=2048,
            temperature=0,  # Deterministic planning
            messages=[{"role": "user", "content": prompt}]
        )

        plan_text = message.content[0].text
        plan = self.parse_plan(plan_text)

        # Validate plan
        self.validate_plan(plan, available_tools)

        return plan

    def validate_plan(self, plan: Dict, available_tools: List[Dict]) -> None:
        """Ensure plan is executable and tools exist."""
        tool_names = {t['name'] for t in available_tools}

        for step in plan['steps']:
            if step['tool'] not in tool_names:
                raise ValueError(f"Step {step['id']} uses unknown tool: {step['tool']}")

            # Check dependencies are valid
            for dep in step['depends_on']:
                if not any(s['id'] == dep for s in plan['steps']):
                    raise ValueError(f"Step {step['id']} has invalid dependency: {dep}")

    def parse_plan(self, plan_text: str) -> Dict:
        """Extract JSON plan from Claude's response."""
        # Find JSON block in response
        try:
            json_start = plan_text.find('{')
            json_end = plan_text.rfind('}') + 1
            plan_json = plan_text[json_start:json_end]
            return json.loads(plan_json)
        except Exception as e:
            raise ValueError(f"Failed to parse plan: {e}")

from typing import Callable, Dict, Any, Optional
from functools import wraps
import time
import logging
from datetime import datetime, timedelta
from collections import defaultdict

logger = logging.getLogger(__name__)

class ToolRegistry:
    def __init__(self):
        self.tools: Dict[str, Dict] = {}
        self.usage_stats = defaultdict(lambda: {"calls": 0, "errors": 0, "total_time": 0})
        self.rate_limits = {}
        self.call_history = defaultdict(list)

    def register(
        self,
        name: str,
        description: str,
        cost_tokens: int = 0,
        rate_limit_per_minute: Optional[int] = None,
        timeout_seconds: int = 30
    ):
        """Register a tool with metadata and constraints."""
        def decorator(func: Callable):
            @wraps(func)
            async def wrapper(*args, **kwargs):
                # Rate limiting
                if rate_limit_per_minute:
                    self._check_rate_limit(name, rate_limit_per_minute)

                # Execution with timeout and error handling
                start_time = time.time()
                try:
                    result = await asyncio.wait_for(
                        func(*args, **kwargs),
                        timeout=timeout_seconds
                    )

                    # Track success
                    self.usage_stats[name]["calls"] += 1
                    self.usage_stats[name]["total_time"] += time.time() - start_time

                    logger.info(f"Tool {name} executed successfully in {time.time() - start_time:.2f}s")
                    return result

                except asyncio.TimeoutError:
                    self.usage_stats[name]["errors"] += 1
                    logger.error(f"Tool {name} timed out after {timeout_seconds}s")
                    raise ToolExecutionError(f"{name} timed out")

                except Exception as e:
                    self.usage_stats[name]["errors"] += 1
                    logger.error(f"Tool {name} failed: {e}")
                    raise ToolExecutionError(f"{name} failed: {str(e)}")

            self.tools[name] = {
                'function': wrapper,
                'original_function': func,
                'description': description,
                'cost_tokens': cost_tokens,
                'schema': self._extract_schema(func),
                'rate_limit': rate_limit_per_minute,
                'timeout': timeout_seconds
            }
            return wrapper
        return decorator

    def _check_rate_limit(self, tool_name: str, limit_per_minute: int):
        """Enforce per-minute rate limits."""
        now = datetime.now()
        cutoff = now - timedelta(minutes=1)

        # Remove old calls
        self.call_history[tool_name] = [
            t for t in self.call_history[tool_name] if t > cutoff
        ]

        if len(self.call_history[tool_name]) >= limit_per_minute:
            raise RateLimitError(
                f"Rate limit exceeded for {tool_name}: "
                f"{limit_per_minute} calls/minute"
            )

        self.call_history[tool_name].append(now)

    def _extract_schema(self, func: Callable) -> Dict:
        """Extract parameter schema from function signature."""
        import inspect
        sig = inspect.signature(func)

        schema = {
            "parameters": {},
            "returns": str(sig.return_annotation) if sig.return_annotation != inspect.Signature.empty else "Any"
        }

        for param_name, param in sig.parameters.items():
            schema["parameters"][param_name] = {
                "type": str(param.annotation) if param.annotation != inspect.Parameter.empty else "Any",
                "required": param.default == inspect.Parameter.empty
            }

        return schema

    async def execute(self, tool_name: str, **kwargs) -> Any:
        """Execute a registered tool with parameters."""
        tool = self.tools.get(tool_name)
        if not tool:
            available = ", ".join(self.tools.keys())
            raise ValueError(
                f"Tool '{tool_name}' not found. Available: {available}"
            )

        return await tool['function'](**kwargs)

    def get_tool_descriptions(self) -> List[Dict]:
        """Get all tool descriptions for agent planning."""
        return [
            {
                "name": name,
                "description": meta["description"],
                "cost_tokens": meta["cost_tokens"],
                "schema": meta["schema"]
            }
            for name, meta in self.tools.items()
        ]

    def get_usage_stats(self) -> Dict:
        """Get tool usage statistics for monitoring."""
        return dict(self.usage_stats)

# Example tool registration
tools = ToolRegistry()

@tools.register(
    "search_web",
    "Search the internet for current information",
    cost_tokens=500,
    rate_limit_per_minute=10,
    timeout_seconds=15
)
async def search_web(query: str, max_results: int = 5) -> Dict[str, Any]:
    """Search web and return structured results."""
    # Implementation using Tavily, Bing, or Google Custom Search
    return {
        "query": query,
        "results": [
            {"title": "Result 1", "url": "https://...", "snippet": "..."},
        ],
        "timestamp": datetime.now().isoformat()
    }

@tools.register(
    "execute_python",
    "Execute Python code in a sandboxed environment",
    cost_tokens=200,
    rate_limit_per_minute=5,
    timeout_seconds=30
)
async def execute_python(code: str) -> Dict[str, Any]:
    """Execute code safely with resource limits."""
    import subprocess

    try:
        result = subprocess.run(
            ["python", "-c", code],
            capture_output=True,
            text=True,
            timeout=30,
            check=False
        )

        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "exit_code": result.returncode,
            "success": result.returncode == 0
        }
    except subprocess.TimeoutExpired:
        return {
            "error": "Execution timed out after 30 seconds",
            "success": False
        }

@tools.register(
    "query_database",
    "Query internal database with SQL",
    cost_tokens=100,
    rate_limit_per_minute=20,
    timeout_seconds=10
)
async def query_database(sql: str, params: Dict = None) -> List[Dict]:
    """Execute read-only SQL queries."""
    # Validate query is read-only
    if not sql.strip().upper().startswith('SELECT'):
        raise ValueError("Only SELECT queries allowed")

    # Execute with connection pooling
    # return results
    pass

from typing import List, Dict, Any, Optional
from datetime import datetime
import numpy as np
from sentence_transformers import SentenceTransformer

class AgentMemory:
    def __init__(self, embedding_model: str = "all-MiniLM-L6-v2"):
        self.short_term: List[Dict] = []  # Recent interactions (last 20)
        self.long_term: Dict[str, Any] = {}  # Summarized sessions
        self.working_memory: Dict[str, Any] = {}  # Current task context
        self.embedding_model = SentenceTransformer(embedding_model)

        # Metrics
        self.interactions_count = 0
        self.consolidations_count = 0

    def add_interaction(
        self,
        role: str,
        content: str,
        metadata: Optional[Dict] = None
    ) -> None:
        """Add interaction to short-term memory."""
        interaction = {
            'id': self.interactions_count,
            'role': role,
            'content': content,
            'timestamp': datetime.now(),
            'metadata': metadata or {},
            'embedding': self.embedding_model.encode(content)
        }
        self.short_term.append(interaction)
        self.interactions_count += 1

        # Auto-consolidate when short-term memory is full
        if len(self.short_term) > 20:
            self.consolidate_memory()

    def consolidate_memory(self) -> None:
        """Summarize old interactions and move to long-term storage."""
        if len(self.short_term) < 10:
            return

        # Take oldest 10 interactions
        to_consolidate = self.short_term[:10]

        # Generate summary using LLM
        summary = self._generate_summary(to_consolidate)

        # Store in long-term memory
        session_id = f'session_{self.consolidations_count}'
        self.long_term[session_id] = {
            'summary': summary,
            'interaction_ids': [i['id'] for i in to_consolidate],
            'timestamp': datetime.now(),
            'embedding': self.embedding_model.encode(summary)
        }

        # Remove from short-term
        self.short_term = self.short_term[10:]
        self.consolidations_count += 1

    def get_relevant_context(
        self,
        query: str,
        max_items: int = 5
    ) -> List[Dict]:
        """Retrieve relevant memories using semantic search."""
        query_embedding = self.embedding_model.encode(query)

        # Search short-term memory
        short_term_results = self._semantic_search(
            query_embedding,
            [i['embedding'] for i in self.short_term],
            [i for i in self.short_term],
            max_items=max_items // 2
        )

        # Search long-term memory
        long_term_results = self._semantic_search(
            query_embedding,
            [v['embedding'] for v in self.long_term.values()],
            list(self.long_term.values()),
            max_items=max_items // 2
        )

        return short_term_results + long_term_results

    def _semantic_search(
        self,
        query_embedding: np.ndarray,
        corpus_embeddings: List[np.ndarray],
        corpus_items: List[Any],
        max_items: int
    ) -> List[Any]:
        """Find most similar items using cosine similarity."""
        if not corpus_embeddings:
            return []

        # Compute similarities
        similarities = [
            np.dot(query_embedding, emb) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(emb)
            )
            for emb in corpus_embeddings
        ]

        # Get top k
        top_indices = np.argsort(similarities)[-max_items:][::-1]
        return [corpus_items[i] for i in top_indices]

    def _generate_summary(self, interactions: List[Dict]) -> str:
        """Generate summary of interactions using LLM."""
        # Concatenate interaction content
        text = "\n".join([
            f"{i['role']}: {i['content']}"
            for i in interactions
        ])

        # Use LLM to summarize (simplified)
        summary = f"Summary of {len(interactions)} interactions"
        return summary

    def clear_working_memory(self) -> None:
        """Clear task-specific context."""
        self.working_memory = {}

import asyncio
from typing import Dict, Any, List
import logging

logger = logging.getLogger(__name__)

class ProductionAgent:
    def __init__(
        self,
        planner: AgentPlanner,
        tools: ToolRegistry,
        memory: AgentMemory,
        max_iterations: int = 10,
        budget_tokens: int = 100000
    ):
        self.planner = planner
        self.tools = tools
        self.memory = memory
        self.max_iterations = max_iterations
        self.budget_tokens = budget_tokens
        self.tokens_used = 0

    async def execute(self, goal: str, context: Dict[str, Any] = None) -> Dict[str, Any]:
        """
        Execute goal autonomously with planning, execution, and reflection.

        Returns:
            {
                "success": bool,
                "result": Any,
                "steps_executed": int,
                "tokens_used": int,
                "execution_time_seconds": float,
                "errors": List[str]
            }
        """
        start_time = time.time()
        context = context or {}
        errors = []

        try:
            # Phase 1: Planning
            logger.info(f"Planning for goal: {goal}")
            available_tools = self.tools.get_tool_descriptions()
            plan = self.planner.create_plan(goal, available_tools, context)

            logger.info(f"Created plan with {len(plan['steps'])} steps")

            # Phase 2: Execution
            results = {}
            for step_num, step in enumerate(plan['steps'], 1):
                if self.tokens_used >= self.budget_tokens:
                    raise BudgetExceededError(
                        f"Token budget exceeded: {self.tokens_used}/{self.budget_tokens}"
                    )

                # Check dependencies
                if not self._dependencies_met(step, results):
                    logger.warning(f"Step {step_num} dependencies not met, skipping")
                    continue

                # Execute step
                logger.info(f"Executing step {step_num}/{len(plan['steps'])}: {step['action']}")

                try:
                    result = await self.tools.execute(
                        step['tool'],
                        **step['input']
                    )

                    results[step['id']] = {
                        "success": True,
                        "result": result,
                        "step": step
                    }

                    # Store in memory
                    self.memory.add_interaction(
                        role="agent",
                        content=f"Executed {step['action']}: {result}",
                        metadata={"step_id": step['id'], "success": True}
                    )

                    # Update token usage
                    self.tokens_used += step.get('cost_estimate', 0)

                except Exception as e:
                    logger.error(f"Step {step_num} failed: {e}")
                    errors.append(f"Step {step_num} ({step['action']}): {str(e)}")

                    # Try fallback if available
                    if step.get('fallback'):
                        logger.info(f"Attempting fallback for step {step_num}")
                        # Implement fallback logic

                    results[step['id']] = {
                        "success": False,
                        "error": str(e),
                        "step": step
                    }

            # Phase 3: Result synthesis
            final_result = self._synthesize_results(results, goal)

            execution_time = time.time() - start_time

            return {
                "success": len(errors) == 0,
                "result": final_result,
                "steps_executed": len(results),
                "tokens_used": self.tokens_used,
                "execution_time_seconds": execution_time,
                "errors": errors,
                "plan": plan,
                "step_results": results
            }

        except Exception as e:
            logger.error(f"Agent execution failed: {e}")
            return {
                "success": False,
                "error": str(e),
                "tokens_used": self.tokens_used,
                "execution_time_seconds": time.time() - start_time,
                "errors": errors + [str(e)]
            }

    def _dependencies_met(self, step: Dict, results: Dict) -> bool:
        """Check if step dependencies have been successfully executed."""
        for dep_id in step.get('depends_on', []):
            if dep_id not in results or not results[dep_id]['success']:
                return False
        return True

    def _synthesize_results(self, step_results: Dict, goal: str) -> Any:
        """Combine step results into final output."""
        # Use LLM to synthesize final result from all step outputs
        successful_results = [
            r for r in step_results.values() if r['success']
        ]

        if not successful_results:
            return None

        # Return last successful result as final output (simplified)
        return successful_results[-1]['result']

class LoopDetector:
    def __init__(self, max_repeats: int = 3):
        self.action_history = []
        self.max_repeats = max_repeats

    def check_action(self, action: str) -> bool:
        """Returns False if action would create a loop."""
        recent = self.action_history[-self.max_repeats:]
        if recent.count(action) >= self.max_repeats:
            return False  # Loop detected

        self.action_history.append(action)
        return True

def validate_tool_call(tool_name: str, available_tools: List[str]) -> None:
    if tool_name not in available_tools:
        raise ToolNotFoundError(
            f"Tool '{tool_name}' not available. "
            f"Available: {', '.join(available_tools)}"
        )

def manage_context(history: List[Dict], max_tokens: int = 100000) -> List[Dict]:
    """Keep recent messages, summarize old ones."""
    current_tokens = sum(len(m['content']) // 4 for m in history)

    if current_tokens <= max_tokens:
        return history

    # Keep last 10 messages, summarize rest
    recent = history[-10:]
    old = history[:-10]

    summary = summarize_conversations(old)
    return [{"role": "system", "content": summary}] + recent