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

Anthropic tool use core Pipecat pieces refactored (#369)

Browse Source 
* processors(rtvi): rtvi 0.1 message protocol

* added a single function call handler

* wip - function calling

* fixup

* fixup

* fixup

* processors(rtvi): no need for configure_on_start()

* processors(rtvi): add new option values if they haven't been set yet

* Add the model name to the LLM usage metrics

* wip - anthropic tool calling

* still wip - anthropic tool use and vision

* anthropic tools and vision working

* anthropic tool calling and vision

* Cartesia error handling

* Anthropic tool use core Pipecat pieces refactored as per plan

* aleix has good ideas

* Usage metrics for Anthropic LLMs

* fix function call result state not getting cleared bug

* Pass **kwargs through from AnthropicLLMService constructor

* about to tinker with anthropic

* added openai function calling

* openai function calling

* fixup

---------

Co-authored-by: Aleix Conchillo Flaqué <aleix@daily.co>
Co-authored-by: Chad Bailey <chadbailey@gmail.com>
Co-authored-by: mattie ruth backman <mattieruth@gmail.com>
Co-authored-by: chadbailey59 <chadbailey59@users.noreply.github.com>
This commit is contained in:
Kwindla Hultman Kramer
2024-08-13 11:01:24 -07:00
committed by GitHub
parent a42d0c9907
commit 29ca1b7855
13 changed files with 1009 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/pipecat/frames/frames.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
from typing import Any, List, Mapping, Tuple
from typing import Any, List, Mapping, Tuple, Optional
from dataclasses import dataclass, field
@@ -177,6 +177,15 @@ class LLMMessagesUpdateFrame(DataFrame):
messages: List[dict]
@dataclass
class LLMSetToolsFrame(DataFrame):
"""A frame containing a list of tools for an LLM to use for function calling.
The specific format depends on the LLM being used, but it should typically
contain JSON Schema objects.
"""
tools: List[dict]
@dataclass
class TTSSpeakFrame(DataFrame):
"""A frame that contains a text that should be spoken by the TTS in the
@@ -389,6 +398,7 @@ class TTSStoppedFrame(ControlFrame):
class UserImageRequestFrame(ControlFrame):
"""A frame user to request an image from the given user."""
user_id: str
context: Optional[any]
def __str__(self):
return f"{self.name}, user: {self.user_id}"
@@ -406,3 +416,22 @@ class TTSVoiceUpdateFrame(ControlFrame):
"""A control frame containing a request to update to a new TTS voice.
"""
voice: str
@dataclass
class FunctionCallInProgressFrame(SystemFrame):
"""A frame signaling that a function call is in progress.
"""
function_name: str
tool_call_id: str
arguments: str
@dataclass
class FunctionCallResultFrame(DataFrame):
"""A frame containing the result of an LLM function (tool) call.
"""
function_name: str
tool_call_id: str
arguments: str
result: any

41
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -4,9 +4,10 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import sys
from typing import List
from pipecat.services.openai import OpenAILLMContextFrame, OpenAILLMContext
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame, OpenAILLMContext
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.frames.frames import (
@@ -17,6 +18,7 @@ from pipecat.frames.frames import (
LLMMessagesAppendFrame,
LLMMessagesFrame,
LLMMessagesUpdateFrame,
LLMSetToolsFrame,
StartInterruptionFrame,
TranscriptionFrame,
TextFrame,
@@ -80,6 +82,10 @@ class LLMResponseAggregator(FrameProcessor):
#
# and T2 would be dropped.
async def _set_tools(self, tools: List):
# noop in the base class
pass
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -129,17 +135,28 @@ class LLMResponseAggregator(FrameProcessor):
elif isinstance(frame, LLMMessagesUpdateFrame):
# We push the frame downstream so the assistant aggregator gets
# updated as well.
await self.push_frame(frame)
# TODO-CB: Now we're replacing the contents of the array so we
# don't need to push the frame here
# await self.push_frame(frame)
# We can now reset this one.
self._reset()
self._messages = frame.messages
messages_frame = LLMMessagesFrame(self._messages)
await self.push_frame(messages_frame)
self._set_messages(frame.messages)
# messages_frame = LLMMessagesFrame(self._messages)
# await self.push_frame(messages_frame)
await self.push_messages_frame()
elif isinstance(frame, LLMSetToolsFrame):
await self.push_frame(frame)
await self._set_tools(frame.tools)
else:
await self.push_frame(frame, direction)
if send_aggregation:
await self._push_aggregation()
# TODO-CB: Types
def _set_messages(self, messages):
self._messages.clear()
self._messages.extend(messages)
async def _push_aggregation(self):
if len(self._aggregation) > 0:
@@ -243,6 +260,20 @@ class LLMContextAggregator(LLMResponseAggregator):
self._context = context
super().__init__(**kwargs)
# TODO-CB: thanks, I hate it
self._messages = context.messages
async def _set_tools(self, tools: List):
# We push the frame downstream so the assistant aggregator gets
# updated as well.
self._context.tools = tools
# TODO-CB: Types
def _set_messages(self, messages):
self._messages.clear()
self._messages.extend(messages)
async def _push_aggregation(self):
if len(self._aggregation) > 0:

41
src/pipecat/processors/aggregators/openai_llm_context.py
View File

@@ -12,7 +12,9 @@ from typing import List
from PIL import Image
from pipecat.frames.frames import Frame, VisionImageRawFrame
from pipecat.frames.frames import Frame, VisionImageRawFrame, FunctionCallInProgressFrame, FunctionCallResultFrame
from pipecat.processors.frame_processor import FrameProcessor
from openai._types import NOT_GIVEN, NotGiven
@@ -50,12 +52,11 @@ class OpenAILLMContext:
@staticmethod
def from_messages(messages: List[dict]) -> "OpenAILLMContext":
context = OpenAILLMContext()
for message in messages:
context.add_message({
"content": message["content"],
"role": message["role"],
"name": message["name"] if "name" in message else message["role"]
})
if "name" not in message:
message["name"] = message["role"]
context.add_message(message)
return context
@staticmethod
@@ -102,6 +103,34 @@ class OpenAILLMContext:
tools = NOT_GIVEN
self.tools = tools
async def call_function(
self,
f: callable,
*,
function_name: str,
tool_call_id: str,
arguments: str,
llm: FrameProcessor) -> None:
# Push a SystemFrame downstream. This frame will let our assistant context aggregator
# know that we are in the middle of a function call. Some contexts/aggregators may
# not need this. But some definitely do (Anthropic, for example).
await llm.push_frame(FunctionCallInProgressFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
))
# Define a callback function that pushes a FunctionCallResultFrame downstream.
async def function_call_result_callback(result):
await llm.push_frame(FunctionCallResultFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
result=result))
await f(function_name=function_name, tool_call_id=tool_call_id, arguments=arguments,
context=self, result_callback=function_call_result_callback)
@dataclass

55
src/pipecat/processors/frameworks/rtvi.py
View File

@@ -20,6 +20,7 @@ from pipecat.frames.frames import (
TranscriptionFrame,
TransportMessageFrame,
UserStartedSpeakingFrame,
FunctionCallResultFrame,
UserStoppedSpeakingFrame)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -176,7 +177,6 @@ class RTVIActionResponse(BaseModel):
id: str
data: RTVIActionResponseData
class RTVIBotReadyData(BaseModel):
version: str
config: List[RTVIServiceConfig]
@@ -187,6 +187,34 @@ class RTVIBotReady(BaseModel):
type: Literal["bot-ready"] = "bot-ready"
data: RTVIBotReadyData
class RTVILLMFunctionCallMessageData(BaseModel):
function_name: str
tool_call_id: str
args: dict
class RTVILLMFunctionCallMessage(BaseModel):
label: Literal["rtvi-ai"] = "rtvi-ai"
type: Literal["llm-function-call"] = "llm-function-call"
data: RTVILLMFunctionCallMessageData
class RTVILLMFunctionCallStartMessageData(BaseModel):
function_name: str
class RTVILLMFunctionCallStartMessage(BaseModel):
label: Literal["rtvi-ai"] = "rtvi-ai"
type: Literal["llm-function-call-start"] = "llm-function-call-start"
data: RTVILLMFunctionCallStartMessageData
class RTVILLMFunctionCallResultData(BaseModel):
function_name: str
tool_call_id: str
arguments: dict
result: dict
class RTVITranscriptionMessageData(BaseModel):
text: str
@@ -232,6 +260,7 @@ class RTVIProcessor(FrameProcessor):
def register_service(self, service: RTVIService):
self._registered_services[service.name] = service
async def interrupt_bot(self):
await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
@@ -273,6 +302,18 @@ class RTVIProcessor(FrameProcessor):
else:
await self.push_frame(frame, direction)
async def handle_function_call(self, function_name, tool_call_id, arguments, context, result_callback):
fn = RTVILLMFunctionCallMessageData(function_name=function_name, tool_call_id=tool_call_id, args=arguments)
message = RTVILLMFunctionCallMessage(data=fn)
frame = TransportMessageFrame(message=message.model_dump())
await self.push_frame(frame)
async def handle_function_call_start(self, function_name):
fn = RTVILLMFunctionCallStartMessageData(function_name=function_name)
message = RTVILLMFunctionCallStartMessage(data=fn)
frame = TransportMessageFrame(message=message.model_dump())
await self.push_frame(frame)
async def cleanup(self):
if self._pipeline:
await self._pipeline.cleanup()
@@ -362,6 +403,10 @@ class RTVIProcessor(FrameProcessor):
case "action":
action = RTVIActionRun.model_validate(message.data)
await self._handle_action(message.id, action)
case "llm-function-call-result":
data = RTVILLMFunctionCallResultData.model_validate(message.data)
await self._handle_function_call_result(data)
case _:
await self._send_error_response(message.id, f"Unsupported type {message.type}")
@@ -419,6 +464,14 @@ class RTVIProcessor(FrameProcessor):
await self.interrupt_bot()
await self._update_config(data)
await self._handle_get_config(request_id)
async def _handle_function_call_result(self, data):
frame = FunctionCallResultFrame(
function_name=data.function_name,
tool_call_id=data.tool_call_id,
arguments=data.arguments,
result=data.result)
await self.push_frame(frame)
async def _handle_action(self, request_id: str, data: RTVIActionRun):
action_id = self._action_id(data.service, data.action)

17
src/pipecat/processors/logger.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
from pipecat.frames.frames import Frame
from pipecat.frames.frames import BotSpeakingFrame, Frame, AudioRawFrame, TransportMessageFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from loguru import logger
from typing import Optional
@@ -12,16 +12,19 @@ logger = logger.opt(ansi=True)
class FrameLogger(FrameProcessor):
def __init__(self, prefix="Frame", color: Optional[str] = None):
def __init__(self, prefix="Frame", color: Optional[str] = None, ignored_frame_types: Optional[list] = [BotSpeakingFrame, AudioRawFrame, TransportMessageFrame]):
super().__init__()
self._prefix = prefix
self._color = color
self._ignored_frame_types = tuple(ignored_frame_types) if ignored_frame_types else None
async def process_frame(self, frame: Frame, direction: FrameDirection):
dir = "<" if direction is FrameDirection.UPSTREAM else ">"
msg = f"{dir} {self._prefix}: {frame}"
if self._color:
msg = f"<{self._color}>{msg}</>"
logger.debug(msg)
if self._ignored_frame_types and not isinstance(frame, self._ignored_frame_types):
dir = "<" if direction is FrameDirection.UPSTREAM else ">"
msg = f"{dir} {self._prefix}: {frame}"
if self._color:
msg = f"<{self._color}>{msg}</>"
logger.debug(msg)
await self.push_frame(frame, direction)

47
src/pipecat/services/ai_services.py
View File

@@ -25,11 +25,14 @@ from pipecat.frames.frames import (
TTSVoiceUpdateFrame,
TextFrame,
VisionImageRawFrame,
FunctionCallResultFrame
)
from pipecat.processors.async_frame_processor import AsyncFrameProcessor
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.audio import calculate_audio_volume
from pipecat.utils.utils import exp_smoothing
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
import re
@@ -115,27 +118,51 @@ class LLMService(AIService):
self._start_callbacks = {}
# TODO-CB: callback function type
def register_function(self, function_name: str, callback, start_callback=None):
def register_function(self, function_name: str | None, callback, start_callback=None):
# Registering a function with the function_name set to None will run that callback
# for all functions
self._callbacks[function_name] = callback
# QUESTION FOR CB: maybe this isn't needed anymore?
if start_callback:
self._start_callbacks[function_name] = start_callback
def unregister_function(self, function_name: str):
def unregister_function(self, function_name: str | None):
del self._callbacks[function_name]
if self._start_callbacks[function_name]:
del self._start_callbacks[function_name]
def has_function(self, function_name: str):
if None in self._callbacks.keys():
return True
return function_name in self._callbacks.keys()
async def call_function(self, function_name: str, args):
async def call_function(
self,
*,
context: OpenAILLMContext,
tool_call_id: str,
function_name: str,
arguments: str) -> None:
f = None
if function_name in self._callbacks.keys():
return await self._callbacks[function_name](self, args)
return None
f = self._callbacks[function_name]
elif None in self._callbacks.keys():
f = self._callbacks[None]
else:
return None
await context.call_function(
f,
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
llm=self)
# QUESTION FOR CB: maybe this isn't needed anymore?
async def call_start_function(self, function_name: str):
if function_name in self._start_callbacks.keys():
await self._start_callbacks[function_name](self)
elif None in self._start_callbacks.keys():
return await self._start_callbacks[None](function_name)
class TTSService(AIService):
@@ -151,12 +178,12 @@ class TTSService(AIService):
self._push_text_frames: bool = push_text_frames
self._current_sentence: str = ""
@abstractmethod
@ abstractmethod
async def set_voice(self, voice: str):
pass
# Converts the text to audio.
@abstractmethod
@ abstractmethod
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
pass
@@ -242,7 +269,7 @@ class STTService(AIService):
self._smoothing_factor = 0.2
self._prev_volume = 0
@abstractmethod
@ abstractmethod
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
"""Returns transcript as a string"""
pass
@@ -308,7 +335,7 @@ class ImageGenService(AIService):
super().__init__(**kwargs)
# Renders the image. Returns an Image object.
@abstractmethod
@ abstractmethod
async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]:
pass
@@ -331,7 +358,7 @@ class VisionService(AIService):
super().__init__(**kwargs)
self._describe_text = None
@abstractmethod
@ abstractmethod
async def run_vision(self, frame: VisionImageRawFrame) -> AsyncGenerator[Frame, None]:
pass

439
src/pipecat/services/anthropic.py
View File

@@ -5,19 +5,33 @@
#
import base64
import json
import io
import copy
from typing import List, Optional
from dataclasses import dataclass
from PIL import Image
from asyncio import CancelledError
import re
from pipecat.frames.frames import (
Frame,
LLMModelUpdateFrame,
TextFrame,
VisionImageRawFrame,
UserImageRequestFrame,
UserImageRawFrame,
LLMMessagesFrame,
LLMFullResponseStartFrame,
LLMFullResponseEndFrame
LLMFullResponseEndFrame,
FunctionCallResultFrame,
FunctionCallInProgressFrame,
StartInterruptionFrame
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import LLMService
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext, OpenAILLMContextFrame
from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator, LLMAssistantContextAggregator
from loguru import logger
@@ -30,21 +44,36 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
@dataclass
class AnthropicImageMessageFrame(Frame):
user_image_raw_frame: UserImageRawFrame
text: Optional[str] = None
@dataclass
class AnthropicContextAggregatorPair:
_user: 'AnthropicUserContextAggregator'
_assistant: 'AnthropicAssistantContextAggregator'
def user(self) -> str:
return self._user
def assistant(self) -> str:
return self._assistant
class AnthropicLLMService(LLMService):
"""This class implements inference with Anthropic's AI models
This service translates internally from OpenAILLMContext to the messages format
expected by the Anthropic Python SDK. We are using the OpenAILLMContext as a lingua
franca for all LLM services, so that it is easy to switch between different LLMs.
"""
def __init__(
self,
*,
api_key: str,
model: str = "claude-3-opus-20240229",
max_tokens: int = 1024):
super().__init__()
model: str = "claude-3-5-sonnet-20240620",
max_tokens: int = 4096,
**kwargs):
super().__init__(**kwargs)
self._client = AsyncAnthropic(api_key=api_key)
self._model = model
self._max_tokens = max_tokens
@@ -52,89 +81,115 @@ class AnthropicLLMService(LLMService):
def can_generate_metrics(self) -> bool:
return True
def _get_messages_from_openai_context(
self, context: OpenAILLMContext):
openai_messages = context.get_messages()
anthropic_messages = []
for message in openai_messages:
role = message["role"]
text = message["content"]
if role == "system":
role = "user"
if message.get("mime_type") == "image/jpeg":
# vision frame
encoded_image = base64.b64encode(message["data"].getvalue()).decode("utf-8")
anthropic_messages.append({
"role": role,
"content": [{
"type": "image",
"source": {
"type": "base64",
"media_type": message.get("mime_type"),
"data": encoded_image,
}
}, {
"type": "text",
"text": text
}]
})
else:
# Text frame. Anthropic needs the roles to alternate. This will
# cause an issue with interruptions. So, if we detect we are the
# ones asking again it probably means we were interrupted.
if role == "user" and len(anthropic_messages) > 1:
last_message = anthropic_messages[-1]
if last_message["role"] == "user":
anthropic_messages = anthropic_messages[:-1]
content = last_message["content"]
anthropic_messages.append(
{"role": "user", "content": f"Sorry, I just asked you about [{content}] but now I would like to know [{text}]."})
else:
anthropic_messages.append({"role": role, "content": text})
else:
anthropic_messages.append({"role": role, "content": text})
return anthropic_messages
@ staticmethod
def create_context_aggregator(context: OpenAILLMContext) -> AnthropicContextAggregatorPair:
user = AnthropicUserContextAggregator(context)
assistant = AnthropicAssistantContextAggregator(user)
return AnthropicContextAggregatorPair(
_user=user,
_assistant=assistant
)
async def _process_context(self, context: OpenAILLMContext):
await self.push_frame(LLMFullResponseStartFrame())
try:
logger.debug(f"Generating chat: {context.get_messages_json()}")
await self.push_frame(LLMFullResponseStartFrame())
await self.start_processing_metrics()
messages = self._get_messages_from_openai_context(context)
logger.debug(f"Generating chat: {context.get_messages_for_logging()}")
messages = context.messages
await self.start_ttfb_metrics()
response = await self._client.messages.create(
messages=messages,
tools=context.tools or [],
model=self._model,
max_tokens=self._max_tokens,
stream=True)
await self.stop_ttfb_metrics()
# Tool use
tool_use_block = None
json_accumulator = ''
# Usage tracking. We track the usage reported by Anthropic in prompt_tokens and
# completion_tokens. We also estimate the completion tokens from output text
# and use that estimate if we are interrupted, because we almost certainly won't
# get a complete usage report if the task we're running in is cancelled.
prompt_tokens = 0
completion_tokens = 0
completion_tokens_estimate = 0
use_completion_tokens_estimate = False
async for event in response:
# logger.debug(f"Anthropic LLM event: {event}")
if (event.type == "content_block_delta"):
await self.push_frame(TextFrame(event.delta.text))
# Aggregate streaming content, create frames, trigger events
if (event.type == "content_block_delta"):
if hasattr(event.delta, 'text'):
await self.push_frame(TextFrame(event.delta.text))
completion_tokens_estimate += self._estimate_tokens(event.delta.text)
elif hasattr(event.delta, 'partial_json') and tool_use_block:
json_accumulator += event.delta.partial_json
completion_tokens_estimate += self._estimate_tokens(
event.delta.partial_json)
elif (event.type == "content_block_start"):
if event.content_block.type == "tool_use":
tool_use_block = event.content_block
json_accumulator = ''
elif (event.type == "message_delta" and
hasattr(event.delta, 'stop_reason') and event.delta.stop_reason == 'tool_use'):
if tool_use_block:
await self.call_function(context=context,
tool_call_id=tool_use_block.id,
function_name=tool_use_block.name,
arguments=json.loads(json_accumulator))
# Calculate usage. Do this here in its own if statement, because there may be usage data
# embedded in messages that we do other processing for, above.
if hasattr(event, "usage"):
prompt_tokens += event.usage.input_tokens if hasattr(
event.usage, "input_tokens") else 0
completion_tokens += event.usage.output_tokens if hasattr(
event.usage, "output_tokens") else 0
elif hasattr(event, "message") and hasattr(event.message, "usage"):
prompt_tokens += event.message.usage.input_tokens if hasattr(
event.message.usage, "input_tokens") else 0
completion_tokens += event.message.usage.output_tokens if hasattr(
event.message.usage, "output_tokens") else 0
except CancelledError as e:
# If we're interrupted, we won't get a complete usage report. So set our flag to use the
# token estimate. The reraise the exception so all the processors running in this task
# also get cancelled.
use_completion_tokens_estimate = True
raise
except Exception as e:
logger.exception(f"{self} exception: {e}")
finally:
await self.stop_processing_metrics()
await self.push_frame(LLMFullResponseEndFrame())
await self._report_usage_metrics(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens if not use_completion_tokens_estimate else completion_tokens_estimate)
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
context = None
if isinstance(frame, OpenAILLMContextFrame):
context: OpenAILLMContext = frame.context
context = frame.context
elif isinstance(frame, LLMMessagesFrame):
context = OpenAILLMContext.from_messages(frame.messages)
context = AnthropicLLMContext.from_messages(frame.messages)
elif isinstance(frame, VisionImageRawFrame):
context = OpenAILLMContext.from_image_frame(frame)
# This is only useful in very simple pipelines because it creates
# a new context. Generally we want a context manager to catch
# UserImageRawFrames coming through the pipeline and add them
# to the context.
context = AnthropicLLMContext.from_image_frame(frame)
elif isinstance(frame, LLMModelUpdateFrame):
logger.debug(f"Switching LLM model to: [{frame.model}]")
self._model = frame.model
@@ -143,3 +198,265 @@ class AnthropicLLMService(LLMService):
if context:
await self._process_context(context)
async def request_image_frame(self, user_id: str, *, text_content: str = None):
await self.push_frame(UserImageRequestFrame(user_id=user_id, context=text_content), FrameDirection.UPSTREAM)
def _estimate_tokens(self, text: str) -> int:
return int(len(re.split(r'[^\w]+', text)) * 1.3)
async def _report_usage_metrics(self, prompt_tokens: int, completion_tokens: int):
if prompt_tokens or completion_tokens:
tokens = {
"processor": self.name,
"model": self._model,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": prompt_tokens + completion_tokens
}
await self.start_llm_usage_metrics(tokens)
class AnthropicLLMContext(OpenAILLMContext):
def __init__(
self,
messages: list[dict] | None = None,
tools: list[dict] | None = None,
tool_choice: dict | None = None,
*,
system: str | None = None
):
super().__init__(messages=messages, tools=tools, tool_choice=tool_choice)
self._user_image_request_context = {}
self.system_message = system
@ classmethod
def from_openai_context(cls, openai_context: OpenAILLMContext):
self = cls(
messages=openai_context.messages,
tools=openai_context.tools,
tool_choice=openai_context.tool_choice,
)
# See if we should pull the system message out of our context.messages list. (For
# compatibility with Open AI messages format.)
if self.messages and self.messages[0]["role"] == "system":
if len(self.messages) == 1:
# If we have only have a system message in the list, all we can really do
# without introducing too much magic is change the role to "user".
self.messages[0]["role"] = "user"
else:
# If we have more than one message, we'll pull the system message out of the
# list.
self.system_message = self.messages[0]["content"]
self.messages.pop(0)
return self
@ classmethod
def from_messages(cls, messages: List[dict]) -> "AnthropicLLMContext":
return cls(messages=messages)
@ classmethod
def from_image_frame(cls, frame: VisionImageRawFrame) -> "AnthropicLLMContext":
context = cls()
context.add_image_frame_message(
format=frame.format,
size=frame.size,
image=frame.image,
text=frame.text)
return context
def add_image_frame_message(
self, *, format: str, size: tuple[int, int], image: bytes, text: str = None):
buffer = io.BytesIO()
Image.frombytes(format, size, image).save(buffer, format="JPEG")
encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
# Anthropic docs say that the image should be the first content block in the message.
content = [{"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": encoded_image,
}}]
if text:
content.append({"type": "text", "text": text})
self.add_message({"role": "user", "content": content})
def add_message(self, message):
try:
if self.messages:
# Anthropic requires that roles alternate. If this message's role is the same as the
# last message, we should add this message's content to the last message.
if self.messages[-1]["role"] == message["role"]:
# if the last message has just a content string, convert it to a list
# in the proper format
if isinstance(self.messages[-1]["content"], str):
self.messages[-1]["content"] = [{"type": "text",
"text": self.messages[-1]["content"]}]
# if this message has just a content string, convert it to a list
# in the proper format
if isinstance(message["content"], str):
message["content"] = [{"type": "text", "text": message["content"]}]
# append the content of this message to the last message
self.messages[-1]["content"].extend(message["content"])
else:
self.messages.append(message)
else:
self.messages.append(message)
except Exception as e:
logger.error(f"Error adding message: {e}")
def get_messages_for_logging(self) -> str:
msgs = []
for message in self.messages:
msg = copy.deepcopy(message)
if "content" in msg:
if isinstance(msg["content"], list):
for item in msg["content"]:
if item["type"] == "image":
item["source"]["data"] = "..."
msgs.append(msg)
return json.dumps(msgs)
class AnthropicUserContextAggregator(LLMUserContextAggregator):
def __init__(self, context: OpenAILLMContext | AnthropicLLMContext):
super().__init__(context=context)
if isinstance(context, OpenAILLMContext):
self._context = AnthropicLLMContext.from_openai_context(context)
async def push_messages_frame(self):
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
# Our parent method has already called push_frame(). So we can't interrupt the
# flow here and we don't need to call push_frame() ourselves. Possibly something
# to talk through (tagging @aleix). At some point we might need to refactor these
# context aggregators.
try:
if isinstance(frame, UserImageRequestFrame):
# The LLM sends a UserImageRequestFrame upstream. Cache any context provided with
# that frame so we can use it when we assemble the image message in the assistant
# context aggregator.
if (frame.context):
if isinstance(frame.context, str):
self._context._user_image_request_context[frame.user_id] = frame.context
else:
logger.error(
f"Unexpected UserImageRequestFrame context type: {type(frame.context)}")
del self._context._user_image_request_context[frame.user_id]
else:
if frame.user_id in self._context._user_image_request_context:
del self._context._user_image_request_context[frame.user_id]
elif isinstance(frame, UserImageRawFrame):
# Push a new AnthropicImageMessageFrame with the text context we cached
# downstream to be handled by our assistant context aggregator. This is
# necessary so that we add the message to the context in the right order.
text = self._context._user_image_request_context.get(frame.user_id) or ""
if text:
del self._context._user_image_request_context[frame.user_id]
frame = AnthropicImageMessageFrame(user_image_raw_frame=frame, text=text)
await self.push_frame(frame)
except Exception as e:
logger.error(f"Error processing frame: {e}")
#
# Claude returns a text content block along with a tool use content block. This works quite nicely
# with streaming. We get the text first, so we can start streaming it right away. Then we get the
# tool_use block. While the text is streaming to TTS and the transport, we can run the tool call.
#
# But Claude is verbose. It would be nice to come up with prompt language that suppresses Claude's
# chattiness about it's tool thinking.
#
class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
def __init__(self, user_context_aggregator: AnthropicUserContextAggregator):
super().__init__(context=user_context_aggregator._context)
self._user_context_aggregator = user_context_aggregator
self._function_call_in_progress = None
self._function_call_result = None
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
# See note above about not calling push_frame() here.
if isinstance(frame, StartInterruptionFrame):
self._function_call_in_progress = None
self._function_call_finished = None
elif isinstance(frame, FunctionCallInProgressFrame):
self._function_call_in_progress = frame
elif isinstance(frame, FunctionCallResultFrame):
if self._function_call_in_progress and self._function_call_in_progress.tool_call_id == frame.tool_call_id:
self._function_call_in_progress = None
self._function_call_result = frame
else:
logger.warning(
f"FunctionCallResultFrame tool_call_id does not match FunctionCallInProgressFrame tool_call_id")
self._function_call_in_progress = None
self._function_call_result = None
elif isinstance(frame, AnthropicImageMessageFrame):
try:
self._context.add_image_frame_message(
format=frame.user_image_raw_frame.format,
size=frame.user_image_raw_frame.size,
image=frame.user_image_raw_frame.image,
text=frame.text)
await self._user_context_aggregator.push_messages_frame()
except Exception as e:
logger.error(f"Error processing AnthropicImageMessageFrame: {e}")
def add_message(self, message):
self._user_context_aggregator.add_message(message)
async def _push_aggregation(self):
if not self._aggregation:
return
run_llm = False
aggregation = self._aggregation
self._aggregation = ""
try:
if self._function_call_result:
frame = self._function_call_result
# TODO-khk: This was _tool_use_frame, which didn't show up anywhere else?
self._function_call_result = None
self._context.add_message({
"role": "assistant",
"content": [
{
"type": "text",
"text": aggregation
},
{
"type": "tool_use",
"id": frame.tool_call_id,
"name": frame.function_name,
"input": frame.arguments
}
]
})
self._context.add_message({
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": frame.tool_call_id,
"content": json.dumps(frame.result)
}
]
})
self._function_call_result = None
run_llm = True
else:
self._context.add_message({"role": "assistant", "content": aggregation})
if run_llm:
await self._user_context_aggregator.push_messages_frame()
except Exception as e:
logger.error(f"Error processing frame: {e}")

7
src/pipecat/services/cartesia.py
View File

@@ -15,6 +15,7 @@ from typing import AsyncGenerator
from pipecat.processors.frame_processor import FrameDirection
from pipecat.frames.frames import (
CancelFrame,
ErrorFrame,
Frame,
AudioRawFrame,
StartInterruptionFrame,
@@ -173,6 +174,12 @@ class CartesiaTTSService(TTSService):
num_channels=1
)
await self.push_frame(frame)
elif msg["type"] == "error":
logger.error(f"{self} error: {msg}")
await self.stop_all_metrics()
await self.push_frame(ErrorFrame(f'{self} error: {msg["error"]}'))
else:
logger.error(f"Cartesia error, unknown message type: {msg}")
except asyncio.CancelledError:
pass
except Exception as e:

158
src/pipecat/services/openai.py
View File

@@ -8,7 +8,9 @@ import aiohttp
import base64
import io
import json
from anthropic.types import tool_use_block
import httpx
from dataclasses import dataclass
from typing import AsyncGenerator, List, Literal
@@ -26,8 +28,13 @@ from pipecat.frames.frames import (
MetricsFrame,
TextFrame,
URLImageRawFrame,
VisionImageRawFrame
VisionImageRawFrame,
FunctionCallResultFrame,
FunctionCallInProgressFrame,
StartInterruptionFrame
)
from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator, LLMAssistantContextAggregator
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame
@@ -58,6 +65,7 @@ class OpenAIUnhandledFunctionException(Exception):
pass
class BaseOpenAILLMService(LLMService):
"""This is the base for all services that use the AsyncOpenAI client.
@@ -85,6 +93,8 @@ class BaseOpenAILLMService(LLMService):
def can_generate_metrics(self) -> bool:
return True
async def get_chat_completions(
self,
@@ -191,44 +201,13 @@ class BaseOpenAILLMService(LLMService):
arguments
):
arguments = json.loads(arguments)
result = await self.call_function(function_name, arguments)
arguments = json.dumps(arguments)
if isinstance(result, (str, dict)):
# Handle it in "full magic mode"
tool_call = ChatCompletionFunctionMessageParam({
"role": "assistant",
"tool_calls": [
{
"id": tool_call_id,
"function": {
"arguments": arguments,
"name": function_name
},
"type": "function"
}
]
})
context.add_message(tool_call)
if isinstance(result, dict):
result = json.dumps(result)
tool_result = ChatCompletionToolParam({
"tool_call_id": tool_call_id,
"role": "tool",
"content": result
})
context.add_message(tool_result)
# re-prompt to get a human answer
await self._process_context(context)
elif isinstance(result, list):
# reduced magic
for msg in result:
context.add_message(msg)
await self._process_context(context)
elif isinstance(result, type(None)):
pass
else:
raise TypeError(f"Unknown return type from function callback: {type(result)}")
await self.call_function(
context=context,
tool_call_id=tool_call_id,
function_name=function_name,
arguments=arguments
)
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -253,13 +232,30 @@ class BaseOpenAILLMService(LLMService):
await self.stop_processing_metrics()
await self.push_frame(LLMFullResponseEndFrame())
@dataclass
class OpenAIContextAggregatorPair:
_user: 'OpenAIUserContextAggregator'
_assistant: 'OpenAIAssistantContextAggregator'
def user(self) -> str:
return self._user
def assistant(self) -> str:
return self._assistant
class OpenAILLMService(BaseOpenAILLMService):
def __init__(self, *, model: str = "gpt-4o", **kwargs):
super().__init__(model=model, **kwargs)
@ staticmethod
def create_context_aggregator(context: OpenAILLMContext) -> OpenAIContextAggregatorPair:
user = OpenAIUserContextAggregator(context)
assistant = OpenAIAssistantContextAggregator(user)
return OpenAIContextAggregatorPair(
_user=user,
_assistant=assistant
)
class OpenAIImageGenService(ImageGenService):
def __init__(
@@ -360,3 +356,85 @@ class OpenAITTSService(TTSService):
yield frame
except BadRequestError as e:
logger.exception(f"{self} error generating TTS: {e}")
class OpenAIUserContextAggregator(LLMUserContextAggregator):
def __init__(self, context: OpenAILLMContext):
super().__init__(context=context)
async def push_messages_frame(self):
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
def __init__(self, user_context_aggregator: OpenAIUserContextAggregator):
super().__init__(context=user_context_aggregator._context)
self._user_context_aggregator = user_context_aggregator
self._function_call_in_progress = None
self._function_call_result = None
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
# See note above about not calling push_frame() here.
if isinstance(frame, StartInterruptionFrame):
self._function_call_in_progress = None
self._function_call_finished = None
elif isinstance(frame, FunctionCallInProgressFrame):
self._function_call_in_progress = frame
elif isinstance(frame, FunctionCallResultFrame):
if self._function_call_in_progress and self._function_call_in_progress.tool_call_id == frame.tool_call_id:
self._function_call_in_progress = None
self._function_call_result = frame
# TODO-CB: Kwin wants us to refactor this out of here but I REFUSE
await self._push_aggregation()
else:
logger.warning(
f"FunctionCallResultFrame tool_call_id does not match FunctionCallInProgressFrame tool_call_id")
self._function_call_in_progress = None
self._function_call_result = None
def add_message(self, message):
self._user_context_aggregator.add_message(message)
async def _push_aggregation(self):
if not (self._aggregation or self._function_call_result):
return
run_llm = False
aggregation = self._aggregation
self._aggregation = ""
try:
if self._function_call_result:
frame = self._function_call_result
self._context.add_message({
"role": "assistant",
"tool_calls": [
{
"id": frame.tool_call_id,
"function": {
"name": frame.function_name,
"arguments": json.dumps(frame.arguments)
},
"type": "function"
}
]
})
self._context.add_message({
"role": "tool",
"content": json.dumps(frame.result),
"tool_call_id": frame.tool_call_id
})
self._function_call_result = None
run_llm = True
else:
self._context.add_message({"role": "assistant", "content": aggregation})
if run_llm:
await self._user_context_aggregator.push_messages_frame()
except Exception as e:
logger.error(f"Error processing frame: {e}")