Service for together.ai, including Llama 3.1 function calling support

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:
+
+<function=example_function_name>{{\"example_name\": \"example_value\"}}</function>
+
+Reminder:
+- Function calls MUST follow the specified format, start with <function= and end with </function>
+- 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())

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
+eval-type-backport==0.2.0
-    # via anyio
+    # 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

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" ]

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})

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"<function=(\w+)>(.*?)</function>"
+        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}")