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

Function calling (#175)

Browse Source 
* added function calling code back

* removed old llm_context file

* added integration testing for openai

* added function calling example

* added function callbacks

* added function start callback

* fixup

* fixup

* added different return type support for function calling

* intake example working

* added frame loggers

* cleanup

* fixup

* Update openai.py

* removed function call frame types

* fixup

* re-added example

* renumbered wake phrase

* fixup for autopep8

* remove unused imports
This commit is contained in:
chadbailey59
2024-05-30 12:25:39 -05:00
committed by GitHub
parent a3ba07c7a3
commit 4c3d19cc8b
31 changed files with 1187 additions and 318 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -119,7 +119,7 @@ class TextFrame(DataFrame):
text: str
def __str__(self):
return f"{self.name}(text: [{self.text}])"
return f"{self.name}(text: {self.text})"
@dataclass
@@ -132,7 +132,7 @@ class TranscriptionFrame(TextFrame):
timestamp: str
def __str__(self):
return f"{self.name}(user_id: {self.user_id}, text: [{self.text}], timestamp: {self.timestamp})"
return f"{self.name}(user: {self.user_id}, text: {self.text}, timestamp: {self.timestamp})"
@dataclass
@@ -143,7 +143,7 @@ class InterimTranscriptionFrame(TextFrame):
timestamp: str
def __str__(self):
return f"{self.name}(user: {self.user_id}, text: [{self.text}], timestamp: {self.timestamp})"
return f"{self.name}(user: {self.user_id}, text: {self.text}, timestamp: {self.timestamp})"
@dataclass

82
src/pipecat/processors/aggregators/llm_context.py
View File

@@ -1,82 +0,0 @@
#
# Copyright (c) 2024, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
from pipecat.frames.frames import Frame, InterimTranscriptionFrame, LLMMessagesFrame, TextFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
class LLMContextAggregator(FrameProcessor):
def __init__(
self,
messages: list[dict],
role: str,
complete_sentences=True,
pass_through=True,
):
super().__init__()
self._messages = messages
self._role = role
self._sentence = ""
self._complete_sentences = complete_sentences
self._pass_through = pass_through
async def process_frame(self, frame: Frame, direction: FrameDirection):
# We don't do anything with non-text frames, pass it along to next in
# the pipeline.
if not isinstance(frame, TextFrame):
await self.push_frame(frame, direction)
return
# If we get interim results, we ignore them.
if isinstance(frame, InterimTranscriptionFrame):
return
# The common case for "pass through" is receiving frames from the LLM that we'll
# use to update the "assistant" LLM messages, but also passing the text frames
# along to a TTS service to be spoken to the user.
if self._pass_through:
await self.push_frame(frame, direction)
# TODO: split up transcription by participant
if self._complete_sentences:
# type: ignore -- the linter thinks this isn't a TextFrame, even
# though we check it above
self._sentence += frame.text
if self._sentence.endswith((".", "?", "!")):
self._messages.append(
{"role": self._role, "content": self._sentence})
self._sentence = ""
await self.push_frame(LLMMessagesFrame(self._messages))
else:
# type: ignore -- the linter thinks this isn't a TextFrame, even
# though we check it above
self._messages.append({"role": self._role, "content": frame.text})
await self.push_frame(LLMMessagesFrame(self._messages))
class LLMUserContextAggregator(LLMContextAggregator):
def __init__(
self,
messages: list[dict],
complete_sentences=True):
super().__init__(
messages,
"user",
complete_sentences,
pass_through=False)
class LLMAssistantContextAggregator(LLMContextAggregator):
def __init__(
self,
messages: list[dict],
complete_sentences=True):
super().__init__(
messages,
"assistant",
complete_sentences,
pass_through=True,
)

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

@@ -6,12 +6,16 @@
from typing import List
from pipecat.services.openai import OpenAILLMContextFrame, OpenAILLMContext
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.frames.frames import (
Frame,
InterimTranscriptionFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMResponseEndFrame,
LLMResponseStartFrame,
LLMMessagesFrame,
StartInterruptionFrame,
TranscriptionFrame,
@@ -211,3 +215,44 @@ class LLMFullResponseAggregator(FrameProcessor):
self._aggregation = ""
else:
await self.push_frame(frame, direction)
class LLMContextAggregator(LLMResponseAggregator):
def __init__(self, *, context: OpenAILLMContext, **kwargs):
self._context = context
super().__init__(**kwargs)
async def _push_aggregation(self):
if len(self._aggregation) > 0:
self._context.add_message({"role": self._role, "content": self._aggregation})
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
# Reset our accumulator state.
self._reset()
class LLMAssistantContextAggregator(LLMContextAggregator):
def __init__(self, context: OpenAILLMContext):
super().__init__(
messages=[],
context=context,
role="assistant",
start_frame=LLMResponseStartFrame,
end_frame=LLMResponseEndFrame,
accumulator_frame=TextFrame
)
class LLMUserContextAggregator(LLMContextAggregator):
def __init__(self, context: OpenAILLMContext):
super().__init__(
messages=[],
context=context,
role="user",
start_frame=UserStartedSpeakingFrame,
end_frame=UserStoppedSpeakingFrame,
accumulator_frame=TranscriptionFrame,
interim_accumulator_frame=InterimTranscriptionFrame
)

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

@@ -6,17 +6,22 @@
from pipecat.frames.frames import Frame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from loguru import logger
from typing import Optional
logger = logger.opt(ansi=True)
class FrameLogger(FrameProcessor):
def __init__(self, prefix="Frame"):
def __init__(self, prefix="Frame", color: Optional[str] = None):
super().__init__()
self._prefix = prefix
self._color = color
async def process_frame(self, frame: Frame, direction: FrameDirection):
match direction:
case FrameDirection.UPSTREAM:
print(f"< {self._prefix}: {frame}")
case FrameDirection.DOWNSTREAM:
print(f"> {self._prefix}: {frame}")
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)

2
src/pipecat/services/azure.py
View File

@@ -46,7 +46,7 @@ class AzureTTSService(TTSService):
self._voice = voice
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Transcribing text: {text}")
logger.debug(f"Generating TTS: {text}")
ssml = (
"<speak version='1.0' xml:lang='en-US' xmlns='http://www.w3.org/2001/10/synthesis' "

2
src/pipecat/services/elevenlabs.py
View File

@@ -32,7 +32,7 @@ class ElevenLabsTTSService(TTSService):
self._model = model
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Transcribing text: [{text}]")
logger.debug(f"Generating TTS: [{text}]")
url = f"https://api.elevenlabs.io/v1/text-to-speech/{self._voice_id}/stream"

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

@@ -5,6 +5,7 @@
#
import io
import json
import time
import aiohttp
import base64
@@ -28,13 +29,19 @@ from pipecat.frames.frames import (
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext, OpenAILLMContextFrame
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import LLMService, ImageGenService
from openai.types.chat import (
ChatCompletionSystemMessageParam,
ChatCompletionFunctionMessageParam,
ChatCompletionToolParam,
ChatCompletionUserMessageParam,
)
from loguru import logger
try:
from openai import AsyncOpenAI, AsyncStream
from openai.types.chat import (
ChatCompletion,
ChatCompletionChunk,
ChatCompletionMessageParam,
)
@@ -45,6 +52,10 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
class OpenAIUnhandledFunctionException(BaseException):
pass
class BaseOpenAILLMService(LLMService):
"""This is the base for all services that use the AsyncOpenAI client.
@@ -59,10 +70,23 @@ class BaseOpenAILLMService(LLMService):
super().__init__()
self._model: str = model
self._client = self.create_client(api_key=api_key, base_url=base_url)
self._callbacks = {}
self._start_callbacks = {}
def create_client(self, api_key=None, base_url=None):
return AsyncOpenAI(api_key=api_key, base_url=base_url)
# TODO-CB: callback function type
def register_function(self, function_name, callback, start_callback=None):
self._callbacks[function_name] = callback
if start_callback:
self._start_callbacks[function_name] = start_callback
def unregister_function(self, function_name):
del self._callbacks[function_name]
if self._start_callbacks[function_name]:
del self._start_callbacks[function_name]
async def _stream_chat_completions(
self, context: OpenAILLMContext
) -> AsyncStream[ChatCompletionChunk]:
@@ -97,16 +121,24 @@ class BaseOpenAILLMService(LLMService):
return chunks
async def _chat_completions(self, messages) -> str | None:
response: ChatCompletion = await self._client.chat.completions.create(
model=self._model, stream=False, messages=messages
)
if response and len(response.choices) > 0:
return response.choices[0].message.content
else:
return None
async def _process_context(self, context: OpenAILLMContext):
function_name = ""
arguments = ""
tool_call_id = ""
chunk_stream: AsyncStream[ChatCompletionChunk] = (
await self._stream_chat_completions(context)
)
await self.push_frame(LLMFullResponseStartFrame())
async for chunk in chunk_stream:
if len(chunk.choices) == 0:
continue
@@ -126,23 +158,77 @@ class BaseOpenAILLMService(LLMService):
tool_call = chunk.choices[0].delta.tool_calls[0]
if tool_call.function and tool_call.function.name:
function_name += tool_call.function.name
# yield LLMFunctionStartFrame(function_name=tool_call.function.name)
tool_call_id = tool_call.id
# only send a function start frame if we're not handling the function call
if function_name in self._callbacks.keys():
if function_name in self._start_callbacks.keys():
await self._start_callbacks[function_name](self)
if tool_call.function and tool_call.function.arguments:
# Keep iterating through the response to collect all the argument fragments and
# yield a complete LLMFunctionCallFrame after run_llm_async
# completes
# Keep iterating through the response to collect all the argument fragments
arguments += tool_call.function.arguments
elif chunk.choices[0].delta.content:
await self.push_frame(LLMResponseStartFrame())
await self.push_frame(TextFrame(chunk.choices[0].delta.content))
await self.push_frame(LLMResponseEndFrame())
await self.push_frame(LLMFullResponseEndFrame())
# if we got a function name and arguments, check to see if it's a function with
# a registered handler. If so, run the registered callback, save the result to
# the context, and re-prompt to get a chat answer. If we don't have a registered
# handler, raise an exception.
if function_name and arguments:
if function_name in self._callbacks.keys():
await self._handle_function_call(context, tool_call_id, function_name, arguments)
# if we got a function name and arguments, yield the frame with all the info so
# frame consumers can take action based on the function call.
# if function_name and arguments:
# yield LLMFunctionCallFrame(function_name=function_name, arguments=arguments)
else:
raise OpenAIUnhandledFunctionException(
f"The LLM tried to call a function named '{function_name}', but there isn't a callback registered for that function.")
async def _handle_function_call(
self,
context,
tool_call_id,
function_name,
arguments
):
arguments = json.loads(arguments)
result = await self._callbacks[function_name](self, 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 BaseException(f"Unknown return type from function callback: {type(result)}")
async def process_frame(self, frame: Frame, direction: FrameDirection):
context = None
@@ -156,7 +242,9 @@ class BaseOpenAILLMService(LLMService):
await self.push_frame(frame, direction)
if context:
await self.push_frame(LLMFullResponseStartFrame())
await self._process_context(context)
await self.push_frame(LLMFullResponseEndFrame())
class OpenAILLMService(BaseOpenAILLMService):

1
src/pipecat/transports/base_output.py
View File

@@ -158,7 +158,6 @@ class BaseOutputTransport(FrameProcessor):
while self._running:
try:
frame = self._sink_queue.get(timeout=1)
if not self._is_interrupted.is_set():
if isinstance(frame, AudioRawFrame):
if self._params.audio_out_enabled:

41
src/pipecat/utils/test_frame_processor.py Normal file
View File

@@ -0,0 +1,41 @@
from typing import List
from pipecat.processors.frame_processor import FrameProcessor
class TestException(BaseException):
pass
class TestFrameProcessor(FrameProcessor):
def __init__(self, test_frames):
self.test_frames = test_frames
self._list_counter = 0
super().__init__()
async def process_frame(self, frame, direction):
if not self.test_frames[0]: # then we've run out of required frames but the generator is still going?
raise TestException(f"Oops, got an extra frame, {frame}")
if isinstance(self.test_frames[0], List):
# We need to consume frames until we see the next frame type after this
next_frame = self.test_frames[1]
if isinstance(frame, next_frame):
# we're done iterating the list I guess
print(f"TestFrameProcessor got expected list exit frame: {frame}")
# pop twice to get rid of the list, as well as the next frame
self.test_frames.pop(0)
self.test_frames.pop(0)
self.list_counter = 0
else:
fl = self.test_frames[0]
fl_el = fl[self._list_counter % len(fl)]
if isinstance(frame, fl_el):
print(f"TestFrameProcessor got expected list frame: {frame}")
self._list_counter += 1
else:
raise TestException(f"Inside a list, expected {fl_el} but got {frame}")
else:
if not isinstance(frame, self.test_frames[0]):
raise TestException(f"Expected {self.test_frames[0]}, but got {frame}")
print(f"TestFrameProcessor got expected frame: {frame}")
self.test_frames.pop(0)