| """Pipecat + Simli bot. Run: python pipecat-simpli.py -t webrtc then open http://localhost:7860 |
| In Playground: enable the bot video (camera) in DEVICES so the avatar is visible.""" |
| import os |
|
|
| from dotenv import load_dotenv |
| from loguru import logger |
|
|
| 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.processors.aggregators.openai_llm_context import OpenAILLMContext |
| from pipecat.runner.types import RunnerArguments |
| from pipecat.runner.utils import create_transport |
| from pipecat.services.cartesia.tts import CartesiaTTSService |
| from pipecat.services.deepgram.stt import DeepgramSTTService |
| from pipecat.services.openai.llm import OpenAILLMService |
| from pipecat.services.simli.video import SimliVideoService |
| from pipecat.transports.base_transport import BaseTransport, TransportParams |
|
|
| load_dotenv(override=True) |
|
|
| _REQUIRED_ENV = ["OPENAI_API_KEY", "DEEPGRAM_API_KEY", "CARTESIA_API_KEY", "CARTESIA_VOICE_ID", "SIMLI_API_KEY", "SIMLI_FACE_ID"] |
| _missing = [k for k in _REQUIRED_ENV if not os.getenv(k)] |
| if _missing: |
| raise SystemExit(f"Missing .env keys: {', '.join(_missing)}. Add them to .env and restart.") |
|
|
|
|
| def _daily_params(): |
| from pipecat.transports.services.daily import DailyParams |
| return DailyParams( |
| audio_in_enabled=True, |
| audio_out_enabled=True, |
| video_out_enabled=True, |
| video_out_is_live=True, |
| video_out_width=512, |
| video_out_height=512, |
| vad_analyzer=SileroVADAnalyzer(), |
| ) |
|
|
|
|
| transport_params = { |
| "daily": _daily_params, |
| "webrtc": lambda: TransportParams( |
| audio_in_enabled=True, |
| audio_out_enabled=True, |
| video_out_enabled=True, |
| video_out_is_live=True, |
| video_out_width=512, |
| video_out_height=512, |
| vad_analyzer=SileroVADAnalyzer(), |
| ), |
| } |
|
|
|
|
| async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): |
| logger.info("Starting bot") |
|
|
| stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) |
|
|
| tts = CartesiaTTSService( |
| api_key=os.getenv("CARTESIA_API_KEY"), |
| voice_id=os.getenv("CARTESIA_VOICE_ID"), |
| ) |
| simli_ai = SimliVideoService( |
| api_key=os.getenv("SIMLI_API_KEY"), |
| face_id=os.getenv("SIMLI_FACE_ID"), |
| params=SimliVideoService.InputParams( |
| max_session_length=3600, |
| max_idle_time=300, |
| ), |
| ) |
|
|
|
|
| llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini") |
|
|
| messages = [ |
| { |
| "role": "system", |
| "content": "You are a helpful assistant in a voice call. As soon as the call starts, say one short greeting (e.g. 'Hi, how can I help you today?') in a single sentence, then answer briefly. Keep responses concise.", |
| }, |
| {"role": "user", "content": "Say a brief greeting to start the call."}, |
| ] |
|
|
| context = OpenAILLMContext(messages) |
| context_aggregator = llm.create_context_aggregator(context) |
|
|
| pipeline = Pipeline( |
| [ |
| transport.input(), |
| stt, |
| context_aggregator.user(), |
| llm, |
| tts, |
| simli_ai, |
| transport.output(), |
| context_aggregator.assistant(), |
| ] |
| ) |
|
|
| task = PipelineTask( |
| pipeline, |
| params=PipelineParams( |
| enable_metrics=True, |
| enable_usage_metrics=True, |
| ), |
| idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, |
| ) |
|
|
| @transport.event_handler("on_client_connected") |
| async def on_client_connected(transport, client): |
| logger.info(f"Client connected") |
| await task.queue_frames([context_aggregator.user().get_context_frame()]) |
|
|
| @transport.event_handler("on_client_disconnected") |
| async def on_client_disconnected(transport, client): |
| logger.info(f"Client disconnected") |
| await task.cancel() |
|
|
| runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) |
| await runner.run(task) |
|
|
|
|
| async def bot(runner_args: RunnerArguments): |
| """Main bot entry point compatible with Pipecat Cloud.""" |
| transport = await create_transport(runner_args, transport_params) |
| await run_bot(transport, runner_args) |
|
|
|
|
| if __name__ == "__main__": |
| from pipecat.runner.run import main |
|
|
| main() |
