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

Remove instantiation of unused http client session

Browse Source 
These examples don't make any HTTP requests with `session` so there
doesn't seem be a need to create one in the first place. Probably a
copy-paste from a previous example.
This commit is contained in:
Brian Mathiyakom
2025-06-05 11:30:07 -07:00
parent 027a82dff1
commit 10bd969636
3 changed files with 148 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
examples/foundational/01b-livekit-audio.py
View File

@@ -77,37 +77,36 @@ async def configure_livekit():
async def main(): async def main():
async with aiohttp.ClientSession() as session: (url, token, room_name) = await configure_livekit()
(url, token, room_name) = await configure_livekit()
transport = LiveKitTransport( transport = LiveKitTransport(
url=url, url=url,
token=token, token=token,
room_name=room_name, room_name=room_name,
params=LiveKitParams(audio_out_enabled=True), params=LiveKitParams(audio_out_enabled=True),
) )
tts = CartesiaTTSService( tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"), api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady
) )
runner = PipelineRunner() runner = PipelineRunner()
task = PipelineTask(Pipeline([tts, transport.output()])) task = PipelineTask(Pipeline([tts, transport.output()]))
# Register an event handler so we can play the audio when the # Register an event handler so we can play the audio when the
# participant joins. # participant joins.
@transport.event_handler("on_first_participant_joined") @transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant_id): async def on_first_participant_joined(transport, participant_id):
await asyncio.sleep(1) await asyncio.sleep(1)
await task.queue_frame( await task.queue_frame(
TextFrame( TextFrame(
"Hello there! How are you doing today? Would you like to talk about the weather?" "Hello there! How are you doing today? Would you like to talk about the weather?"
)
) )
)
await runner.run(task) await runner.run(task)
if __name__ == "__main__": if __name__ == "__main__":

121
examples/foundational/41a-text-only-webrtc.py
View File

@@ -82,78 +82,77 @@ transport_params = {
async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool): async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool):
logger.info(f"Starting bot") logger.info(f"Starting bot")
# Create an HTTP session for API calls llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
async with aiohttp.ClientSession() as session:
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
messages = [ messages = [
{ {
"role": "system", "role": "system",
"content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Respond to what the user said in a creative and helpful way.", "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Respond to what the user said in a creative and helpful way.",
}, },
]
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
action_llm_append_to_messages = create_action_llm_append_to_messages(context_aggregator)
rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
rtvi.register_action(action_llm_append_to_messages)
pipeline = Pipeline(
[
transport.input(),
rtvi,
context_aggregator.user(),
llm,
transport.output(),
context_aggregator.assistant(),
] ]
)
context = OpenAILLMContext(messages) task = PipelineTask(
context_aggregator = llm.create_context_aggregator(context) pipeline,
params=PipelineParams(
allow_interruptions=True,
enable_metrics=True,
),
observers=[RTVIObserver(rtvi)],
)
action_llm_append_to_messages = create_action_llm_append_to_messages(context_aggregator) @rtvi.event_handler("on_client_ready")
rtvi = RTVIProcessor(config=RTVIConfig(config=[])) async def on_client_ready(rtvi):
rtvi.register_action(action_llm_append_to_messages) logger.info("Pipecat client ready.")
await rtvi.set_bot_ready()
pipeline = Pipeline( # This block is frontend UI specific
[ # These messages are intended for small webrtc UI to only handle text
transport.input(), # https://github.com/pipecat-ai/small-webrtc-prebuilt
rtvi, messages = {
context_aggregator.user(), "show_text_container": True,
llm, "show_video_container": False,
transport.output(), "show_debug_container": False,
context_aggregator.assistant(), }
]
)
task = PipelineTask( rtvi_frame = RTVIServerMessageFrame(data=messages)
pipeline, await task.queue_frames([rtvi_frame])
params=PipelineParams(
allow_interruptions=True,
enable_metrics=True,
),
observers=[RTVIObserver(rtvi)],
)
@rtvi.event_handler("on_client_ready") @transport.event_handler("on_client_connected")
async def on_client_ready(rtvi): async def on_client_connected(transport, client):
logger.info("Pipecat client ready.") logger.info(f"Client connected: {client}")
await rtvi.set_bot_ready() # Kick off the conversation.
await task.queue_frames([context_aggregator.user().get_context_frame()])
# This block is frontend UI specific @transport.event_handler("on_client_disconnected")
# These messages are intended for small webrtc UI to only handle text async def on_client_disconnected(transport, client):
# https://github.com/pipecat-ai/small-webrtc-prebuilt logger.info(f"Client disconnected")
messages = {
"show_text_container": True,
"show_video_container": False,
"show_debug_container": False,
}
rtvi_frame = RTVIServerMessageFrame(data=messages)
await task.queue_frames([rtvi_frame])
@transport.event_handler("on_client_connected") @transport.event_handler("on_client_closed")
async def on_client_connected(transport, client): async def on_client_closed(transport, client):
logger.info(f"Client connected: {client}") logger.info(f"Client closed connection")
# Kick off the conversation. await task.cancel()
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected") runner = PipelineRunner(handle_sigint=False)
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
@transport.event_handler("on_client_closed") await runner.run(task)
async def on_client_closed(transport, client):
logger.info(f"Client closed connection")
await task.cancel()
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)
if __name__ == "__main__": if __name__ == "__main__":

132
examples/foundational/41b-text-and-audio-webrtc.py
View File

@@ -92,85 +92,83 @@ transport_params = {
async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool): async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool):
logger.info(f"Starting bot") logger.info(f"Starting bot")
# Create an HTTP session for API calls stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
async with aiohttp.ClientSession() as session:
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
tts = CartesiaTTSService( tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"), voice_id="71a7ad14-091c-4e8e-a314-022ece01c121" api_key=os.getenv("CARTESIA_API_KEY"), voice_id="71a7ad14-091c-4e8e-a314-022ece01c121"
) )
messages = [ messages = [
{ {
"role": "system", "role": "system",
"content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Respond to what the user says in a creative and helpful way. Explain to the User they can speak or type text to communicate with you.", "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Respond to what the user says in a creative and helpful way. Explain to the User they can speak or type text to communicate with you.",
}, },
]
context = OpenAILLMContext(messages)
context_aggregator = llm.create_context_aggregator(context)
action_llm_append_to_messages = create_action_llm_append_to_messages(context_aggregator)
rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
rtvi.register_action(action_llm_append_to_messages)
pipeline = Pipeline(
[
transport.input(),
rtvi,
stt,
context_aggregator.user(),
llm,
tts,
transport.output(),
context_aggregator.assistant(),
] ]
)
context = OpenAILLMContext(messages) task = PipelineTask(
context_aggregator = llm.create_context_aggregator(context) pipeline,
params=PipelineParams(
allow_interruptions=True,
enable_metrics=True,
),
observers=[RTVIObserver(rtvi)],
)
action_llm_append_to_messages = create_action_llm_append_to_messages(context_aggregator) @rtvi.event_handler("on_client_ready")
rtvi = RTVIProcessor(config=RTVIConfig(config=[])) async def on_client_ready(rtvi):
rtvi.register_action(action_llm_append_to_messages) logger.info("Pipecat client ready.")
await rtvi.set_bot_ready()
pipeline = Pipeline( # This block is frontend UI specific
[ # These messages are intended for small webrtc UI to only handle text
transport.input(), # https://github.com/pipecat-ai/small-webrtc-prebuilt
rtvi, messages = {
stt, "show_text_container": True,
context_aggregator.user(), "show_debug_container": False,
llm, }
tts, rtvi_frame = RTVIServerMessageFrame(data=messages)
transport.output(), await task.queue_frames([rtvi_frame])
context_aggregator.assistant(),
]
)
task = PipelineTask( @transport.event_handler("on_client_connected")
pipeline, async def on_client_connected(transport, client):
params=PipelineParams( logger.info(f"Client connected: {client}")
allow_interruptions=True, # Kick off the conversation.
enable_metrics=True, await task.queue_frames([context_aggregator.user().get_context_frame()])
),
observers=[RTVIObserver(rtvi)],
)
@rtvi.event_handler("on_client_ready") @transport.event_handler("on_client_disconnected")
async def on_client_ready(rtvi): async def on_client_disconnected(transport, client):
logger.info("Pipecat client ready.") logger.info(f"Client disconnected")
await rtvi.set_bot_ready()
# This block is frontend UI specific @transport.event_handler("on_client_closed")
# These messages are intended for small webrtc UI to only handle text async def on_client_closed(transport, client):
# https://github.com/pipecat-ai/small-webrtc-prebuilt logger.info(f"Client closed connection")
messages = { await task.cancel()
"show_text_container": True,
"show_debug_container": False,
}
rtvi_frame = RTVIServerMessageFrame(data=messages)
await task.queue_frames([rtvi_frame])
@transport.event_handler("on_client_connected") runner = PipelineRunner(handle_sigint=False)
async def on_client_connected(transport, client):
logger.info(f"Client connected: {client}")
# Kick off the conversation.
await task.queue_frames([context_aggregator.user().get_context_frame()])
@transport.event_handler("on_client_disconnected") await runner.run(task)
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
@transport.event_handler("on_client_closed")
async def on_client_closed(transport, client):
logger.info(f"Client closed connection")
await task.cancel()
runner = PipelineRunner(handle_sigint=False)
await runner.run(task)
if __name__ == "__main__": if __name__ == "__main__":