Merge pull request #374 from pipecat-ai/khk/together

Together.ai service implementation with Llama 3.1 function calling

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

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

@@ -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.openai_llm_context import (
-from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator, LLMAssistantContextAggregator
+    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
+                elif ((event.type == "message_delta" and
-                      hasattr(event.delta, 'stop_reason') and event.delta.stop_reason == 'tool_use'):
+                      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
+                # Calculate usage. Do this here in its own if statement, because there may be usage
-                # embedded in messages that we do other processing for, above.
+                # 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})

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