integrated simli
This commit is contained in:
Waleed
141
examples/foundational/25-simli-layer.py
Normal file
141
examples/foundational/25-simli-layer.py
Normal file
@@ -0,0 +1,141 @@
|
||||
import asyncio
|
||||
import aiohttp
|
||||
import os
|
||||
import sys
|
||||
|
||||
from pipecat.audio.vad.silero import SileroVADAnalyzer
|
||||
from pipecat.pipeline.pipeline import Pipeline
|
||||
from pipecat.pipeline.runner import PipelineRunner
|
||||
from pipecat.pipeline.task import PipelineParams, PipelineTask
|
||||
from pipecat.frames.frames import LLMMessagesFrame
|
||||
|
||||
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
|
||||
from pipecat.services.ai_services import AIService
|
||||
from pipecat.services.cartesia import CartesiaTTSService
|
||||
from pipecat.services.elevenlabs import ElevenLabsTTSService
|
||||
from pipecat.services.openai import OpenAILLMService
|
||||
from pipecat.transports.services.daily import DailyParams, DailyTransport
|
||||
|
||||
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper, DailyRoomParams
|
||||
from runner import configure
|
||||
from loguru import logger
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from simli import SimliConfig
|
||||
from pipecat.services.simli import SimliVideoService
|
||||
|
||||
load_dotenv(override=True)
|
||||
|
||||
logger.remove(0)
|
||||
logger.add(sys.stderr, level="DEBUG")
|
||||
|
||||
|
||||
async def main():
|
||||
async with aiohttp.ClientSession() as session:
|
||||
(room_url, token) = await configure(session)
|
||||
|
||||
print("Creating room")
|
||||
aiohttp_session = aiohttp.ClientSession()
|
||||
daily_helper = DailyRESTHelper(
|
||||
daily_api_key=os.getenv("DAILY_API_KEY", ""),
|
||||
daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
|
||||
aiohttp_session=aiohttp_session,
|
||||
)
|
||||
|
||||
room = await daily_helper.create_room(DailyRoomParams())
|
||||
expiry_time: float = 60 * 60
|
||||
|
||||
token = await daily_helper.get_token(room.url, expiry_time)
|
||||
|
||||
|
||||
print("Room created ", room.url)
|
||||
|
||||
transport = DailyTransport(
|
||||
room.url,
|
||||
token,
|
||||
"Chatbot",
|
||||
DailyParams(
|
||||
audio_out_enabled=True,
|
||||
camera_out_enabled=True,
|
||||
camera_out_width=512,
|
||||
camera_out_height=512,
|
||||
vad_enabled=True,
|
||||
vad_analyzer=SileroVADAnalyzer(),
|
||||
transcription_enabled=True,
|
||||
#
|
||||
# Spanish
|
||||
#
|
||||
# transcription_settings=DailyTranscriptionSettings(
|
||||
# language="es",
|
||||
# tier="nova",
|
||||
# model="2-general"
|
||||
# )
|
||||
),
|
||||
)
|
||||
|
||||
# tts = ElevenLabsTTSService(
|
||||
# api_key=os.getenv("ELEVENLABS_API_KEY"),
|
||||
# voice_id="pNInz6obpgDQGcFmaJgB",
|
||||
# )
|
||||
|
||||
tts = CartesiaTTSService(
|
||||
api_key=os.getenv("CARTESIA"),
|
||||
voice_id="a167e0f3-df7e-4d52-a9c3-f949145efdab",
|
||||
)
|
||||
|
||||
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini")
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "system",
|
||||
#
|
||||
# English
|
||||
#
|
||||
"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.",
|
||||
#
|
||||
# Spanish
|
||||
#
|
||||
# "content": "Eres Chatbot, un amigable y útil robot. Tu objetivo es demostrar tus capacidades de una manera breve. Tus respuestas se convertiran a audio así que nunca no debes incluir caracteres especiales. Contesta a lo que el usuario pregunte de una manera creativa, útil y breve. Empieza por presentarte a ti mismo.",
|
||||
},
|
||||
]
|
||||
simliAi = SimliVideoService(
|
||||
SimliConfig(os.getenv("SIMLI_API_KEY"), os.getenv("SIMLI_FACE_ID"))
|
||||
)
|
||||
print("starting connection to simi")
|
||||
await simliAi.startConnection()
|
||||
print("connection started")
|
||||
|
||||
context = OpenAILLMContext(messages)
|
||||
context_aggregator = llm.create_context_aggregator(context)
|
||||
|
||||
pipeline = Pipeline(
|
||||
[
|
||||
transport.input(),
|
||||
context_aggregator.user(),
|
||||
llm,
|
||||
tts,
|
||||
simliAi,
|
||||
transport.output(),
|
||||
context_aggregator.assistant(),
|
||||
]
|
||||
)
|
||||
|
||||
task = PipelineTask(
|
||||
pipeline,
|
||||
PipelineParams(
|
||||
allow_interruptions=True,
|
||||
enable_metrics=True,
|
||||
),
|
||||
)
|
||||
|
||||
@transport.event_handler("on_first_participant_joined")
|
||||
async def on_first_participant_joined(transport, participant):
|
||||
await transport.capture_participant_transcription(participant["id"])
|
||||
await task.queue_frames([LLMMessagesFrame(messages)])
|
||||
|
||||
runner = PipelineRunner()
|
||||
await runner.run(task)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -66,7 +66,7 @@ soundfile = [ "soundfile~=0.12.1" ]
|
||||
together = [ "openai~=1.50.2" ]
|
||||
websocket = [ "websockets~=13.1", "fastapi~=0.115.0" ]
|
||||
whisper = [ "faster-whisper~=1.0.3" ]
|
||||
|
||||
simli = [ "simli-ai>=0.1.7"]
|
||||
[tool.setuptools.packages.find]
|
||||
# All the following settings are optional:
|
||||
where = ["src"]
|
||||
|
||||
116
src/pipecat/services/simli.py
Normal file
116
src/pipecat/services/simli.py
Normal file
@@ -0,0 +1,116 @@
|
||||
import asyncio
|
||||
|
||||
from pipecat.frames.frames import (
|
||||
Frame,
|
||||
OutputImageRawFrame,
|
||||
TTSAudioRawFrame,
|
||||
StartInterruptionFrame,
|
||||
EndFrame,
|
||||
CancelFrame,
|
||||
)
|
||||
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
|
||||
|
||||
import numpy as np
|
||||
from av import AudioFrame
|
||||
from av.audio.resampler import AudioResampler
|
||||
from simli import SimliClient, SimliConfig
|
||||
|
||||
|
||||
class SimliVideoService(FrameProcessor):
|
||||
def __init__(
|
||||
self, simliConfig: SimliConfig, useTurnServer=False, latencyInterval=60
|
||||
):
|
||||
super().__init__()
|
||||
self.simliClient = SimliClient(simliConfig, useTurnServer, latencyInterval)
|
||||
|
||||
self.pipecatResampler: AudioResampler = None
|
||||
self.name = "SimliAi"
|
||||
self.ready = False
|
||||
self.simliResampler = AudioResampler("s16", 1, 16000)
|
||||
self.AudioTask: asyncio.Task = None
|
||||
self.VideoTask: asyncio.Task = None
|
||||
|
||||
async def startConnection(self):
|
||||
await self.simliClient.Initialize()
|
||||
self.ready = True
|
||||
# Create task to consume and process audio and video
|
||||
self.AudioTask = asyncio.create_task(self.consume_and_process_audio())
|
||||
self.VideoTask = asyncio.create_task(self.consume_and_process_video())
|
||||
|
||||
async def consume_and_process_audio(self):
|
||||
async for audio_frame in self.simliClient.getAudioStreamIterator():
|
||||
# Process the audio frame
|
||||
try:
|
||||
resampledFrames = self.pipecatResampler.resample(audio_frame)
|
||||
for resampled_frame in resampledFrames:
|
||||
await self.push_frame(
|
||||
TTSAudioRawFrame(
|
||||
audio=resampled_frame.to_ndarray().tobytes(),
|
||||
sample_rate=self.pipecatResampler.rate,
|
||||
num_channels=1,
|
||||
),
|
||||
)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
import traceback
|
||||
|
||||
traceback.print_exc()
|
||||
|
||||
async def consume_and_process_video(self):
|
||||
async for video_frame in self.simliClient.getVideoStreamIterator(
|
||||
targetFormat="rgb24"
|
||||
):
|
||||
# Process the video frame
|
||||
convertedFrame: OutputImageRawFrame = OutputImageRawFrame(
|
||||
image=video_frame.to_rgb().to_image().tobytes(),
|
||||
size=(video_frame.width, video_frame.height),
|
||||
format="RGB",
|
||||
)
|
||||
convertedFrame.pts = video_frame.pts
|
||||
await self.push_frame(
|
||||
convertedFrame,
|
||||
)
|
||||
|
||||
async def process_frame(self, frame: Frame, direction: FrameDirection):
|
||||
await super().process_frame(frame, direction)
|
||||
|
||||
if isinstance(frame, TTSAudioRawFrame):
|
||||
# Send audio frame to Simli
|
||||
try:
|
||||
if self.ready:
|
||||
AudioFrame
|
||||
oldFrame = AudioFrame.from_ndarray(
|
||||
np.frombuffer(frame.audio, dtype=np.int16)[None, :],
|
||||
layout=frame.num_channels,
|
||||
)
|
||||
oldFrame.sample_rate = frame.sample_rate
|
||||
if self.pipecatResampler is None:
|
||||
self.pipecatResampler = AudioResampler(
|
||||
"s16", oldFrame.layout, oldFrame.sample_rate
|
||||
)
|
||||
|
||||
resampledFrame = self.simliResampler.resample(oldFrame)
|
||||
for frame in resampledFrame:
|
||||
await self.simliClient.send(
|
||||
frame.to_ndarray().astype(np.int16).tobytes()
|
||||
)
|
||||
return
|
||||
else:
|
||||
print(
|
||||
"Simli Connection is not Initialized properly, passing audio to next processor"
|
||||
)
|
||||
await self.push_frame(frame, direction)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
import traceback
|
||||
|
||||
traceback.print_exc()
|
||||
elif isinstance(frame, (EndFrame, CancelFrame)):
|
||||
await self.simliClient.stop()
|
||||
self.AudioTask.cancel()
|
||||
self.VideoTask.cancel()
|
||||
|
||||
elif isinstance(frame, StartInterruptionFrame):
|
||||
await self.simliClient.clearBuffer()
|
||||
|
||||
await self.push_frame(frame, direction)
|
||||
Reference in New Issue
Block a user