> For AI agents: the documentation index is at https://platform.detectif.ai/llms.txt. Every page is also available as Markdown by adding `.md` to its path.

# 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](https://platform.detectif.ai/docs/noesis/stream-protocol/#audio). To not record calls, add `&record=false` to `URL`.