Noesis / Integrations

Pipecat

A Pipecat frame processor that streams the input audio to Noesis and reports every answer.

A frame processor that streams the call's input audio to Noesis, tells it when your bot speaks, and reports every answer.

1. Add the processor

Needs pip install websockets (version 14 or later).

Python
# detectif_noesis.py
import asyncio, json

import websockets
from pipecat.frames.frames import (BotStartedSpeakingFrame, BotStoppedSpeakingFrame,
                                   CancelFrame, EndFrame, InputAudioRawFrame)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor

URL = "wss://api.noesis.detectif.ai/v1/detect/stream?encoding=pcm_s16le"

class DetectifNoesis(FrameProcessor):
    def __init__(self, api_key, call_id, on_classification, **kwargs):
        super().__init__(**kwargs)
        self.key, self.call_id, self.on = api_key, call_id, on_classification
        self.ws, self.connecting = None, False

    async def _connect(self, rate, channels):
        self.connecting = True
        try:
            self.ws = await websockets.connect(
                f"{URL}&sample_rate={rate}&channels={channels}&call_id={self.call_id}",
                additional_headers={"Authorization": f"Bearer {self.key}"})
            await self.ws.recv()                                          # "ready"
            asyncio.create_task(self._read())
        except Exception:
            self.ws = None                                                # never block the call

    async def _read(self):
        async for raw in self.ws:
            ev = json.loads(raw)
            if ev.get("type") == "classification":
                await self.on(ev)

    async def _send(self, data):
        if self.ws is not None:
            try:
                await self.ws.send(data)
            except Exception:
                self.ws = None

    async def process_frame(self, frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, InputAudioRawFrame):
            if self.ws is None and not self.connecting:
                asyncio.create_task(self._connect(frame.sample_rate, frame.num_channels))
            await self._send(frame.audio)
        elif isinstance(frame, BotStartedSpeakingFrame):
            await self._send(json.dumps({"type": "agent.speech", "state": "started"}))
        elif isinstance(frame, BotStoppedSpeakingFrame):
            await self._send(json.dumps({"type": "agent.speech", "state": "ended"}))
        elif isinstance(frame, (EndFrame, CancelFrame)):
            await self._send(json.dumps({"type": "end"}))
        await self.push_frame(frame, direction)

2. Place it after the transport input

Python
noesis = DetectifNoesis(os.environ["DETECTIF_API_KEY"], call_sid, on_classification)
pipeline = Pipeline([transport.input(), noesis, stt, context_aggregator.user(), llm, tts,
                     transport.output(), context_aggregator.assistant()])

async def on_classification(ev):
    if ev["label"] == "voicemail":
        await task.queue_frames([TTSSpeakFrame("Hi, this is Acme ..."), EndFrame()])
    elif ev["label"] in ("unavailable", "announcement"):
        await task.queue_frame(EndFrame())

The processor sends the audio in the format the transport delivers: 16-bit PCM at its sample rate, which must be one of the accepted rates. To not record calls, add &record=false to URL.