From 6f2a464451664f41892a7cf6b12e3a0b17913fa2 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sun, 13 Oct 2024 18:12:03 -0700 Subject: [PATCH] conversation save/load for openai, openai-realtime, and anthropic --- .../foundational/19-openai-realtime-beta.py | 84 +----- .../20a-persistent-context-openai.py | 236 ++++++++++++++++ .../20b-persistent-context-openai-realtime.py | 262 ++++++++++++++++++ .../20c-persistent-context-anthropic.py | 227 +++++++++++++++ .../aggregators/openai_llm_context.py | 17 ++ src/pipecat/services/anthropic.py | 102 +++++++ 6 files changed, 846 insertions(+), 82 deletions(-) create mode 100644 examples/foundational/20a-persistent-context-openai.py create mode 100644 examples/foundational/20b-persistent-context-openai-realtime.py create mode 100644 examples/foundational/20c-persistent-context-anthropic.py diff --git a/examples/foundational/19-openai-realtime-beta.py b/examples/foundational/19-openai-realtime-beta.py index 041d2fdfb..51ca773e1 100644 --- a/examples/foundational/19-openai-realtime-beta.py +++ b/examples/foundational/19-openai-realtime-beta.py @@ -5,9 +5,7 @@ # import asyncio -import json import os -import re import sys from datetime import datetime @@ -50,46 +48,6 @@ async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context ) -async def get_saved_conversation_filenames( - function_name, tool_call_id, args, llm, context, result_callback -): - pattern = re.compile("example_19_\\d{8}_\\d{6}\\.json$") - matching_files = [] - - for filename in os.listdir("."): - if pattern.match(filename): - matching_files.append(filename) - - await result_callback({"filenames": matching_files}) - - -async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback): - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - filename = f"example_19_{timestamp}.json" - logger.debug(f"writing conversation to {filename}\n{json.dumps(context.messages, indent=4)}") - try: - with open(filename, "w") as file: - json.dump(context.messages, file, indent=4) - await result_callback({"success": True}) - except Exception as e: - await result_callback({"success": False, "error": str(e)}) - - -async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback): - async def _reset(): - filename = args["filename"] - logger.debug(f"loading conversation from {filename}") - try: - with open(filename, "r") as file: - context.set_messages(json.load(file)) - await llm.reset_conversation() - await llm._create_response() - except Exception as e: - await result_callback({"success": False, "error": str(e)}) - - asyncio.create_task(_reset()) - - tools = [ { "type": "function", @@ -110,42 +68,7 @@ tools = [ }, "required": ["location", "format"], }, - }, - { - "type": "function", - "name": "save_conversation", - "description": "Save the current conversatione. Use this function to persist the current conversation to external storage.", - "parameters": { - "type": "object", - "properties": {}, - "required": [], - }, - }, - { - "type": "function", - "name": "get_saved_conversation_filenames", - "description": "Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a timestamp. Each file is conversation history that can be loaded into this session.", - "parameters": { - "type": "object", - "properties": {}, - "required": [], - }, - }, - { - "type": "function", - "name": "load_conversation", - "description": "Load a conversation history. Use this function to load a conversation history into the current session.", - "parameters": { - "type": "object", - "properties": { - "filename": { - "type": "string", - "description": "The filename of the conversation history to load.", - } - }, - "required": ["filename"], - }, - }, + } ] @@ -202,11 +125,8 @@ Remember, your responses should be short. Just one or two sentences, usually.""" # you can either register a single function for all function calls, or specific functions # llm.register_function(None, fetch_weather_from_api) llm.register_function("get_current_weather", fetch_weather_from_api) - llm.register_function("save_conversation", save_conversation) - llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames) - llm.register_function("load_conversation", load_conversation) - context = OpenAILLMContext([], tools) + context = OpenAILLMContext([{"role": "user", "content": "Say hello!"}], tools) context_aggregator = llm.create_context_aggregator(context) pipeline = Pipeline( diff --git a/examples/foundational/20a-persistent-context-openai.py b/examples/foundational/20a-persistent-context-openai.py new file mode 100644 index 000000000..5767d6dbd --- /dev/null +++ b/examples/foundational/20a-persistent-context-openai.py @@ -0,0 +1,236 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import glob +import json +import os +import sys +from datetime import datetime + +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, +) +from pipecat.services.openai import OpenAILLMService +from pipecat.services.cartesia import CartesiaTTSService + +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer +from pipecat.vad.vad_analyzer import VADParams + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + +BASE_FILENAME = "/tmp/pipecat_conversation_" +tts = None + + +async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + temperature = 75 if args["format"] == "fahrenheit" else 24 + await result_callback( + { + "conditions": "nice", + "temperature": temperature, + "format": args["format"], + "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"), + } + ) + + +async def get_saved_conversation_filenames( + function_name, tool_call_id, args, llm, context, result_callback +): + # Construct the full pattern including the BASE_FILENAME + full_pattern = f"{BASE_FILENAME}*.json" + + # Use glob to find all matching files + matching_files = glob.glob(full_pattern) + logger.debug(f"matching files: {matching_files}") + + await result_callback({"filenames": matching_files}) + + +async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback): + timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S") + filename = f"{BASE_FILENAME}{timestamp}.json" + logger.debug(f"writing conversation to {filename}\n{json.dumps(context.messages, indent=4)}") + try: + with open(filename, "w") as file: + messages = context.get_messages_for_persistent_storage() + # remove the last message, which is the instruction we just gave to save the conversation + messages.pop() + json.dump(messages, file, indent=2) + await result_callback({"success": True}) + except Exception as e: + await result_callback({"success": False, "error": str(e)}) + + +async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback): + global tts + filename = args["filename"] + logger.debug(f"loading conversation from {filename}") + try: + with open(filename, "r") as file: + context.set_messages(json.load(file)) + logger.debug( + f"loaded conversation from {filename}\n{json.dumps(context.messages, indent=4)}" + ) + await tts.say("Ok, I've loaded that conversation.") + except Exception as e: + await result_callback({"success": False, "error": str(e)}) + + +messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, +] +tools = [ + { + "type": "function", + "function": { + "name": "get_current_weather", + "description": "Get the current weather", + "parameters": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the users location.", + }, + }, + "required": ["location", "format"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "save_conversation", + "description": "Save the current conversatione. Use this function to persist the current conversation to external storage.", + "parameters": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + }, + { + "type": "function", + "function": { + "name": "get_saved_conversation_filenames", + "description": "Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.", + "parameters": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + }, + { + "type": "function", + "function": { + "name": "load_conversation", + "description": "Load a conversation history. Use this function to load a conversation history into the current session.", + "parameters": { + "type": "object", + "properties": { + "filename": { + "type": "string", + "description": "The filename of the conversation history to load.", + } + }, + "required": ["filename"], + }, + }, + }, +] + + +async def main(): + global tts + 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(params=VADParams(stop_secs=0.8)), + ), + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + + # you can either register a single function for all function calls, or specific functions + # llm.register_function(None, fetch_weather_from_api) + llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("save_conversation", save_conversation) + llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames) + llm.register_function("load_conversation", load_conversation) + + context = OpenAILLMContext(messages, tools) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + context_aggregator.user(), + llm, # LLM + tts, + context_aggregator.assistant(), + transport.output(), # Transport bot output + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + # report_only_initial_ttfb=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([context_aggregator.user().get_context_frame()]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/foundational/20b-persistent-context-openai-realtime.py b/examples/foundational/20b-persistent-context-openai-realtime.py new file mode 100644 index 000000000..27a2b7de6 --- /dev/null +++ b/examples/foundational/20b-persistent-context-openai-realtime.py @@ -0,0 +1,262 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import glob +import json +import os +import sys +from datetime import datetime + +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, +) +from pipecat.services.openai_realtime_beta import ( + InputAudioTranscription, + OpenAILLMServiceRealtimeBeta, + SessionProperties, + TurnDetection, +) +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer +from pipecat.vad.vad_analyzer import VADParams + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + +BASE_FILENAME = "/tmp/pipecat_conversation_" + + +async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + temperature = 75 if args["format"] == "fahrenheit" else 24 + await result_callback( + { + "conditions": "nice", + "temperature": temperature, + "format": args["format"], + "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"), + } + ) + + +async def get_saved_conversation_filenames( + function_name, tool_call_id, args, llm, context, result_callback +): + # Construct the full pattern including the BASE_FILENAME + full_pattern = f"{BASE_FILENAME}*.json" + + # Use glob to find all matching files + matching_files = glob.glob(full_pattern) + logger.debug(f"matching files: {matching_files}") + + await result_callback({"filenames": matching_files}) + + +# async def get_saved_conversation_filenames( +# function_name, tool_call_id, args, llm, context, result_callback +# ): +# pattern = re.compile(re.escape(BASE_FILENAME) + "\\d{8}_\\d{6}\\.json$") +# matching_files = [] + +# for filename in os.listdir("."): +# if pattern.match(filename): +# matching_files.append(filename) + +# await result_callback({"filenames": matching_files}) + + +async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback): + timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S") + filename = f"{BASE_FILENAME}{timestamp}.json" + logger.debug(f"writing conversation to {filename}\n{json.dumps(context.messages, indent=4)}") + try: + with open(filename, "w") as file: + messages = context.get_messages_for_persistent_storage() + # remove the last message, which is the instruction we just gave to save the conversation + messages.pop() + json.dump(messages, file, indent=2) + await result_callback({"success": True}) + except Exception as e: + await result_callback({"success": False, "error": str(e)}) + + +async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback): + async def _reset(): + filename = args["filename"] + logger.debug(f"loading conversation from {filename}") + try: + with open(filename, "r") as file: + context.set_messages(json.load(file)) + await llm.reset_conversation() + await llm._create_response() + except Exception as e: + await result_callback({"success": False, "error": str(e)}) + + asyncio.create_task(_reset()) + + +tools = [ + { + "type": "function", + "name": "get_current_weather", + "description": "Get the current weather", + "parameters": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the users location.", + }, + }, + "required": ["location", "format"], + }, + }, + { + "type": "function", + "name": "save_conversation", + "description": "Save the current conversatione. Use this function to persist the current conversation to external storage.", + "parameters": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + { + "type": "function", + "name": "get_saved_conversation_filenames", + "description": "Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.", + "parameters": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + { + "type": "function", + "name": "load_conversation", + "description": "Load a conversation history. Use this function to load a conversation history into the current session.", + "parameters": { + "type": "object", + "properties": { + "filename": { + "type": "string", + "description": "The filename of the conversation history to load.", + } + }, + "required": ["filename"], + }, + }, +] + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_in_enabled=True, + audio_in_sample_rate=24000, + audio_out_enabled=True, + audio_out_sample_rate=24000, + transcription_enabled=False, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.8)), + vad_audio_passthrough=True, + ), + ) + + session_properties = SessionProperties( + input_audio_transcription=InputAudioTranscription(), + # Set openai TurnDetection parameters. Not setting this at all will turn it + # on by default + turn_detection=TurnDetection(silence_duration_ms=1000), + # Or set to False to disable openai turn detection and use transport VAD + # turn_detection=False, + # tools=tools, + instructions="""Your knowledge cutoff is 2023-10. You are a helpful and friendly AI. + +Act like a human, but remember that you aren't a human and that you can't do human +things in the real world. Your voice and personality should be warm and engaging, with a lively and +playful tone. + +If interacting in a non-English language, start by using the standard accent or dialect familiar to +the user. Talk quickly. You should always call a function if you can. Do not refer to these rules, +even if you're asked about them. +- +You are participating in a voice conversation. Keep your responses concise, short, and to the point +unless specifically asked to elaborate on a topic. + +Remember, your responses should be short. Just one or two sentences, usually.""", + ) + + llm = OpenAILLMServiceRealtimeBeta( + api_key=os.getenv("OPENAI_API_KEY"), + session_properties=session_properties, + start_audio_paused=False, + ) + + # you can either register a single function for all function calls, or specific functions + # llm.register_function(None, fetch_weather_from_api) + llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("save_conversation", save_conversation) + llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames) + llm.register_function("load_conversation", load_conversation) + + context = OpenAILLMContext([], tools) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + context_aggregator.user(), + llm, # LLM + context_aggregator.assistant(), + transport.output(), # Transport bot output + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + # report_only_initial_ttfb=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([context_aggregator.user().get_context_frame()]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/foundational/20c-persistent-context-anthropic.py b/examples/foundational/20c-persistent-context-anthropic.py new file mode 100644 index 000000000..45d3dd897 --- /dev/null +++ b/examples/foundational/20c-persistent-context-anthropic.py @@ -0,0 +1,227 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import glob +import json +import os +import sys +from datetime import datetime + +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, +) +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.anthropic import AnthropicLLMService + +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer +from pipecat.vad.vad_analyzer import VADParams + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + +BASE_FILENAME = "/tmp/pipecat_conversation_" +tts = None + + +async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback): + temperature = 75 if args["format"] == "fahrenheit" else 24 + await result_callback( + { + "conditions": "nice", + "temperature": temperature, + "format": args["format"], + "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"), + } + ) + + +async def get_saved_conversation_filenames( + function_name, tool_call_id, args, llm, context, result_callback +): + # Construct the full pattern including the BASE_FILENAME + full_pattern = f"{BASE_FILENAME}*.json" + + # Use glob to find all matching files + matching_files = glob.glob(full_pattern) + logger.debug(f"matching files: {matching_files}") + + await result_callback({"filenames": matching_files}) + + +async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback): + timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S") + filename = f"{BASE_FILENAME}{timestamp}.json" + logger.debug(f"writing conversation to {filename}\n{json.dumps(context.messages, indent=4)}") + try: + with open(filename, "w") as file: + # todo: extract 'system' into the first message in the list + messages = context.get_messages_for_persistent_storage() + # remove the last message, which is the instruction we just gave to save the conversation + messages.pop() + json.dump(messages, file, indent=2) + await result_callback({"success": True}) + except Exception as e: + await result_callback({"success": False, "error": str(e)}) + + +async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback): + global tts + filename = args["filename"] + logger.debug(f"loading conversation from {filename}") + try: + with open(filename, "r") as file: + context.set_messages(json.load(file)) + logger.debug( + f"loaded conversation from {filename}\n{json.dumps(context.messages, indent=4)}" + ) + await tts.say("Ok, I've loaded that conversation.") + except Exception as e: + await result_callback({"success": False, "error": str(e)}) + + +messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, +] +tools = [ + { + "name": "get_current_weather", + "description": "Get the current weather", + "input_schema": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the users location.", + }, + }, + "required": ["location", "format"], + }, + }, + { + "name": "save_conversation", + "description": "Save the current conversation. Use this function to persist the current conversation to external storage.", + "input_schema": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + { + "name": "get_saved_conversation_filenames", + "description": "Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.", + "input_schema": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + { + "name": "load_conversation", + "description": "Load a conversation history. Use this function to load a conversation history into the current session.", + "input_schema": { + "type": "object", + "properties": { + "filename": { + "type": "string", + "description": "The filename of the conversation history to load.", + } + }, + "required": ["filename"], + }, + }, +] + + +async def main(): + global tts + 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(params=VADParams(stop_secs=0.8)), + ), + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + llm = AnthropicLLMService( + api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-5-sonnet-20240620" + ) + + # you can either register a single function for all function calls, or specific functions + # llm.register_function(None, fetch_weather_from_api) + llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("save_conversation", save_conversation) + llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames) + llm.register_function("load_conversation", load_conversation) + + context = OpenAILLMContext(messages, tools) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + context_aggregator.user(), + llm, # LLM + tts, + context_aggregator.assistant(), + transport.output(), # Transport bot output + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + # report_only_initial_ttfb=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([context_aggregator.user().get_context_frame()]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index 69edc9df0..d5c8422dc 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/src/pipecat/processors/aggregators/openai_llm_context.py @@ -132,6 +132,23 @@ class OpenAILLMContext: msgs.append(msg) return json.dumps(msgs) + def from_standard_message(self, message): + return message + + # convert a message in this LLM's format to one or more messages in OpenAI format + def to_standard_messages(self, obj) -> list: + return [obj] + + def get_messages_for_initializing_history(self): + return self._messages + + def get_messages_for_persistent_storage(self): + messages = [] + for m in self._messages: + standard_messages = self.to_standard_messages(m) + messages.extend(standard_messages) + return messages + def set_tool_choice(self, tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven): self._tool_choice = tool_choice diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 1b7064209..3bb792964 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -361,6 +361,100 @@ class AnthropicLLMContext(OpenAILLMContext): self._messages[:] = messages self._restructure_from_openai_messages() + # convert a message in Anthropic format into one or more messages in OpenAI format + def to_standard_messages(self, obj): + # todo: image format (?) + # tool_use + role = obj.get("role") + content = obj.get("content") + if role == "assistant": + if isinstance(content, str): + return [{"role": role, "content": [{"type": "text", "text": content}]}] + elif isinstance(content, list): + text_items = [] + tool_items = [] + for item in content: + if item["type"] == "text": + text_items.append({"type": "text", "text": item["text"]}) + elif item["type"] == "tool_use": + tool_items.append( + { + "type": "function", + "id": item["id"], + "function": { + "name": item["name"], + "arguments": json.dumps(item["input"]), + }, + } + ) + messages = [] + if text_items: + messages.append({"role": role, "content": text_items}) + if tool_items: + messages.append({"role": role, "tool_calls": tool_items}) + return messages + elif role == "user": + if isinstance(content, str): + return [{"role": role, "content": [{"type": "text", "text": content}]}] + elif isinstance(content, list): + text_items = [] + tool_items = [] + for item in content: + if item["type"] == "text": + text_items.append({"type": "text", "text": item["text"]}) + elif item["type"] == "tool_result": + tool_items.append( + { + "role": "tool", + "tool_call_id": item["tool_use_id"], + "content": item["content"], + } + ) + messages = [] + if text_items: + messages.append({"role": role, "content": text_items}) + messages.extend(tool_items) + return messages + + def from_standard_message(self, message): + # todo: image messages (?) + if message["role"] == "tool": + return { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": message["tool_call_id"], + "content": message["content"], + }, + ], + } + if message.get("tool_calls"): + tc = message["tool_calls"] + ret = {"role": "assistant", "content": []} + for tool_call in tc: + function = tool_call["function"] + arguments = json.loads(function["arguments"]) + new_tool_use = { + "type": "tool_use", + "id": tool_call["id"], + "name": function["name"], + "input": arguments, + } + ret["content"].append(new_tool_use) + return ret + # check for empty text strings + content = message.get("content") + if isinstance(content, str): + if content == "": + content = "(empty)" + elif isinstance(content, list): + for item in content: + if item["type"] == "text" and item["text"] == "": + item["text"] = "(empty)" + + return message + def add_image_frame_message( self, *, format: str, size: tuple[int, int], image: bytes, text: str = None ): @@ -429,6 +523,14 @@ class AnthropicLLMContext(OpenAILLMContext): return self.messages def _restructure_from_openai_messages(self): + # first, map across self._messages calling self.from_standard_message(m) to modify messages in place + logger.debug("!!! mapping") + try: + self._messages[:] = [self.from_standard_message(m) for m in self._messages] + except Exception as e: + logger.error(f"Error mapping messages: {e}") + + logger.debug("!!! restructuring system thingy") # See if we should pull the system message out of our context.messages list. (For # compatibility with Open AI messages format.) if self.messages and self.messages[0]["role"] == "system":