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

fixes for settings updates, context updates, and response creation

Browse Source
This commit is contained in:
Kwindla Hultman Kramer
2024-10-07 15:22:00 -07:00
parent 425ad3e90d
commit 856a0e321b
2 changed files with 89 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/pipecat/services/openai_realtime_beta/events.py
View File

@@ -1,8 +1,8 @@
import json import json
import uuid import uuid
from typing import Any, Dict, List, Literal, Optional, Union
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from typing import Any, Dict, List, Literal, Optional, Union
# #
# session properties # session properties
@@ -21,17 +21,17 @@ class TurnDetection(BaseModel):
class SessionProperties(BaseModel): class SessionProperties(BaseModel):
modalities: Optional[List[Literal["text", "audio"]]] = ["text", "audio"] modalities: Optional[List[Literal["text", "audio"]]] = None
instructions: Optional[str] = None instructions: Optional[str] = None
voice: Optional[str] = "alloy" voice: Optional[str] = None
input_audio_format: Optional[Literal["pcm16", "g711_ulaw", "g711_alaw"]] = "pcm16" input_audio_format: Optional[Literal["pcm16", "g711_ulaw", "g711_alaw"]] = None
output_audio_format: Optional[Literal["pcm16", "g711_ulaw", "g711_alaw"]] = "pcm16" output_audio_format: Optional[Literal["pcm16", "g711_ulaw", "g711_alaw"]] = None
input_audio_transcription: Optional[InputAudioTranscription] = InputAudioTranscription() input_audio_transcription: Optional[InputAudioTranscription] = None
turn_detection: Optional[TurnDetection] = TurnDetection() turn_detection: Optional[TurnDetection] = None
tools: Optional[List[Dict]] = [] tools: Optional[List[Dict]] = None
tool_choice: Optional[Literal["auto", "none", "required"]] = "auto" tool_choice: Optional[Literal["auto", "none", "required"]] = None
temperature: Optional[float] = 0.8 temperature: Optional[float] = None
max_response_output_tokens: Optional[Union[int, Literal["inf"]]] = Field(default=4096) max_response_output_tokens: Optional[Union[int, Literal["inf"]]] = None
# #
@@ -366,7 +366,7 @@ class Usage(BaseModel):
class Response(BaseModel): class Response(BaseModel):
id: str id: str
object: Literal["realtime.response"] object: Literal["realtime.response"]
status: Literal["completed", "in_progress", "incomplete", "canceled"] status: Literal["completed", "in_progress", "incomplete", "cancelled"]
status_details: Any status_details: Any
output: List[ConversationItem] output: List[ConversationItem]
usage: Optional[Usage] = None usage: Optional[Usage] = None

129
src/pipecat/services/openai_realtime_beta/llm_and_context.py
View File

@@ -1,21 +1,24 @@
import asyncio import asyncio
import base64 import base64
import json
import random import random
import traceback import traceback
import json
import websockets
from copy import deepcopy from copy import deepcopy
from dataclasses import dataclass
import websockets
from loguru import logger
from pipecat.frames.frames import ( from pipecat.frames.frames import (
CancelFrame, CancelFrame,
DataFrame,
EndFrame, EndFrame,
ErrorFrame, ErrorFrame,
Frame, Frame,
InputAudioRawFrame, InputAudioRawFrame,
LLMFullResponseStartFrame,
LLMFullResponseEndFrame, LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesUpdateFrame,
LLMUpdateSettingsFrame, LLMUpdateSettingsFrame,
StartFrame, StartFrame,
StartInterruptionFrame, StartInterruptionFrame,
@@ -26,22 +29,20 @@ from pipecat.frames.frames import (
TTSStoppedFrame, TTSStoppedFrame,
) )
from pipecat.metrics.metrics import LLMTokenUsage from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import LLMService
from pipecat.services.openai import (
OpenAIAssistantContextAggregator,
OpenAIUserContextAggregator,
OpenAIContextAggregatorPair,
)
from pipecat.processors.aggregators.openai_llm_context import ( from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext, OpenAILLMContext,
OpenAILLMContextFrame, OpenAILLMContextFrame,
) )
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import LLMService
from pipecat.services.openai import (
OpenAIAssistantContextAggregator,
OpenAIContextAggregatorPair,
OpenAIUserContextAggregator,
)
from . import events from . import events
from loguru import logger
# temp: websocket logger # temp: websocket logger
# import logging # import logging
# logging.basicConfig( # logging.basicConfig(
@@ -50,6 +51,11 @@ from loguru import logger
# ) # )
@dataclass
class _InternalMessagesUpdateFrame(DataFrame):
context: "OpenAIRealtimeLLMContext"
class OpenAIUnhandledFunctionException(Exception): class OpenAIUnhandledFunctionException(Exception):
pass pass
@@ -63,6 +69,8 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext):
obj._marker = random.randint(1, 1000) obj._marker = random.randint(1, 1000)
return obj return obj
# todo: do we need to also override add_messages() ?
def add_message(self, message): def add_message(self, message):
super().add_message(message) super().add_message(message)
if message.get("role") == "tool": if message.get("role") == "tool":
@@ -80,6 +88,17 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext):
class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator): class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator):
async def process_frame(
self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM
):
await super().process_frame(frame, direction)
# Parent does not push LLMMessagesUpdateFrame. This ensures that in a typical pipeline,
# messages are only processed by the user context aggregator, which is generally what we want. But
# we also need to send new messages over the websocket, in case audio mode triggers a response before
# we get any other context frames through the pipeline.
if isinstance(frame, LLMMessagesUpdateFrame):
await self.push_frame(_InternalMessagesUpdateFrame(context=self._context))
async def _push_aggregation(self): async def _push_aggregation(self):
# for the moment, ignore all user input coming into the pipeline. # for the moment, ignore all user input coming into the pipeline.
# todo: fix this to allow text prompting # todo: fix this to allow text prompting
@@ -162,8 +181,6 @@ class OpenAILLMServiceRealtimeBeta(LLMService):
self._receive_task.cancel() self._receive_task.cancel()
await self._receive_task await self._receive_task
self._receive_task = None self._receive_task = None
self._context_id = None
except Exception as e: except Exception as e:
logger.error(f"{self} error closing websocket: {e}") logger.error(f"{self} error closing websocket: {e}")
@@ -172,6 +189,9 @@ class OpenAILLMServiceRealtimeBeta(LLMService):
return self._websocket return self._websocket
raise Exception("Websocket not connected") raise Exception("Websocket not connected")
async def _update_settings(self, settings: events.SessionProperties):
await self.send_client_event(events.SessionUpdateEvent(session=settings))
async def _receive_task_handler(self): async def _receive_task_handler(self):
try: try:
async for message in self._get_websocket(): async for message in self._get_websocket():
@@ -180,9 +200,7 @@ class OpenAILLMServiceRealtimeBeta(LLMService):
if evt.type == "session.created": if evt.type == "session.created":
# session.created is received right after connecting. send a message # session.created is received right after connecting. send a message
# to configure the session properties. # to configure the session properties.
await self.send_client_event( await self._update_settings(self._session_properties)
events.SessionUpdateEvent(session=self._session_properties)
)
elif evt.type == "session.updated": elif evt.type == "session.updated":
self._session_properties = evt.session self._session_properties = evt.session
elif evt.type == "input_audio_buffer.speech_started": elif evt.type == "input_audio_buffer.speech_started":
@@ -303,44 +321,48 @@ class OpenAILLMServiceRealtimeBeta(LLMService):
# used with the HTTP API # used with the HTTP API
pass pass
async def _create_response(self, context: OpenAIRealtimeLLMContext): async def _send_messages_context_update(self):
try: if not self._context:
messages = context.get_unsent_messages() return
context.update_all_messages_sent() context = self._context
logger.debug( messages = context.get_unsent_messages()
f"Creating response: {context._marker} {context.get_messages_for_logging()}" context.update_all_messages_sent()
) logger.debug(
f"Sending message context updates: {context._marker} {context.get_messages_for_logging()}"
)
items = [] items = []
for m in messages: for m in messages:
if m and m.get("role") == "user": if m and (m.get("role") == "user" or m.get("role") == "system"):
content = m.get("content") content = m.get("content")
if isinstance(content, str): if isinstance(content, str):
items.append(
events.ConversationItem(
type="message",
status="completed",
role="user",
content=[events.ItemContent(type="input_text", text=content)],
)
)
else:
raise Exception(f"Invalid message content {m}")
elif m and m.get("role") == "tool":
items.append( items.append(
events.ConversationItem( events.ConversationItem(
type="function_call_output", type="message",
call_id=m.get("tool_call_id"), status="completed",
output=m["content"], role="user",
content=[events.ItemContent(type="input_text", text=content)],
) )
) )
await self.push_frame(LLMFullResponseStartFrame()) else:
await self.start_processing_metrics() raise Exception(f"Invalid message content {m}")
for item in items: elif m and m.get("role") == "tool":
await self.send_client_event(events.ConversationItemCreateEvent(item=item)) items.append(
await self.send_client_event(events.ResponseCreateEvent()) events.ConversationItem(
except Exception as e: type="function_call_output",
logger.error(f"{self} exception: {e}") call_id=m.get("tool_call_id"),
output=m["content"],
)
)
for item in items:
await self.send_client_event(events.ConversationItemCreateEvent(item=item))
async def _create_response(self):
await self._send_messages_context_update()
logger.debug(f"Creating response: {self._context.get_messages_for_logging()}")
await self.push_frame(LLMFullResponseStartFrame())
await self.start_processing_metrics()
await self.send_client_event(events.ResponseCreateEvent())
async def _send_user_audio(self, frame): async def _send_user_audio(self, frame):
payload = base64.b64encode(frame.audio).decode("utf-8") payload = base64.b64encode(frame.audio).decode("utf-8")
@@ -362,11 +384,14 @@ class OpenAILLMServiceRealtimeBeta(LLMService):
frame.context frame.context
) )
self._context = context self._context = context
await self._create_response(context) await self._create_response()
elif isinstance(frame, InputAudioRawFrame): elif isinstance(frame, InputAudioRawFrame):
await self._send_user_audio(frame) await self._send_user_audio(frame)
elif isinstance(frame, StartInterruptionFrame): elif isinstance(frame, StartInterruptionFrame):
await self._handle_interruption(frame) await self._handle_interruption(frame)
elif isinstance(frame, _InternalMessagesUpdateFrame):
self._context = frame.context
await self._send_messages_context_update()
elif isinstance(frame, LLMUpdateSettingsFrame): elif isinstance(frame, LLMUpdateSettingsFrame):
await self._update_settings(frame.settings) await self._update_settings(frame.settings)