Implementation:Arize ai Phoenix Legacy BedrockModel

Overview

BedrockModel is a legacy model wrapper in the phoenix-evals package that provides an interface for using LLM models hosted on AWS Bedrock via the Converse API. It extends BaseModel and integrates with the boto3 AWS SDK, providing synchronous generation with rate limiting, context limit error handling for multiple model families (Anthropic, Titan, AI21), and executor-based async support. The wrapper constructs Bedrock Converse API request bodies with configurable inference parameters and model-specific top_k handling.

LLM_Evaluation Model_Integration

Description

The BedrockModel class is implemented as a Python dataclass that extends the abstract BaseModel. Key characteristics include:

• AWS Bedrock Converse API: Uses the unified converse() API endpoint rather than model-specific invoke endpoints, providing a consistent interface across different model providers.
• Flexible client initialization: Supports three modes -- providing a pre-built client, providing a session to create a client from, or auto-creating a client via boto3.client("bedrock-runtime").
• Dynamic rate limiting: Configures the RateLimiter to intercept ThrottlingException from the Bedrock client with an initial rate of 5 requests per second.
• Multi-provider context limit handling: Catches and translates context limit errors from Anthropic ("Input is too long"), Titan ("expected maxLength"), and AI21 ("Prompt has too many tokens") into PhoenixContextLimitExceeded.
• Executor-based async: Since boto3 does not support native async, the _async_generate_with_extra() method wraps the synchronous call using asyncio.get_event_loop().run_in_executor().
• Model-aware top_k support: The _model_supports_top_k() method excludes Meta Llama and Titan models from top_k configuration, and routes Amazon Nova models through a different parameter path (inferenceConfig.topK).
• Tool use extraction: Parses toolUse blocks from Converse API responses and serializes them as JSON.

Usage

# Configure your AWS credentials using the AWS CLI
# https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-files.html

from phoenix.evals.models import BedrockModel

# Basic usage with default model (anthropic.claude-v2)
model = BedrockModel()

# Use a specific model
model = BedrockModel(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    temperature=0.0,
    max_tokens=1024,
)

response = model("Explain the difference between SQL and NoSQL databases.")
print(response)

Code Reference

Source Location

Property	Value
Repository	Arize-ai/phoenix
File	packages/phoenix-evals/src/phoenix/evals/legacy/models/bedrock.py
Lines	252
Module	phoenix.evals.legacy.models.bedrock

Class Signature

@dataclass
class BedrockModel(BaseModel):
    model_id: str = "anthropic.claude-v2"
    temperature: float = 0.0
    max_tokens: int = 1024
    top_p: float = 1
    top_k: Optional[int] = None
    stop_sequences: List[str] = field(default_factory=list)
    session: Any = None
    client: Any = None
    max_content_size: Optional[int] = None
    extra_parameters: Dict[str, Any] = field(default_factory=dict)
    initial_rate_limit: int = 5
    timeout: int = 120

Constructor Parameters

Parameter	Type	Default	Description
model_id	str	"anthropic.claude-v2"	The Bedrock model identifier.
temperature	float	0.0	Sampling temperature for generation.
max_tokens	int	1024	Maximum number of tokens to generate.
top_p	float	1	Nucleus sampling probability mass.
top_k	Optional[int]	None	Top-K sampling cutoff (not supported by all models).
stop_sequences	List[str]	[]	Sequences that halt generation.
session	Any	None	A boto3 session to create the client from.
client	Any	None	A pre-built bedrock-runtime client.
max_content_size	Optional[int]	None	Maximum content size for fine-tuned models.
extra_parameters	Dict[str, Any]	{}	Extra parameters for the request body.
initial_rate_limit	int	5	Initial requests-per-second rate limit.
timeout	int	120	Timeout for API requests in seconds.

Key Methods

Method	Signature	Description
__post_init__	(self) -> None	Initializes the Bedrock client and rate limiter.
_init_client	(self) -> None	Creates the bedrock-runtime client from session, provided client, or boto3.
_init_rate_limiter	(self) -> None	Configures rate limiter with ThrottlingException.
_generate_with_extra	(self, prompt, **kwargs) -> Tuple[str, ExtraInfo]	Synchronous generation via the Converse API.
_async_generate_with_extra	async (self, prompt, **kwargs) -> Tuple[str, ExtraInfo]	Wraps sync generation in an executor for async compatibility.
_create_request_body	(self, prompt: MultimodalPrompt) -> Dict[str, Any]	Constructs the Converse API request body with inference config.
_rate_limited_completion	(self, **kwargs) -> Tuple[str, ExtraInfo]	Rate-limited wrapper around client.converse().
_extract_text	(self, response: ConverseResponseTypeDef) -> str	Extracts text or tool-use JSON from the Converse response.
_extract_usage	(self, response_usage: Optional[TokenUsageTypeDef]) -> Optional[Usage]	Extracts token usage from the response.
_model_supports_top_k	(self) -> bool	Returns whether the current model supports the top_k parameter.

Import

from phoenix.evals.models import BedrockModel

I/O Contract

Direction	Type	Description
Input	Union[str, MultimodalPrompt]	A text string or multimodal prompt (converted to text-only for Bedrock).
Input (optional)	Optional[str]	Instruction parameter (ignored; stripped before API call).
Output	str	Generated text response, or JSON-serialized tool use input.
Output (with extra)	Tuple[str, ExtraInfo]	Generated text paired with ExtraInfo containing optional Usage token counts.
Error	PhoenixContextLimitExceeded	Raised when prompt exceeds the model's context window (Anthropic, Titan, or AI21 error patterns).
Error	ImportError	Raised if boto3 package is not installed.

Usage Examples

Using a Custom Session

import boto3
from phoenix.evals.models import BedrockModel

session = boto3.Session(profile_name="my-aws-profile", region_name="us-east-1")
model = BedrockModel(
    model_id="anthropic.claude-3-sonnet-20240229-v1:0",
    session=session,
)

response = model("What is serverless computing?")
print(response)

With Amazon Nova Model

from phoenix.evals.models import BedrockModel

model = BedrockModel(
    model_id="amazon.nova-pro-v1:0",
    top_k=50,
    temperature=0.7,
)
# top_k is routed to additionalModelRequestFields.inferenceConfig.topK for Nova models

With Extra Parameters

from phoenix.evals.models import BedrockModel

model = BedrockModel(
    model_id="ai21.j2-ultra-v1",
    extra_parameters={"countPenalty": {"scale": 0.5}},
)

Related Pages

• Implementation:Arize_ai_Phoenix_Legacy_BaseModel - Abstract base class that BedrockModel extends
• Implementation:Arize_ai_Phoenix_Legacy_AnthropicModel - Direct Anthropic API wrapper (alternative to using Anthropic via Bedrock)
• Implementation:Arize_ai_Phoenix_Legacy_OpenAIModel - OpenAI model wrapper (similar pattern)
• Implementation:Arize_ai_Phoenix_Legacy_LiteLLMModel - LiteLLM wrapper that also supports Bedrock models via its universal interface