​
Overview

AnthropicLLMService provides integration with Anthropic’s Claude models, supporting streaming responses, function calling, and prompt caching. It includes specialized context handling for Anthropic’s message format.

​
Installation

To use AnthropicLLMService, install the required dependencies:

pip install "pipecat-ai[anthropic]"

You’ll also need to set up your Anthropic API key as an environment variable: ANTHROPIC_API_KEY

​
Configuration

​
Constructor Parameters

​
api_key
str
required

Anthropic API key

​
model
str
default:"claude-sonnet-4-20250514"

Model identifier

​
params
InputParams

Model configuration parameters

​
Input Parameters

​
enable_prompt_caching_beta
Optional[bool]
default:"False"

Enables beta prompt caching functionality

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

Additional parameters to pass to the model

​
max_tokens
Optional[int]
default:"4096"

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

​
temperature
Optional[float]
default:"NOT_GIVEN"

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

​
top_k
Optional[int]
default:"NOT_GIVEN"

Controls diversity via nucleus sampling. Must be greater than or equal to 0

​
top_p
Optional[float]
default:"NOT_GIVEN"

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

​
Input Frames

​
OpenAILLMContextFrame
Frame

Contains conversation context

​
LLMMessagesFrame
Frame

Contains conversation messages

​
VisionImageRawFrame
Frame

Contains image for vision processing

​
LLMUpdateSettingsFrame
Frame

Updates model settings

​
LLMEnablePromptCachingFrame
Frame

Controls prompt caching behavior

​
Output Frames

​
TextFrame
Frame

Contains generated text

​
FunctionCallInProgressFrame
Frame

Indicates ongoing function call

​
FunctionCallResultFrame
Frame

Contains function call results

​
Context Management

The Anthropic 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 Anthropic message formats.

​
AnthropicLLMContext

The base context manager for Anthropic conversations:

context = AnthropicLLMContext(
    messages=[],  # Conversation history
    tools=[],     # Available function calling 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
static method

Creates user and assistant aggregators for handling message formatting.

@staticmethod
def create_context_aggregator(
    context: OpenAILLMContext,
    *,
    assistant_expect_stripped_words: bool = True
) -> AnthropicContextAggregatorPair

​
Parameters

​
context
OpenAILLMContext
required

The context object containing conversation history and settings

​
assistant_expect_stripped_words
bool
default:"True"

Controls text preprocessing for assistant responses

​
Usage Example


# 1. Create the context
context = AnthropicLLMContext(
    messages=[],
    system="You are a helpful assistant"
)

# 2. Create aggregators for message handling
aggregators = AnthropicLLMService.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
])

The context management system ensures proper message formatting and history tracking throughout the conversation while handling the conversion between OpenAI and Anthropic message formats automatically.

​
Methods

See the LLM base class methods for additional functionality.

​
Function Calling

This service supports function calling (also known as tool calling) which allows the LLM to request information from external services and APIs. For example, you can enable your bot to:

  • Check current weather conditions
  • Query databases
  • Access external APIs
  • Perform custom actions

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.

​
Usage Examples

​
Basic Usage

# Configure service
llm = AnthropicLLMService(
    api_key="your-api-key",
    model="claude-3-5-sonnet-20240620",
    params=AnthropicLLMService.InputParams(
        temperature=0.7,
        max_tokens=1000
    )
)

# Create pipeline
pipeline = Pipeline([
    transport.input(),
    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
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext

messages = [{"role": "user", "content": "Say 'hello' to start the conversation."}]

# Define the weather function using standardized schema
weather_function = FunctionSchema(
    name="get_weather",
    description="Get weather information",
    properties={
        "location": {"type": "string"}
    },
    required=["location"]  # Specify required parameters
)

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

# Configure function calling
context = OpenAILLMContext(
    messages=messages,
    tools=tools
)

​
Use in pipeline

pipeline = Pipeline([ context_aggregator, llm, function_handler ])


## Frame Flow

```mermaid
graph TD
    A[Input Context] --> B[AnthropicLLMService]
    B --> C[LLMFullResponseStartFrame]
    B --> D[TextFrame Chunks]
    B --> E[Function Calls]
    B --> F[LLMFullResponseEndFrame]
    E --> G[Function Results]
    G --> B

​
Metrics Support

The service collects various metrics:

  • Token usage (prompt and completion)
  • Cache metrics (creation and read tokens)
  • Processing time
  • Time to first byte (TTFB)

​
Features

​
Prompt Caching

# Enable prompt caching
await pipeline.push_frame(
    LLMEnablePromptCachingFrame(enable=True)
)

​
Message Format Conversion

Automatically handles conversion between:

  • OpenAI message format
  • Anthropic message format
  • Function calling format

​
Token Estimation

Provides token usage tracking and estimation:

def _estimate_tokens(self, text: str) -> int:
    """Estimates token count for partial responses"""

​
Notes

  • Supports streaming responses
  • Handles function calling
  • Provides prompt caching
  • Manages conversation context
  • Supports vision inputs
  • Includes metrics collection
  • Thread-safe processing
  • Handles interruptions