LLM Agents in Production: Memory, Tools, and Planning That Actually Work

import anthropic
from collections import deque
from typing import Optional
 
 
class ContextManager:
    """Manages conversation context with token budget enforcement."""
 
    def __init__(self, max_tokens: int = 60_000):
        self.messages: list[dict] = []
        self.max_tokens = max_tokens
        self.client = anthropic.Anthropic()
 
    def add_message(self, role: str, content: str | list):
        self.messages.append({"role": role, "content": content})
        self._trim_if_needed()
 
    def _trim_if_needed(self):
        """Remove oldest messages (after system) if context is too long."""
        while len(self.messages) > 4:
            # Rough token estimate: 1 token ≈ 4 chars
            total_chars = sum(
                len(str(m["content"])) for m in self.messages
            )
            if total_chars < self.max_tokens * 4:
                break
            # Remove oldest user+assistant pair (keep last N exchanges)
            self.messages.pop(0)
            if self.messages and self.messages[0]["role"] == "assistant":
                self.messages.pop(0)
 
    def get_messages(self) -> list[dict]:
        return self.messages

import json
import hashlib
from pathlib import Path
from datetime import datetime
 
 
class AgentMemory:
    """
    Persistent memory for agents across sessions.
    Stores: facts learned, past actions, user preferences.
    """
 
    def __init__(self, agent_id: str, storage_dir: str = ".agent_memory"):
        self.agent_id = agent_id
        self.storage_path = Path(storage_dir) / f"{agent_id}.json"
        self.storage_path.parent.mkdir(exist_ok=True)
        self.memory = self._load()
 
    def _load(self) -> dict:
        if self.storage_path.exists():
            return json.loads(self.storage_path.read_text())
        return {
            "facts": {},
            "action_history": [],
            "learned_patterns": {},
            "created_at": datetime.utcnow().isoformat()
        }
 
    def _save(self):
        self.storage_path.write_text(json.dumps(self.memory, indent=2))
 
    def remember_fact(self, key: str, value: str, confidence: float = 1.0):
        """Store a fact with confidence score."""
        self.memory["facts"][key] = {
            "value": value,
            "confidence": confidence,
            "timestamp": datetime.utcnow().isoformat()
        }
        self._save()
 
    def recall_fact(self, key: str) -> Optional[str]:
        fact = self.memory["facts"].get(key)
        return fact["value"] if fact else None
 
    def log_action(self, action: str, result: str, success: bool):
        """Log what the agent did and whether it worked."""
        self.memory["action_history"].append({
            "action": action,
            "result": result[:500],
            "success": success,
            "timestamp": datetime.utcnow().isoformat()
        })
        # Keep last 100 actions
        self.memory["action_history"] = self.memory["action_history"][-100:]
        self._save()
 
    def get_relevant_context(self, query: str) -> str:
        """Return memory context relevant to the current query."""
        lines = []
 
        # Return all facts (in production, use embedding search)
        if self.memory["facts"]:
            lines.append("Known facts:")
            for key, fact in list(self.memory["facts"].items())[:10]:
                lines.append(f"- {key}: {fact['value']}")
 
        # Recent successful actions
        recent_successes = [
            a for a in self.memory["action_history"][-20:]
            if a["success"]
        ]
        if recent_successes:
            lines.append("\nRecent successful actions:")
            for action in recent_successes[-5:]:
                lines.append(f"- {action['action']}: {action['result'][:100]}")
 
        return "\n".join(lines)

import functools
import time
import logging
from typing import Callable, Any
 
 
logger = logging.getLogger(__name__)
 
 
def reliable_tool(
    max_retries: int = 3,
    retry_delay: float = 1.0,
    timeout: float = 30.0,
    fallback: Any = None
):
    """Decorator that makes tool calls reliable with retry + timeout + fallback."""
 
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
 
            for attempt in range(max_retries):
                try:
                    import signal
 
                    def timeout_handler(signum, frame):
                        raise TimeoutError(f"Tool {func.__name__} timed out after {timeout}s")
 
                    signal.signal(signal.SIGALRM, timeout_handler)
                    signal.alarm(int(timeout))
 
                    try:
                        result = func(*args, **kwargs)
                        signal.alarm(0)  # Cancel alarm
                        return result
                    finally:
                        signal.alarm(0)
 
                except TimeoutError as e:
                    last_error = e
                    logger.warning(f"Tool {func.__name__} attempt {attempt+1} timed out")
                    if attempt < max_retries - 1:
                        time.sleep(retry_delay * (2 ** attempt))
 
                except Exception as e:
                    last_error = e
                    logger.warning(f"Tool {func.__name__} attempt {attempt+1} failed: {e}")
                    if attempt < max_retries - 1:
                        time.sleep(retry_delay * (2 ** attempt))
 
            # All retries failed
            if fallback is not None:
                logger.error(f"Tool {func.__name__} failed after {max_retries} retries, using fallback")
                return fallback
 
            raise last_error
 
        return wrapper
    return decorator
 
 
# Usage
@reliable_tool(max_retries=3, timeout=10.0, fallback="kubectl not available")
def run_kubectl(command: str) -> str:
    import subprocess
    result = subprocess.run(
        ["kubectl"] + command.split(),
        capture_output=True, text=True, timeout=9
    )
    return result.stdout or result.stderr

import anthropic
import json
from dataclasses import dataclass, field
from enum import Enum
 
 
class StepStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    DONE = "done"
    FAILED = "failed"
    SKIPPED = "skipped"
 
 
@dataclass
class PlanStep:
    description: str
    tool_to_use: str
    expected_output: str
    status: StepStatus = StepStatus.PENDING
    result: str = ""
    error: str = ""
 
 
@dataclass
class AgentPlan:
    goal: str
    steps: list[PlanStep] = field(default_factory=list)
    current_step: int = 0
    completed: bool = False
 
    @property
    def current(self) -> Optional[PlanStep]:
        if self.current_step < len(self.steps):
            return self.steps[self.current_step]
        return None
 
    def advance(self):
        self.steps[self.current_step].status = StepStatus.DONE
        self.current_step += 1
        if self.current_step >= len(self.steps):
            self.completed = True
 
 
client = anthropic.Anthropic()
 
 
def create_plan(goal: str, available_tools: list[str]) -> AgentPlan:
    """Ask Claude to create a structured plan for achieving the goal."""
 
    prompt = (
        f"Create a step-by-step plan to achieve this goal:\n{goal}\n\n"
        f"Available tools: {', '.join(available_tools)}\n\n"
        "Return a JSON array of steps:\n"
        '[{"description": "what to do", "tool_to_use": "tool_name", '
        '"expected_output": "what success looks like"}]\n\n'
        "Keep it to 5 steps maximum. Be specific."
    )
 
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1000,
        messages=[{"role": "user", "content": prompt}]
    )
 
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1]
        if text.startswith("json"):
            text = text[4:]
 
    steps_data = json.loads(text)
    steps = [PlanStep(**s) for s in steps_data]
 
    return AgentPlan(goal=goal, steps=steps)
 
 
def execute_plan(plan: AgentPlan, tools: dict, memory: AgentMemory, max_iterations: int = 20) -> str:
    """Execute a plan step by step with Claude as the executor."""
    context = ContextManager()
 
    system = (
        "You are executing a structured plan. For each step, use the specified tool "
        "and verify the result matches the expected output. If a step fails, explain "
        "why and whether to retry, skip, or abort."
    )
 
    iterations = 0
 
    while not plan.completed and iterations < max_iterations:
        iterations += 1
        step = plan.current
 
        if not step:
            break
 
        step.status = StepStatus.IN_PROGRESS
 
        # Build prompt for this step
        memory_context = memory.get_relevant_context(step.description)
        message = (
            f"Execute step {plan.current_step + 1}/{len(plan.steps)}:\n"
            f"Goal: {step.description}\n"
            f"Tool to use: {step.tool_to_use}\n"
            f"Expected output: {step.expected_output}\n\n"
            f"Memory context:\n{memory_context}\n\n"
            "Call the tool and report the result."
        )
 
        context.add_message("user", message)
 
        # Let Claude call the tool
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=1000,
            system=system,
            tools=[{"name": name, "description": f"Execute {name}", "input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}} for name in tools.keys()],
            messages=context.get_messages()
        )
 
        context.add_message("assistant", response.content)
 
        # Execute tool calls
        tool_results = []
        for block in response.content:
            if block.type == "tool_use" and block.name in tools:
                result = tools[block.name](block.input.get("command", ""))
                memory.log_action(
                    action=f"{block.name}: {block.input}",
                    result=result,
                    success=True
                )
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })
 
        if tool_results:
            context.add_message("user", tool_results)
 
        plan.advance()
 
    return f"Plan executed: {plan.current_step}/{len(plan.steps)} steps completed"

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
 
tracer = trace.get_tracer("llm-agent")
 
 
def traced_agent_run(goal: str, agent_fn):
    """Wrap any agent function with OTel tracing."""
    with tracer.start_as_current_span("agent.run") as span:
        span.set_attribute("agent.goal", goal)
        span.set_attribute("agent.model", "claude-sonnet-5")
 
        start = time.time()
        try:
            result = agent_fn(goal)
            span.set_attribute("agent.success", True)
            span.set_attribute("agent.duration_ms", (time.time() - start) * 1000)
            return result
        except Exception as e:
            span.set_attribute("agent.success", False)
            span.set_attribute("agent.error", str(e))
            span.record_exception(e)
            raise