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

Merge pull request #531 from pipecat-ai/khk/function-calling-improvements

Browse Source
This commit is contained in:
Kwindla Hultman Kramer
2024-10-01 07:23:38 -07:00
committed by GitHub
parent 0801fc578b 92d8b37229
commit 8aae4e9856
12 changed files with 426 additions and 450 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/pipecat/processors/aggregators/openai_llm_context.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import base64
import copy
import io
import json
@@ -60,6 +62,7 @@ class OpenAILLMContext:
self._messages: List[ChatCompletionMessageParam] = messages if messages else []
self._tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = tool_choice
self._tools: List[ChatCompletionToolParam] | NotGiven = tools
self._user_image_request_context = {}
@staticmethod
def from_messages(messages: List[dict]) -> "OpenAILLMContext":
@@ -114,6 +117,19 @@ class OpenAILLMContext:
def get_messages_json(self) -> str:
return json.dumps(self._messages, cls=CustomEncoder)
def get_messages_for_logging(self) -> str:
msgs = []
for message in self.messages:
msg = copy.deepcopy(message)
if "content" in msg:
if isinstance(msg["content"], list):
for item in msg["content"]:
if item["type"] == "image_url":
if item["image_url"]["url"].startswith("data:image/"):
item["image_url"]["url"] = "data:image/..."
msgs.append(msg)
return json.dumps(msgs)
def set_tool_choice(self, tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven):
self._tool_choice = tool_choice
@@ -122,6 +138,21 @@ class OpenAILLMContext:
tools = NOT_GIVEN
self._tools = tools
def add_image_frame_message(
self, *, format: str, size: tuple[int, int], image: bytes, text: str = None
):
buffer = io.BytesIO()
Image.frombytes(format, size, image).save(buffer, format="JPEG")
encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
content = [
{"type": "text", "text": text},
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"}},
]
if text:
content.append({"type": "text", "text": text})
self.add_message({"role": "user", "content": content})
async def call_function(
self,
f: Callable[

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

@@ -116,7 +116,7 @@ class LLMService(AIService):
tool_call_id: str,
function_name: str,
arguments: str,
run_llm: bool,
run_llm: bool = True,
) -> None:
f = None
if function_name in self._callbacks.keys():

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

@@ -55,6 +55,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
# internal use only -- todo: refactor
@dataclass
class AnthropicImageMessageFrame(Frame):
user_image_raw_frame: UserImageRawFrame
@@ -359,7 +360,6 @@ class AnthropicLLMContext(OpenAILLMContext):
system: str | NotGiven = NOT_GIVEN,
):
super().__init__(messages=messages, tools=tools, tool_choice=tool_choice)
self._user_image_request_context = {}
# For beta prompt caching. This is a counter that tracks the number of turns
# we've seen above the cache threshold. We reset this when we reset the

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

@@ -31,6 +31,8 @@ from pipecat.frames.frames import (
TTSStartedFrame,
TTSStoppedFrame,
URLImageRawFrame,
UserImageRawFrame,
UserImageRequestFrame,
VisionImageRawFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
@@ -181,7 +183,7 @@ class BaseOpenAILLMService(LLMService):
async def _stream_chat_completions(
self, context: OpenAILLMContext
) -> AsyncStream[ChatCompletionChunk]:
logger.debug(f"Generating chat: {context.get_messages_json()}")
logger.debug(f"Generating chat: {context.get_messages_for_logging()}")
messages: List[ChatCompletionMessageParam] = context.get_messages()
@@ -476,10 +478,49 @@ class OpenAITTSService(TTSService):
logger.exception(f"{self} error generating TTS: {e}")
# internal use only -- todo: refactor
@dataclass
class OpenAIImageMessageFrame(Frame):
user_image_raw_frame: UserImageRawFrame
text: Optional[str] = None
class OpenAIUserContextAggregator(LLMUserContextAggregator):
def __init__(self, context: OpenAILLMContext):
super().__init__(context=context)
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.
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]
elif isinstance(frame, UserImageRawFrame):
# Push a new AnthropicImageMessageFrame with the text context we cached
# downstream to be handled by our assistant context aggregator. This is
# necessary so that we add the message to the context in the right order.
text = self._context._user_image_request_context.get(frame.user_id) or ""
if text:
del self._context._user_image_request_context[frame.user_id]
frame = OpenAIImageMessageFrame(user_image_raw_frame=frame, text=text)
await self.push_frame(frame)
except Exception as e:
logger.error(f"Error processing frame: {e}")
class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
def __init__(self, user_context_aggregator: OpenAIUserContextAggregator, **kwargs):
@@ -487,6 +528,7 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
self._user_context_aggregator = user_context_aggregator
self._function_calls_in_progress = {}
self._function_call_result = None
self._pending_image_frame_message = None
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
@@ -507,9 +549,14 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
"FunctionCallResultFrame tool_call_id does not match any function call in progress"
)
self._function_call_result = None
elif isinstance(frame, OpenAIImageMessageFrame):
self._pending_image_frame_message = frame
await self._push_aggregation()
async def _push_aggregation(self):
if not (self._aggregation or self._function_call_result):
if not (
self._aggregation or self._function_call_result or self._pending_image_frame_message
):
return
run_llm = False
@@ -548,6 +595,17 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
else:
self._context.add_message({"role": "assistant", "content": aggregation})
if self._pending_image_frame_message:
frame = self._pending_image_frame_message
self._pending_image_frame_message = None
self._context.add_image_frame_message(
format=frame.user_image_raw_frame.format,
size=frame.user_image_raw_frame.size,
image=frame.user_image_raw_frame.image,
text=frame.text,
)
run_llm = True
if run_llm:
await self._user_context_aggregator.push_context_frame()

323
src/pipecat/services/together.py
View File

@@ -4,42 +4,21 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import json
import re
import uuid
from asyncio import CancelledError
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from typing import Any, Dict, Optional
import httpx
from loguru import logger
from pydantic import BaseModel, Field
from pipecat.frames.frames import (
Frame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesFrame,
LLMUpdateSettingsFrame,
StartInterruptionFrame,
TextFrame,
UserImageRequestFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_response import (
LLMAssistantContextAggregator,
LLMUserContextAggregator,
)
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import LLMService
from pipecat.services.openai import OpenAILLMService
try:
from together import AsyncTogether
# Together.ai is recommending OpenAI-compatible function calling, so we've switched over
# to using the OpenAI client library here rather than the Together Python client library.
from openai import AsyncOpenAI, DefaultAsyncHttpxClient
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
@@ -48,19 +27,7 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
@dataclass
class TogetherContextAggregatorPair:
_user: "TogetherUserContextAggregator"
_assistant: "TogetherAssistantContextAggregator"
def user(self) -> "TogetherUserContextAggregator":
return self._user
def assistant(self) -> "TogetherAssistantContextAggregator":
return self._assistant
class TogetherLLMService(LLMService):
class TogetherLLMService(OpenAILLMService):
"""This class implements inference with Together's Llama 3.1 models"""
class InputParams(BaseModel):
@@ -68,20 +35,23 @@ class TogetherLLMService(LLMService):
max_tokens: Optional[int] = Field(default=4096, ge=1)
presence_penalty: Optional[float] = Field(default=None, ge=-2.0, le=2.0)
temperature: Optional[float] = Field(default=None, ge=0.0, le=1.0)
# Note: top_k is currently not supported by the OpenAI client library,
# so top_k is ignore right now.
top_k: Optional[int] = Field(default=None, ge=0)
top_p: Optional[float] = Field(default=None, ge=0.0, le=1.0)
extra: Optional[Dict[str, Any]] = Field(default_factory=dict)
seed: Optional[int] = Field(default=None)
def __init__(
self,
*,
api_key: str,
base_url: str = "https://api.together.xyz/v1",
model: str = "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo",
params: InputParams = InputParams(),
**kwargs,
):
super().__init__(**kwargs)
self._client = AsyncTogether(api_key=api_key)
super().__init__(api_key=api_key, base_url=base_url, model=model, params=params, **kwargs)
self.set_model_name(model)
self._max_tokens = params.max_tokens
self._frequency_penalty = params.frequency_penalty
@@ -94,15 +64,17 @@ class TogetherLLMService(LLMService):
def can_generate_metrics(self) -> bool:
return True
@staticmethod
def create_context_aggregator(
context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = True
) -> TogetherContextAggregatorPair:
user = TogetherUserContextAggregator(context)
assistant = TogetherAssistantContextAggregator(
user, expect_stripped_words=assistant_expect_stripped_words
def create_client(self, api_key=None, base_url=None, **kwargs):
logger.debug(f"Creating Together.ai client with api {base_url}")
return AsyncOpenAI(
api_key=api_key,
base_url=base_url,
http_client=DefaultAsyncHttpxClient(
limits=httpx.Limits(
max_keepalive_connections=100, max_connections=1000, keepalive_expiry=None
)
),
)
return TogetherContextAggregatorPair(_user=user, _assistant=assistant)
async def set_frequency_penalty(self, frequency_penalty: float):
logger.debug(f"Switching LLM frequency_penalty to: [{frequency_penalty}]")
@@ -150,252 +122,3 @@ class TogetherLLMService(LLMService):
await self.set_top_p(frame.top_p)
if frame.extra:
await self.set_extra(frame.extra)
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()
params = {
"messages": context.messages,
"model": self.model_name,
"max_tokens": self._max_tokens,
"stream": True,
"frequency_penalty": self._frequency_penalty,
"presence_penalty": self._presence_penalty,
"temperature": self._temperature,
"top_k": self._top_k,
"top_p": self._top_p,
}
params.update(self._extra)
stream = await self._client.chat.completions.create(**params)
# 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 = LLMTokenUsage(
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:
# 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, LLMUpdateSettingsFrame):
await self._update_settings(frame)
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=str(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, **kwargs):
super().__init__(context=user_context_aggregator._context, **kwargs)
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(
"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._reset()
try:
if self._function_call_result:
frame = self._function_call_result
self._function_call_result = None
self._context.add_message(
{
"role": "tool",
# Together expects the content here to be a string, so stringify it
"content": str(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()
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
except Exception as e:
logger.error(f"Error processing frame: {e}")