Noesis / Integrations

LiveKit Agents

Stream the SIP callee's audio from a LiveKit agent to Noesis and act on every answer.

Stream the SIP callee's audio from your LiveKit agent to Noesis, and act on every answer. The module sends 16-bit PCM at 16 kHz in 100 ms frames, and tells Noesis when your agent speaks.

1. Add the module

Needs pip install aiohttp (LiveKit Agents already uses it).

Python
# detectif_noesis.py
import asyncio, json

import aiohttp
from livekit import rtc

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

class DetectifNoesis:
    def __init__(self, api_key, call_id, on_classification):
        self.key, self.call_id, self.on = api_key, call_id, on_classification
        self.http = self.ws = None

    async def start(self, track: rtc.Track):
        self.http = aiohttp.ClientSession()
        try:
            self.ws = await self.http.ws_connect(f"{URL}&call_id={self.call_id}",
                                                 headers={"Authorization": f"Bearer {self.key}"},
                                                 heartbeat=20)
            await self.ws.receive_json()                                  # "ready"
        except Exception:
            return                                                        # never block the call
        asyncio.create_task(self._read())
        stream = rtc.AudioStream(track, sample_rate=16000, num_channels=1, frame_size_ms=100)
        async for ev in stream:
            if self.ws.closed:
                break
            await self.ws.send_bytes(bytes(ev.frame.data))

    async def _read(self):
        async for msg in self.ws:
            if msg.type == aiohttp.WSMsgType.TEXT:
                ev = json.loads(msg.data)
                if ev.get("type") == "classification":
                    await self.on(ev)

    def agent_speech(self, state):                                        # "started" / "ended"
        if self.ws is not None and not self.ws.closed:
            msg = json.dumps({"type": "agent.speech", "state": state})
            asyncio.create_task(self.ws.send_str(msg))

    async def aclose(self):
        if self.ws is not None and not self.ws.closed:
            await self.ws.send_str(json.dumps({"type": "end"}))
            await self.ws.close()
        if self.http is not None:
            await self.http.close()

2. Wire it into your entrypoint

After the SIP participant has joined:

Python
# in your entrypoint (uses: import asyncio, os; from livekit import api, rtc)
callee = await ctx.wait_for_participant(kind=rtc.ParticipantKind.PARTICIPANT_KIND_SIP)
noesis = DetectifNoesis(os.environ["DETECTIF_API_KEY"], ctx.room.name, on_classification)
ctx.add_shutdown_callback(noesis.aclose)

@ctx.room.on("track_subscribed")
def _on_track(track, publication, participant):
    if participant.identity == callee.identity and track.kind == rtc.TrackKind.KIND_AUDIO:
        asyncio.create_task(noesis.start(track))

@session.on("agent_state_changed")
def _on_agent_state(ev):
    if ev.new_state == "speaking":
        noesis.agent_speech("started")
    elif ev.old_state == "speaking":
        noesis.agent_speech("ended")

async def on_classification(ev):
    if ev["label"] == "human":
        await session.generate_reply(instructions="Greet the caller.")
    elif ev["label"] == "voicemail":
        await session.say("Hi, this is Acme ...", allow_interruptions=False)
        await ctx.api.room.delete_room(api.DeleteRoomRequest(room=ctx.room.name))
    elif ev["label"] in ("unavailable", "announcement"):
        await ctx.api.room.delete_room(api.DeleteRoomRequest(room=ctx.room.name))
    # ivr / hold / ai_agent: stay silent and keep listening

To not record calls, add &record=false to URL.