​
Overview

AWSBedrockLLMService provides integration with Amazon Bedrock models, including Anthropic Claude and Amazon Titan models. It supports streaming responses, function calling, and multimodal inputs with specialized context handling for AWS Bedrock’s message format while maintaining compatibility with OpenAI-style contexts.

​
Installation

To use AWSBedrockLLMService, install the required dependencies:

pip install "pipecat-ai[aws]"

You’ll also need to set up your AWS credentials as environment variables:

  • AWS_ACCESS_KEY_ID
  • AWS_SECRET_ACCESS_KEY
  • AWS_SESSION_TOKEN (if using temporary credentials)
  • AWS_REGION (defaults to “us-east-1”)

You can obtain AWS credentials by setting up an IAM user with access to Amazon Bedrock in your AWS account.

​
Configuration

​
Constructor Parameters

​
aws_access_key
str

Your AWS access key ID (can also use environment variable)

​
aws_secret_key
str

Your AWS secret access key (can also use environment variable)

​
aws_session_token
str

Your AWS session token for temporary credentials (can also use environment variable)

​
aws_region
str
default:"us-east-1"

AWS region to use for Bedrock service

​
model
str
required

Bedrock model identifier (e.g., “us.anthropic.claude-3-5-haiku-20241022-v1:0”)

​
params
InputParams

Model configuration parameters

​
client_config
Config

AWS Boto3 client configuration

​
Input Parameters

​
max_tokens
Optional[int]
default:"4096"

Maximum number of tokens to generate. Must be greater than or equal to 1.

​
temperature
Optional[float]
default:"0.7"

Controls randomness in the output. Range: [0.0, 1.0].

​
top_p
Optional[float]
default:"0.999"

Controls diversity via nucleus sampling. Range: [0.0, 1.0].

​
stop_sequences
Optional[List[str]]
default:"[]"

Sequences where the model will stop generating further tokens.

​
latency
Optional[str]
default:"standard"

Response latency preference. Options: “standard” or “optimized”.

​
additional_model_request_fields
Optional[Dict[str, Any]]
default:"{}"

Additional parameters to pass to the model.

​
Input Frames

​
OpenAILLMContextFrame
Frame

Contains conversation context

​
LLMMessagesFrame
Frame

Contains conversation messages

​
VisionImageRawFrame
Frame

Contains image for vision processing

​
LLMUpdateSettingsFrame
Frame

Updates model settings

​
Output Frames

​
LLMTextFrame
Frame

Contains generated text

​
LLMFullResponseStartFrame
Frame

Signals start of response

​
LLMFullResponseEndFrame
Frame

Signals end of response

​
FunctionCallInProgressFrame
Frame

Indicates function call initiation

​
FunctionCallResultFrame
Frame

Contains function call results

​
FunctionCallCancelFrame
Frame

Indicates function call cancellation

​
Context Management

The AWS Bedrock service uses specialized context management to handle conversations and message formatting. This includes managing the conversation history, system prompts, function calls, and converting between OpenAI and AWS Bedrock message formats.

​
AWSBedrockLLMContext

The base context manager for AWS Bedrock conversations:

context = AWSBedrockLLMContext(
    messages=[],       # Conversation history
    tools=[],          # Available function calling tools
    tool_choice={},    # How the model should use the provided tools
    system="You are a helpful assistant"  # System prompt
)

​
Context Aggregators

Context aggregators handle message format conversion and management. The service provides a method to create paired aggregators:

​
create_context_aggregator
method

Creates user and assistant aggregators for handling message formatting.

def create_context_aggregator(
    context: OpenAILLMContext,
    *,
    user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
    assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> AWSBedrockContextAggregatorPair

​
Parameters

​
context
OpenAILLMContext
required

The context object containing conversation history and settings

​
user_params
LLMUserAggregatorParams

Parameters for user message handling

​
assistant_params
LLMAssistantAggregatorParams

Parameters for assistant message handling

​
Usage Example

# 1. Create the context
context = OpenAILLMContext(
    messages=[
        {"role": "system", "content": "You are a helpful assistant"}
    ]
)

# 2. Create aggregators for message handling
aggregators = llm.create_context_aggregator(context)

# 3. Access individual aggregators
user_aggregator = aggregators.user()      # Handles user message formatting
assistant_aggregator = aggregators.assistant()  # Handles assistant responses

# 4. Use in a pipeline
pipeline = Pipeline([
    user_aggregator,
    llm,
    assistant_aggregator
])

​
Message Format Conversion

AWSBedrockLLMContext automatically handles the conversion between OpenAI-style messages and AWS Bedrock’s format:

  • System messages are handled separately in AWS Bedrock
  • User and assistant messages are converted to AWS Bedrock’s format
  • Function calls are mapped to AWS Bedrock’s “toolUse” format
  • Function results are mapped to AWS Bedrock’s “toolResult” format

​
Methods

See the LLM base class methods for additional functionality.

​
Function Calling

AWS Bedrock LLM service supports function calling (tool calling) which allows the LLM to request information from external services and APIs. The service automatically handles the conversion between OpenAI-style function calls and AWS Bedrock’s tool format.

Function Calling Guide

Learn how to implement function calling with standardized schemas, register handlers, manage context properly, and control execution flow in your conversational AI applications.

​
Function Call Example

# Define the function schema
weather_function = FunctionSchema(
    name="get_current_weather",
    description="Get the current weather",
    properties={
        "location": {
            "type": "string",
            "description": "The city and state, e.g. San Francisco, CA",
        },
        "format": {
            "type": "string",
            "enum": ["celsius", "fahrenheit"],
            "description": "The temperature unit to use.",
        },
    },
    required=["location", "format"],
)

# Create tools schema
tools = ToolsSchema(standard_tools=[weather_function])

# Register function handler
async def fetch_weather_from_api(params: FunctionCallParams):
    await params.result_callback({"conditions": "sunny", "temperature": "75"})

llm.register_function("get_current_weather", fetch_weather_from_api)

# Create context with tools
context = OpenAILLMContext(messages, tools)

​
Multimodal Support

AWSBedrockLLMService supports processing images with compatible models (such as Claude 3):

# Handle image in context
context.add_image_frame_message(
    format="RGB",
    size=(800, 600),
    image=image_bytes,
    text="What can you see in this image?"
)

The service automatically handles the base64 encoding and proper formatting required by AWS Bedrock for image processing.

​
Usage Examples

​
Basic Usage

from pipecat.services.aws.llm import AWSBedrockLLMService

# Configure service using environment variables for credentials
llm = AWSBedrockLLMService(
    aws_region="us-west-2",
    model="us.anthropic.claude-3-5-haiku-20241022-v1:0",
    params=AWSBedrockLLMService.InputParams(
        temperature=0.7,
        max_tokens=1000,
        latency="optimized"
    )
)

# Create context and aggregators
messages = [
    {"role": "system", "content": "You are a helpful assistant"}
]
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)

# Create pipeline
pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    llm,
    tts,
    transport.output(),
    context_aggregator.assistant()
])

​
With Function Calling

from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema

# Define function schema
weather_function = FunctionSchema(
    name="get_current_weather",
    description="Get the current weather",
    properties={
        "location": {
            "type": "string",
            "description": "The city and state, e.g. San Francisco, CA",
        },
        "format": {
            "type": "string",
            "enum": ["celsius", "fahrenheit"],
            "description": "The temperature unit to use.",
        },
    },
    required=["location", "format"],
)

# Create tools schema
tools = ToolsSchema(standard_tools=[weather_function])

# Register function handler
async def fetch_weather_from_api(params: FunctionCallParams):
    await params.result_callback({"conditions": "sunny", "temperature": "75"})

llm.register_function("get_current_weather", fetch_weather_from_api)

# Create context with tools
context = OpenAILLMContext(messages, tools)
context_aggregator = llm.create_context_aggregator(context)

​
Frame Flow

​
Metrics Support

The service collects various metrics:

  • Token usage (prompt and completion)
  • Processing time
  • Time to first byte (TTFB)
  • Cache utilization (if available)

​
Notes

  • Supports streaming responses
  • Handles function calling with AWS Bedrock’s “toolUse” format
  • Provides OpenAI compatibility layer
  • Manages conversation context with proper format conversion
  • Supports vision inputs for compatible models
  • Includes metrics collection
  • Thread-safe processing
  • Handles reconnection and error recovery