Claude Model

The ClaudeModel class implements the interface for Amazon's Claude language models, providing access to various Claude model versions through Amazon Bedrock.

Class Documentation

bedrock_swarm.models.claude.ClaudeModel(model_id: str)

Bases: BedrockModel

Implementation for Claude 3.5 models.

Source code in src/bedrock_swarm/models/base.py

def __init__(self, model_id: str):
    """Initialize the model.

    Args:
        model_id: The Bedrock model ID to use
    """
    self._model_id = model_id
    self._config: Dict[str, Any] = {
        "max_tokens": 4096,  # Default maximum tokens
        "default_tokens": 2048,  # Default response length
    }

Functions

get_model_id() -> str

Get the Bedrock model ID.

Source code in src/bedrock_swarm/models/claude.py

def get_model_id(self) -> str:
    """Get the Bedrock model ID."""
    return "us.anthropic.claude-3-5-sonnet-20241022-v2:0"

format_request(message: str, system: Optional[str] = None, temperature: float = 0.7, max_tokens: Optional[int] = None) -> Dict[str, Any]

Format a request for Claude.

PARAMETER	DESCRIPTION
message	The message to send to the model TYPE: str
system	Optional system prompt TYPE: Optional[str] DEFAULT: None
temperature	Temperature for response generation (0.0 to 1.0) TYPE: float DEFAULT: 0.7
max_tokens	Maximum number of tokens to generate TYPE: Optional[int] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	Formatted request dictionary

Source code in src/bedrock_swarm/models/claude.py

def format_request(
    self,
    message: str,
    system: Optional[str] = None,
    temperature: float = 0.7,
    max_tokens: Optional[int] = None,
) -> Dict[str, Any]:
    """Format a request for Claude.

    Args:
        message: The message to send to the model
        system: Optional system prompt
        temperature: Temperature for response generation (0.0 to 1.0)
        max_tokens: Maximum number of tokens to generate

    Returns:
        Formatted request dictionary
    """
    # Combine system prompt and message if provided
    content = f"{system}\n\n{message}" if system else message

    return {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": max_tokens or 4096,
        "temperature": temperature,
        "messages": [{"role": "user", "content": content}],
    }

_extract_content(response: Dict[str, Any]) -> str

Extract content from Claude response.

PARAMETER	DESCRIPTION
response	Raw response from Claude TYPE: Dict[str, Any]

RETURNS	DESCRIPTION
str	Extracted content as string

RAISES	DESCRIPTION
ResponseParsingError	If content cannot be extracted

Source code in src/bedrock_swarm/models/claude.py

def _extract_content(self, response: Dict[str, Any]) -> str:
    """Extract content from Claude response.

    Args:
        response: Raw response from Claude

    Returns:
        Extracted content as string

    Raises:
        ResponseParsingError: If content cannot be extracted
    """
    content = []
    for event in response["body"]:
        try:
            chunk = json.loads(event.get("chunk").get("bytes").decode())
            if chunk.get("type") == "content_block_delta":
                content.append(chunk["delta"]["text"])
        except json.JSONDecodeError as e:
            raise ResponseParsingError(f"Error parsing chunk: {str(e)}")
        except (KeyError, AttributeError) as e:
            raise ResponseParsingError(f"Invalid chunk format: {str(e)}")

    return "".join(content)

Model Variants

The following Claude model variants are supported:

• Claude 3.5 Sonnet: Latest general purpose model with strong performance across tasks
• Model ID: us.anthropic.claude-3-5-sonnet-20241022-v2:0
• Best for: Complex reasoning, analysis, and generation tasks

Request Format

request = {
    "anthropic_version": "bedrock-2023-05-31",
    "max_tokens": 4096,  # Optional, defaults to 4096
    "temperature": 0.7,  # Optional, defaults to 0.7
    "messages": [
        {
            "role": "user",
            "content": "Your message here"
        }
    ]
}

Response Format

The model returns responses in a streaming format with content blocks:

async for chunk in model.invoke(request):
    # Each chunk contains a content block delta
    print(chunk)  # Process each chunk as it arrives

Error Handling

The Claude model implementation includes comprehensive error handling for:

1. Token validation (enforces model-specific limits)
2. Temperature validation (must be between 0 and 1)
3. Response parsing (handles various response formats)
4. API errors (with automatic retries)

Usage Example

from bedrock_swarm.models import ClaudeModel

model = ClaudeModel()

request = model.format_request(
    message="Explain quantum computing in simple terms.",
    temperature=0.7,
    max_tokens=1000
)

async for chunk in model.invoke(request):
    print(chunk)

Implementation Details

The Claude model implementation includes:

1. Request formatting according to Claude's specifications
2. Response streaming with chunk processing
3. Automatic retry logic for transient errors
4. Comprehensive error handling
5. Token limit enforcement

See Also

• Base Model - Base model implementation details
• Model Factory - Model creation and configuration
• Titan Model - Titan model implementation
• Models Overview - Overview of model system