Perceptor

Contesto

Perceptor nasce come organo sensoriale real-time per un bot multimodale: un bot che guarda e ascolta una live (Twitch o YouTube) e decide quando intervenire in chat. Il problema di fondo: trasformare input destrutturati — chat testuale, audio, video — in un unico flusso di eventi strutturati e temporizzati.

L’output è un file JSONL dove ogni riga è un evento con timestamp di cattura:

{"ts":1783070000.123,"source":"chat","user":"marc3lly99K","text":"che figata questa live"}
{"ts":1783070005.991,"source":"speech","speaker_id":"spk_01","text":"questo è un test di trascrizione"}
{"ts":1783070010.441,"source":"video","caption":"The scene shows a streamer looking at the monitor."}

Architettura

Il sistema usa multiprocessing (non thread) e Redis come message broker. Ogni componente gira in un processo separato e comunica solo tramite code.

flowchart TB
  subgraph CHAT["Chat pipeline"]
      C1["Twitch IRC / YouTube chat"] --> C2["Chat Worker"] --> AGG
  end
  subgraph AUDIO["Audio pipeline"]
      A1["ffmpeg (PCM 16kHz)"] --> A2["VAD<br/>WebRTC"]
      A2 --> A3["Speaker ID<br/>sherpa-onnx"]
      A3 --> A4["ASR<br/>faster-whisper"]
      A4 --> AGG
  end
  subgraph VIDEO["Video pipeline"]
      V1["ffmpeg (frame)"] --> V2["Filtro pHash"]
      V2 --> V3["Captioning VLM<br/>(Qwen3-VL)"]
      V3 --> AGG
  end
  AGG["Aggregator<br/>(unico scrittore)"] --> OUT["data/perceptions.jsonl"]

Perché multiprocessing

Python ha il GIL (Global Interpreter Lock): con i thread, un task pesante blocca gli altri. Perceptor usa processi separati — chat, audio VAD, speaker, ASR, video, aggregator, producer — così chat, audio e video continuano in parallelo senza interferenze.

Perché Redis

Redis funge da message broker in RAM: i processi non comunicano direttamente ma tramite code (queue:chat_raw, queue:audio_raw, queue:speech_segments, …). Vantaggi: disaccoppiamento producer/worker, resilienza se un worker è più lento, monitoraggio delle code, scalabilità selettiva dei worker.

Componenti principali

• Producers: Twitch via IRC (OAuth, JOIN, parsing PRIVMSG, PING/PONG); YouTube via Data API REST; audio e video via ffmpeg (normalizzazione a PCM 16kHz mono per l’audio; frame JPEG per il video); metadata piattaforma (titolo, viewer count, categoria) direttamente su queue:perceptions.
• Audio pipeline: VAD (WebRTC, su CPU) → segmenti parlati → speaker identification (sherpa-onnx o mock, centroidi persistiti per canale) → ASR con faster-whisper (CPU/CUDA, int8/float16).
• Video pipeline: estrazione frame → filtro pHash (scarta frame visivamente identici, evitando chiamate API inutili) → captioning VLM (OpenRouter, es. Qwen3-VL) con LRU per scene già viste.
• Aggregator: l’unico componente che scrive su data/perceptions.jsonl, con consumo at-least-once, code processing/dead-letter e deduplica.

Reactor (MVP)

Sopra il Perceptor c’è il Reactor: un modulo che segue il JSONL e decide quando il bot dovrebbe intervenire in chat. Usa trigger deterministici (menzioni, domande, temi ripetuti), un FocusSelector con TTL per sorgente, un PromptBuilder con memoria breve/facts, una decisione JSON dell’LLM, guardrail anti-stale e dry-run.

Scelte progettuali

• Processi invece di thread: il GIL rende i thread inadatti al real-time su più modalità; il costo della memoria dei processi è accettabile rispetto alla latenza garantita.
• Redis come cuscinetto: se Whisper è lento, la coda cresce ma chat e video non si bloccano. Il debug si fa misurando la lunghezza delle code (documentato nel README).
• pHash prima del VLM: le chiamate VLM costano; il filtro di novità visiva le riduce drasticamente senza perdere eventi semantici.
• Configurazione via .env + profili: default operativi in profiles/*.toml, credenziali e tuning nel .env; il launcher python -m perceptor.launch twitch <canale> deriva le sorgenti dal canale.

Risultati

Un sistema funzionante end-to-end che acquisisce live reali (Twitch/YouTube/file locali), produce un flusso JSONL sincronizzato e alimenta un bot che interviene in chat in modo controllato. Il codice è privato, ma l’architettura completa è documentata nel README e posso presentarla in dettaglio.

Note

Roadmap annotate nel progetto: migrare le code a Redis Streams/consumer group, metriche Prometheus su latenza e drop, diarization offline, replay di sessioni anonimizzate.