wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Anthropic docstrings

Browse Source
This commit is contained in:
Mark Backman
2025-06-26 07:42:22 -04:00
parent 1cd42066a6
commit 990ee436e1
1 changed files with 210 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

226
src/pipecat/services/anthropic/llm.py
View File

@@ -4,6 +4,12 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Anthropic AI service integration for Pipecat.
This module provides LLM services and context management for Anthropic's Claude models,
including support for function calling, vision, and prompt caching features.
"""
import asyncio
import base64
import copy
@@ -59,27 +65,66 @@ except ModuleNotFoundError as e:
@dataclass
class AnthropicContextAggregatorPair:
"""Pair of context aggregators for Anthropic conversations.
Encapsulates both user and assistant context aggregators
to manage conversation flow and message formatting.
Parameters:
_user: The user context aggregator.
_assistant: The assistant context aggregator.
"""
_user: "AnthropicUserContextAggregator"
_assistant: "AnthropicAssistantContextAggregator"
def user(self) -> "AnthropicUserContextAggregator":
"""Get the user context aggregator.
Returns:
The user context aggregator instance.
"""
return self._user
def assistant(self) -> "AnthropicAssistantContextAggregator":
"""Get the assistant context aggregator.
Returns:
The assistant context aggregator instance.
"""
return self._assistant
class AnthropicLLMService(LLMService):
"""This class implements inference with Anthropic's AI models.
"""LLM service for Anthropic's Claude models.
Can provide a custom client via the `client` kwarg, allowing you to
use `AsyncAnthropicBedrock` and `AsyncAnthropicVertex` clients
Provides inference capabilities with Claude models including support for
function calling, vision processing, streaming responses, and prompt caching.
Can use custom clients like AsyncAnthropicBedrock and AsyncAnthropicVertex.
Args:
api_key: Anthropic API key for authentication.
model: Model name to use. Defaults to "claude-sonnet-4-20250514".
params: Optional model parameters for inference.
client: Optional custom Anthropic client instance.
**kwargs: Additional arguments passed to parent LLMService.
"""
# Overriding the default adapter to use the Anthropic one.
adapter_class = AnthropicLLMAdapter
class InputParams(BaseModel):
"""Input parameters for Anthropic model inference.
Parameters:
enable_prompt_caching_beta: Whether to enable beta prompt caching feature.
max_tokens: Maximum tokens to generate. Must be at least 1.
temperature: Sampling temperature between 0.0 and 1.0.
top_k: Top-k sampling parameter.
top_p: Top-p sampling parameter between 0.0 and 1.0.
extra: Additional parameters to pass to the API.
"""
enable_prompt_caching_beta: Optional[bool] = False
max_tokens: Optional[int] = Field(default_factory=lambda: 4096, ge=1)
temperature: Optional[float] = Field(default_factory=lambda: NOT_GIVEN, ge=0.0, le=1.0)
@@ -112,10 +157,20 @@ class AnthropicLLMService(LLMService):
}
def can_generate_metrics(self) -> bool:
"""Check if this service can generate usage metrics.
Returns:
True, as Anthropic provides detailed token usage metrics.
"""
return True
@property
def enable_prompt_caching_beta(self) -> bool:
"""Check if prompt caching beta feature is enabled.
Returns:
True if prompt caching is enabled.
"""
return self._enable_prompt_caching_beta
def create_context_aggregator(
@@ -125,22 +180,19 @@ class AnthropicLLMService(LLMService):
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> AnthropicContextAggregatorPair:
"""Create an instance of AnthropicContextAggregatorPair from an
OpenAILLMContext. Constructor keyword arguments for both the user and
assistant aggregators can be provided.
"""Create Anthropic-specific context aggregators.
Creates a pair of context aggregators optimized for Anthropic's message format,
including support for function calls, tool usage, and image handling.
Args:
context (OpenAILLMContext): The LLM context.
user_params (LLMUserAggregatorParams, optional): User aggregator
parameters.
assistant_params (LLMAssistantAggregatorParams, optional): User
aggregator parameters.
context: The LLM context.
user_params: User aggregator parameters.
assistant_params: Assistant aggregator parameters.
Returns:
AnthropicContextAggregatorPair: A pair of context aggregators, one
for the user and one for the assistant, encapsulated in an
AnthropicContextAggregatorPair.
A pair of context aggregators, one for the user and one for the assistant,
encapsulated in an AnthropicContextAggregatorPair.
"""
context.set_llm_adapter(self.get_llm_adapter())
@@ -310,6 +362,15 @@ class AnthropicLLMService(LLMService):
)
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process incoming frames and route them appropriately.
Handles various frame types including context frames, message frames,
vision frames, and settings updates.
Args:
frame: The frame to process.
direction: The direction of frame processing.
"""
await super().process_frame(frame, direction)
context = None
@@ -361,6 +422,19 @@ class AnthropicLLMService(LLMService):
class AnthropicLLMContext(OpenAILLMContext):
"""LLM context specialized for Anthropic's message format and features.
Extends OpenAILLMContext to handle Anthropic-specific features like
system messages, prompt caching, and message format conversions.
Manages conversation state and message history formatting.
Args:
messages: Initial list of conversation messages.
tools: Available function calling tools.
tool_choice: Tool selection preference.
system: System message content.
"""
def __init__(
self,
messages: Optional[List[dict]] = None,
@@ -381,6 +455,16 @@ class AnthropicLLMContext(OpenAILLMContext):
@staticmethod
def upgrade_to_anthropic(obj: OpenAILLMContext) -> "AnthropicLLMContext":
"""Upgrade an OpenAI context to Anthropic format.
Converts message format and restructures content for Anthropic compatibility.
Args:
obj: The OpenAI context to upgrade.
Returns:
The upgraded Anthropic context.
"""
logger.debug(f"Upgrading to Anthropic: {obj}")
if isinstance(obj, OpenAILLMContext) and not isinstance(obj, AnthropicLLMContext):
obj.__class__ = AnthropicLLMContext
@@ -389,6 +473,14 @@ class AnthropicLLMContext(OpenAILLMContext):
@classmethod
def from_openai_context(cls, openai_context: OpenAILLMContext):
"""Create Anthropic context from OpenAI context.
Args:
openai_context: The OpenAI context to convert.
Returns:
New Anthropic context with converted messages.
"""
self = cls(
messages=openai_context.messages,
tools=openai_context.tools,
@@ -400,12 +492,28 @@ class AnthropicLLMContext(OpenAILLMContext):
@classmethod
def from_messages(cls, messages: List[dict]) -> "AnthropicLLMContext":
"""Create context from a list of messages.
Args:
messages: List of conversation messages.
Returns:
New Anthropic context with the provided messages.
"""
self = cls(messages=messages)
self._restructure_from_openai_messages()
return self
@classmethod
def from_image_frame(cls, frame: VisionImageRawFrame) -> "AnthropicLLMContext":
"""Create context from a vision image frame.
Args:
frame: The vision image frame to process.
Returns:
New Anthropic context with the image message.
"""
context = cls()
context.add_image_frame_message(
format=frame.format, size=frame.size, image=frame.image, text=frame.text
@@ -413,11 +521,15 @@ class AnthropicLLMContext(OpenAILLMContext):
return context
def set_messages(self, messages: List):
"""Set the messages list and reset cache tracking.
Args:
messages: New list of messages to set.
"""
self.turns_above_cache_threshold = 0
self._messages[:] = messages
self._restructure_from_openai_messages()
# convert a message in Anthropic format into one or more messages in OpenAI format
def to_standard_messages(self, obj):
"""Convert Anthropic message format to standard structured format.
@@ -558,6 +670,17 @@ class AnthropicLLMContext(OpenAILLMContext):
def add_image_frame_message(
self, *, format: str, size: tuple[int, int], image: bytes, text: str = None
):
"""Add an image message to the context.
Converts the image to base64 JPEG format and adds it as a user message
with optional accompanying text.
Args:
format: The image format (e.g., 'RGB', 'RGBA').
size: Image dimensions as (width, height).
image: Raw image bytes.
text: Optional text to accompany the image.
"""
buffer = io.BytesIO()
Image.frombytes(format, size, image).save(buffer, format="JPEG")
encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
@@ -578,6 +701,14 @@ class AnthropicLLMContext(OpenAILLMContext):
self.add_message({"role": "user", "content": content})
def add_message(self, message):
"""Add a message to the context, merging with previous message if same role.
Anthropic requires alternating roles, so consecutive messages from the same
role are merged together.
Args:
message: The message to add to the context.
"""
try:
if self.messages:
# Anthropic requires that roles alternate. If this message's role is the same as the
@@ -603,6 +734,14 @@ class AnthropicLLMContext(OpenAILLMContext):
logger.error(f"Error adding message: {e}")
def get_messages_with_cache_control_markers(self) -> List[dict]:
"""Get messages with prompt caching markers applied.
Adds cache control markers to appropriate messages based on the
number of turns above the cache threshold.
Returns:
List of messages with cache control markers added.
"""
try:
messages = copy.deepcopy(self.messages)
if self.turns_above_cache_threshold >= 1 and messages[-1]["role"] == "user":
@@ -670,12 +809,26 @@ class AnthropicLLMContext(OpenAILLMContext):
message["content"] = [{"type": "text", "text": "(empty)"}]
def get_messages_for_persistent_storage(self):
"""Get messages formatted for persistent storage.
Includes system message at the beginning if present.
Returns:
List of messages suitable for storage.
"""
messages = super().get_messages_for_persistent_storage()
if self.system:
messages.insert(0, {"role": "system", "content": self.system})
return messages
def get_messages_for_logging(self) -> str:
"""Get messages formatted for logging with sensitive data redacted.
Replaces image data with placeholder text for cleaner logs.
Returns:
JSON string representation of messages for logging.
"""
msgs = []
for message in self.messages:
msg = copy.deepcopy(message)
@@ -689,6 +842,12 @@ class AnthropicLLMContext(OpenAILLMContext):
class AnthropicUserContextAggregator(LLMUserContextAggregator):
"""Anthropic-specific user context aggregator.
Handles aggregation of user messages for Anthropic LLM services.
Inherits all functionality from the base LLMUserContextAggregator.
"""
pass
@@ -703,7 +862,20 @@ class AnthropicUserContextAggregator(LLMUserContextAggregator):
class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
"""Context aggregator for assistant messages in Anthropic conversations.
Handles function call lifecycle management including in-progress tracking,
result handling, and cancellation for Anthropic's tool use format.
"""
async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
"""Handle a function call that is starting.
Creates tool use message and placeholder tool result for tracking.
Args:
frame: Frame containing function call details.
"""
assistant_message = {"role": "assistant", "content": []}
assistant_message["content"].append(
{
@@ -728,6 +900,13 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
)
async def handle_function_call_result(self, frame: FunctionCallResultFrame):
"""Handle the result of a completed function call.
Updates the tool result with actual return value or completion status.
Args:
frame: Frame containing function call result.
"""
if frame.result:
result = json.dumps(frame.result)
await self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
@@ -737,6 +916,13 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
)
async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
"""Handle cancellation of a function call.
Updates the tool result to indicate cancellation.
Args:
frame: Frame containing function call cancellation details.
"""
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, "CANCELLED"
)
@@ -755,6 +941,14 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
content["content"] = result
async def handle_user_image_frame(self, frame: UserImageRawFrame):
"""Handle a user image frame with function call context.
Marks the associated function call as completed and adds the image
to the conversation context.
Args:
frame: User image frame with request context.
"""
await self._update_function_call_result(
frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
)