Custom Agent Architecture

123456789101112
import boto3
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum

class AgentState(Enum):
    OBSERVING = "observing"
    REASONING = "reasoning"
    ACTING = "acting"
    COMPLETED = "completed"
    ERROR = "error"

123456789101112
import boto3
import json
from dataclasses import dataclass, asdict
from typing import List, Dict, Any, Optional
from enum import Enum
import time

class AgentState(Enum):
    PLANNING = "planning"
    EXECUTING = "executing"
    REFLECTING = "reflecting"
    COMPLETE = "complete"

123456789101112
from langchain_aws import ChatBedrock
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain_community.callbacks import get_openai_callback
import boto3

# Initialize Bedrock client
llm = ChatBedrock(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    model_kwargs={"temperature": 0.1, "max_tokens": 4096},
    region_name="us-east-1"

123456789101112
import boto3
import json
from datetime import datetime
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, asdict
from enum import Enum
import anthropic

class AgentState(Enum):
    PLANNING = "planning"
    EXECUTING = "executing"
    WAITING_FOR_TOOL = "waiting_for_tool"

123456789101112
from abc import ABC, abstractmethod
from typing import Dict, Any, Callable, Optional
from dataclasses import dataclass
import json

@dataclass
class ToolDefinition:
    name: str
    description: str
    parameters: Dict[str, Any]
    handler: Callable
    timeout_seconds: int = 30

123456789101112
class CheckpointManager:
    def __init__(self, session_id: str):
        self.dynamodb = boto3.resource('dynamodb')
        self.table = self.dynamodb.Table('agent-checkpoints')
        self.session_id = session_id
    
    def save_checkpoint(self, step: int, state: Dict[str, Any]):
        self.table.put_item(Item={
            'session_id': self.session_id,
            'checkpoint_id': f"{self.session_id}#{step}",
            'step': step,
            'state': json.dumps(state),