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/macos-py3.10-requirements.txt b/macos-py3.10-requirements.txt index a764e62d4..1c985d470 100644 --- a/macos-py3.10-requirements.txt +++ b/macos-py3.10-requirements.txt @@ -1,5 +1,6 @@ +WARNING: --strip-extras is becoming the default in version 8.0.0. To silence this warning, either use --strip-extras to opt into the new default or use --no-strip-extras to retain the existing behavior. # -# This file is autogenerated by pip-compile with Python 3.10 +# This file is autogenerated by pip-compile with Python 3.11 # by the following command: # # pip-compile --all-extras pyproject.toml @@ -12,6 +13,7 @@ aiohttp==3.9.5 # langchain # langchain-community # pipecat-ai (pyproject.toml) + # together aiosignal==1.3.1 # via aiohttp annotated-types==0.7.0 @@ -27,10 +29,6 @@ anyio==4.4.0 # openai # starlette # watchfiles -async-timeout==4.0.3 - # via - # aiohttp - # langchain attrs==23.2.0 # via # aiohttp @@ -53,6 +51,7 @@ charset-normalizer==3.3.2 click==8.1.7 # via # flask + # together # typer # uvicorn coloredlogs==15.0.1 @@ -77,8 +76,8 @@ einops==0.8.0 # via pipecat-ai (pyproject.toml) email-validator==2.2.0 # via fastapi -exceptiongroup==1.2.2 - # via anyio +eval-type-backport==0.2.0 + # via together fal-client==0.4.1 # via pipecat-ai (pyproject.toml) fastapi==0.111.1 @@ -91,6 +90,7 @@ filelock==3.15.4 # via # huggingface-hub # pyht + # together # torch # transformers flask==3.0.3 @@ -192,13 +192,13 @@ jsonpatch==1.33 # via langchain-core jsonpointer==3.0.0 # via jsonpatch -langchain==0.2.12 +langchain==0.2.13 # via # langchain-community # pipecat-ai (pyproject.toml) -langchain-community==0.2.11 +langchain-community==0.2.12 # via pipecat-ai (pyproject.toml) -langchain-core==0.2.29 +langchain-core==0.2.30 # via # langchain # langchain-community @@ -208,7 +208,7 @@ langchain-openai==0.1.20 # via pipecat-ai (pyproject.toml) langchain-text-splitters==0.2.2 # via langchain -langsmith==0.1.98 +langsmith==0.1.99 # via # langchain # langchain-community @@ -247,9 +247,11 @@ numpy==1.26.4 # numba # onnxruntime # pipecat-ai (pyproject.toml) + # pyarrow # pyloudnorm # resampy # scipy + # together # torchvision # transformers onnxruntime==1.18.1 @@ -275,6 +277,7 @@ packaging==24.1 pillow==10.3.0 # via # pipecat-ai (pyproject.toml) + # together # torchvision proto-plus==1.24.0 # via @@ -291,6 +294,8 @@ protobuf==4.25.4 # pipecat-ai (pyproject.toml) # proto-plus # pyht +pyarrow==17.0.0 + # via together pyasn1==0.6.0 # via # pyasn1-modules @@ -308,6 +313,7 @@ pydantic==2.8.2 # langchain-core # langsmith # openai + # together pydantic-core==2.20.1 # via pydantic pygments==2.18.0 @@ -349,6 +355,7 @@ requests==2.32.3 # langsmith # pyht # tiktoken + # together # transformers resampy==0.4.3 # via pipecat-ai (pyproject.toml) @@ -380,10 +387,12 @@ sqlalchemy==2.0.32 # langchain-community starlette==0.37.2 # via fastapi -sympy==1.13.1 +sympy==1.13.2 # via # onnxruntime # torch +tabulate==0.9.0 + # via together tenacity==8.5.0 # via # langchain @@ -393,6 +402,8 @@ tiktoken==0.7.0 # via langchain-openai timm==0.9.16 # via pipecat-ai (pyproject.toml) +together==1.2.7 + # via pipecat-ai (pyproject.toml) tokenizers==0.19.1 # via # anthropic @@ -413,15 +424,17 @@ tqdm==4.66.5 # google-generativeai # huggingface-hub # openai + # together # transformers transformers==4.40.2 # via pipecat-ai (pyproject.toml) typer==0.12.3 - # via fastapi-cli + # via + # fastapi-cli + # together typing-extensions==4.12.2 # via # anthropic - # anyio # deepgram-sdk # fastapi # google-generativeai @@ -435,14 +448,13 @@ typing-extensions==4.12.2 # torch # typer # typing-inspect - # uvicorn typing-inspect==0.9.0 # via dataclasses-json uritemplate==4.1.1 # via google-api-python-client urllib3==2.2.2 # via requests -uvicorn[standard]==0.30.5 +uvicorn[standard]==0.30.6 # via # fastapi # fastapi-cli 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..49cfff52c 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -110,7 +110,7 @@ class AnthropicLLMService(LLMService): await self.stop_ttfb_metrics() - # Tool use + # Function calling tool_use_block = None json_accumulator = '' @@ -423,7 +423,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 +449,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}")