diff --git a/examples/foundational/19c-tools-togetherai.py b/examples/foundational/19c-tools-togetherai.py new file mode 100644 index 000000000..c1ef328b9 --- /dev/null +++ b/examples/foundational/19c-tools-togetherai.py @@ -0,0 +1,137 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys +import json + +from pipecat.frames.frames import LLMMessagesFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.services.cartesia import CartesiaTTSService + +from pipecat.services.together import TogetherLLMService, TogetherContextAggregatorPair +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer + +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +from runner import configure + +from loguru import logger + +from dotenv import load_dotenv +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def get_current_weather(function_name, tool_call_id, arguments, context, result_callback): + logger.debug("IN get_current_weather") + location = arguments["location"] + await result_callback(f"The weather in {location} is currently 72 degrees and sunny.") + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() + ) + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + sample_rate=16000, + ) + + llm = TogetherLLMService( + api_key=os.getenv("TOGETHER_API_KEY"), + model=os.getenv("TOGETHER_MODEL"), + ) + llm.register_function("get_current_weather", get_current_weather) + + weatherTool = { + "name": "get_current_weather", + "description": "Get the current weather in a given location", + "parameters": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + "required": ["location"], + }, + } + + system_prompt = f"""\ +You have access to the following functions: + +Use the function '{weatherTool["name"]}' to '{weatherTool["description"]}': +{json.dumps(weatherTool)} + +If you choose to call a function ONLY reply in the following format with no prefix or suffix: + +{{\"example_name\": \"example_value\"}} + +Reminder: +- Function calls MUST follow the specified format, start with +- Required parameters MUST be specified +- Only call one function at a time +- Put the entire function call reply on one line +- If there is no function call available, answer the question like normal with your current knowledge and do not tell the user about function calls + +""" + + messages = [{"role": "system", + "content": system_prompt}, + {"role": "user", + "content": "Wait for the user to say something."}] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline([ + transport.input(), # Transport user input + context_aggregator.user(), # User speech to text + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses and tool context + ]) + + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True)) + + @ transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + await task.queue_frames([LLMMessagesFrame(messages)]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/pyproject.toml b/pyproject.toml index 8b9c6cb64..71d5d3f75 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,6 +51,7 @@ openai = [ "openai~=1.35.0" ] openpipe = [ "openpipe~=4.18.0" ] playht = [ "pyht~=0.0.28" ] silero = [ "silero-vad~=5.1" ] +together = [ "together~=1.2.7" ] websocket = [ "websockets~=12.0", "fastapi~=0.111.0" ] whisper = [ "faster-whisper~=1.0.3" ] xtts = [ "resampy~=0.4.3" ] diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 2c64cf167..89f13125a 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -30,8 +30,14 @@ from pipecat.frames.frames import ( ) 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 pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, + OpenAILLMContextFrame +) +from pipecat.processors.aggregators.llm_response import ( + LLMUserContextAggregator, + LLMAssistantContextAggregator +) from loguru import logger @@ -40,7 +46,8 @@ try: except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error( - "In order to use Anthropic, you need to `pip install pipecat-ai[anthropic]`. Also, set `ANTHROPIC_API_KEY` environment variable.") + "In order to use Anthropic, you need to `pip install pipecat-ai[anthropic]`. " + + "Also, set `ANTHROPIC_API_KEY` environment variable.") raise Exception(f"Missing module: {e}") @@ -81,7 +88,7 @@ class AnthropicLLMService(LLMService): def can_generate_metrics(self) -> bool: return True - @ staticmethod + @staticmethod def create_context_aggregator(context: OpenAILLMContext) -> AnthropicContextAggregatorPair: user = AnthropicUserContextAggregator(context) assistant = AnthropicAssistantContextAggregator(user) @@ -110,7 +117,7 @@ class AnthropicLLMService(LLMService): await self.stop_ttfb_metrics() - # Tool use + # Function calling tool_use_block = None json_accumulator = '' @@ -140,16 +147,17 @@ class AnthropicLLMService(LLMService): 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'): + 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. + # 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 @@ -161,7 +169,7 @@ class AnthropicLLMService(LLMService): completion_tokens += event.message.usage.output_tokens if hasattr( event.message.usage, "output_tokens") else 0 - except CancelledError as e: + except CancelledError: # 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. @@ -174,7 +182,8 @@ class AnthropicLLMService(LLMService): 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) + 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) @@ -200,7 +209,8 @@ class AnthropicLLMService(LLMService): 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) + 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) @@ -231,7 +241,7 @@ class AnthropicLLMContext(OpenAILLMContext): self.system_message = system - @ classmethod + @classmethod def from_openai_context(cls, openai_context: OpenAILLMContext): self = cls( messages=openai_context.messages, @@ -252,11 +262,11 @@ class AnthropicLLMContext(OpenAILLMContext): self.messages.pop(0) return self - @ classmethod + @classmethod def from_messages(cls, messages: List[dict]) -> "AnthropicLLMContext": return cls(messages=messages) - @ classmethod + @classmethod def from_image_frame(cls, frame: VisionImageRawFrame) -> "AnthropicLLMContext": context = cls() context.add_image_frame_message( @@ -389,12 +399,13 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): 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: + 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") + "FunctionCallResultFrame tool_call_id != InProgressFrame tool_call_id") self._function_call_in_progress = None self._function_call_result = None elif isinstance(frame, AnthropicImageMessageFrame): @@ -423,7 +434,6 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): 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", @@ -450,7 +460,6 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): } ] }) - self._function_call_result = None run_llm = True else: self._context.add_message({"role": "assistant", "content": aggregation}) diff --git a/src/pipecat/services/together.py b/src/pipecat/services/together.py new file mode 100644 index 000000000..bf34dd099 --- /dev/null +++ b/src/pipecat/services/together.py @@ -0,0 +1,314 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import base64 +import json +import io +import copy +from typing import List, Optional +from dataclasses import dataclass +from asyncio import CancelledError +import re +import uuid + +from pipecat.frames.frames import ( + Frame, + LLMModelUpdateFrame, + TextFrame, + VisionImageRawFrame, + UserImageRequestFrame, + UserImageRawFrame, + LLMMessagesFrame, + LLMFullResponseStartFrame, + 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 + +try: + from together import AsyncTogether +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use Together.ai, you need to `pip install pipecat-ai[together]`. Also, set `TOGETHER_API_KEY` environment variable.") + raise Exception(f"Missing module: {e}") + + +@dataclass +class TogetherContextAggregatorPair: + _user: 'TogetherUserContextAggregator' + _assistant: 'TogetherAssistantContextAggregator' + + def user(self) -> str: + return self._user + + def assistant(self) -> str: + return self._assistant + + +class TogetherLLMService(LLMService): + """This class implements inference with Together's Llama 3.1 models + """ + + def __init__( + self, + *, + api_key: str, + model: str = "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo", + max_tokens: int = 4096, + **kwargs): + super().__init__(**kwargs) + self._client = AsyncTogether(api_key=api_key) + self._model = model + self._max_tokens = max_tokens + + def can_generate_metrics(self) -> bool: + return True + + @ staticmethod + def create_context_aggregator(context: OpenAILLMContext) -> TogetherContextAggregatorPair: + user = TogetherUserContextAggregator(context) + assistant = TogetherAssistantContextAggregator(user) + return TogetherContextAggregatorPair( + _user=user, + _assistant=assistant + ) + + async def _process_context(self, context: OpenAILLMContext): + try: + await self.push_frame(LLMFullResponseStartFrame()) + await self.start_processing_metrics() + + logger.debug(f"Generating chat: {context.get_messages_for_logging()}") + + await self.start_ttfb_metrics() + + stream = await self._client.chat.completions.create( + messages=context.messages, + model=self._model, + max_tokens=self._max_tokens, + stream=True, + ) + + # Function calling + got_first_chunk = False + accumulating_function_call = False + function_call_accumulator = "" + + async for chunk in stream: + # logger.debug(f"Together LLM event: {chunk}") + if chunk.usage: + tokens = { + "processor": self.name, + "model": self._model, + "prompt_tokens": chunk.usage.prompt_tokens, + "completion_tokens": chunk.usage.completion_tokens, + "total_tokens": chunk.usage.total_tokens + } + await self.start_llm_usage_metrics(tokens) + + if len(chunk.choices) == 0: + continue + + if not got_first_chunk: + await self.stop_ttfb_metrics() + if chunk.choices[0].delta.content: + got_first_chunk = True + if chunk.choices[0].delta.content[0] == "<": + accumulating_function_call = True + + if chunk.choices[0].delta.content: + if accumulating_function_call: + function_call_accumulator += chunk.choices[0].delta.content + else: + await self.push_frame(TextFrame(chunk.choices[0].delta.content)) + + if chunk.choices[0].finish_reason == 'eos' and accumulating_function_call: + await self._extract_function_call(context, function_call_accumulator) + + except CancelledError as e: + # todo: implement token counting estimates for use when the user interrupts a long generation + # we do this in the anthropic.py service + raise + except Exception as e: + logger.exception(f"{self} exception: {e}") + finally: + await self.stop_processing_metrics() + await self.push_frame(LLMFullResponseEndFrame()) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + context = None + if isinstance(frame, OpenAILLMContextFrame): + context = frame.context + elif isinstance(frame, LLMMessagesFrame): + context = TogetherLLMContext.from_messages(frame.messages) + elif isinstance(frame, LLMModelUpdateFrame): + logger.debug(f"Switching LLM model to: [{frame.model}]") + self._model = frame.model + else: + await self.push_frame(frame, direction) + + if context: + await self._process_context(context) + + async def _extract_function_call(self, context, function_call_accumulator): + context.add_message({"role": "assistant", "content": function_call_accumulator}) + + function_regex = r"(.*?)" + match = re.search(function_regex, function_call_accumulator) + if match: + function_name, args_string = match.groups() + try: + arguments = json.loads(args_string) + await self.call_function(context=context, + tool_call_id=uuid.uuid4(), + function_name=function_name, + arguments=arguments) + return + except json.JSONDecodeError as error: + # We get here if the LLM returns a function call with invalid JSON arguments. This could happen + # because of LLM non-determinism, or maybe more often because of user error in the prompt. + # Should we do anything more than log a warning? + logger.debug(f"Error parsing function arguments: {error}") + + +class TogetherLLMContext(OpenAILLMContext): + def __init__( + self, + messages: list[dict] | None = None, + ): + super().__init__(messages=messages) + + @ classmethod + def from_openai_context(cls, openai_context: OpenAILLMContext): + self = cls( + messages=openai_context.messages, + ) + return self + + @ classmethod + def from_messages(cls, messages: List[dict]) -> "TogetherLLMContext": + return cls(messages=messages) + + def add_message(self, message): + try: + self.messages.append(message) + except Exception as e: + logger.error(f"Error adding message: {e}") + + def get_messages_for_logging(self) -> str: + return json.dumps(self.messages) + + +class TogetherUserContextAggregator(LLMUserContextAggregator): + def __init__(self, context: OpenAILLMContext | TogetherLLMContext): + super().__init__(context=context) + + if isinstance(context, OpenAILLMContext): + self._context = TogetherLLMContext.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] + 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 TogetherAssistantContextAggregator(LLMAssistantContextAggregator): + def __init__(self, user_context_aggregator: TogetherUserContextAggregator): + 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 + 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._function_call_result = None + self._context.add_message({ + "role": "tool", + "content": frame.result + }) + 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}")