| name | hearing-architecture |
| description | Use when designing, wiring, or modifying the hearing project's pipeline architecture — the SSOT for HOW the four layers (audio capture, STT, diarization/speaker-ID, agents) compose by dependency injection. Triggers on capture source, channel split, mic vs system audio, STT engine facade, transcribe(audio)->segments, streaming transcription, diarization interface, speaker label, TranscriptSegment, agent consumer, pipeline assembly, batch vs live, recipe-style composition, swap/substitute a component, Protocol/dataclass contracts, asyncio.Queue between stages, backpressure/decoupling, progressive disclosure of the pipeline, and any "how do the hearing-* skills fit together" or "what's the shared data model" question. This is the contract every other hearing-* skill (hearing-audio-capture, hearing-stt, hearing-diarization, hearing-agents) implements against. |
hearing Architecture
The SSOT for how the four layers of the hearing project wire together. Read this before touching any pipeline code; sibling skills ([[hearing-audio-capture]], [[hearing-stt]], [[hearing-diarization]], [[hearing-agents]]) implement the contracts defined here.
The core abstraction: four separable concerns, one data spine
The pipeline decomposes into four swappable concerns, each behind a clear interface [DIY]:
capture ──audio frames──▶ STT ──segments──▶ diarization ──labeled segments──▶ agents
(mic+system, (pluggable (who spoke (consume
channel-split) facade) when/who) the stream)
The one non-negotiable principle: the only difference between batch and live is (a) the source (a file vs a streaming sink) and (b) the trigger cadence (run-once vs VAD/timer-driven loop). Everything downstream — the segment shape, the diarization interface, the agent contract — is identical. Do not build two pipelines. Build one pipeline parameterized by its source and trigger.
| Source | Trigger | STT call | Agent cadence |
|---|
| Batch | meeting.wav (file) | run once | transcribe(audio) | after full transcript (or per segment) |
| Live | streaming capture sink | VAD turn-end / timer | stream_transcribe(frames) | on each finalized segment |
Everything is composition + dependency injection, never baked together. A pipeline is assembled from components that satisfy the Protocols below; you add/remove/substitute one without touching the rest (see "Recipe-style assembly").
Non-negotiable design rules
- Facade each concern behind a Protocol (
typing.Protocol, structural — implementers don't subclass). The rest of the system depends on the interface, never a concrete engine. This is the whole point: swap Whisper for Deepgram, pyannote for the channel trick, Claude for a local LLM, by changing one line of assembly.
TranscriptSegment is the data spine. Every concern speaks it. It is a frozen dataclass; it is standoff interval annotation (see annotation-systems) — segments reference the audio by time, they don't contain it.- Rational/integer time, never bare floats. Times are integer milliseconds (or
(value, rate) ticks), not float seconds. Float time accumulates error and breaks equality/round-trips. This is a day-one commitment (annotation-systems rule 2).
- Channel is first-class, not an afterthought. Mic and system audio are separate channels and stay separate as far downstream as possible — the channel itself tells you "me vs them" for free (the channel trick, below), which is more reliable than voice fingerprinting [DIY].
- Stages are decoupled async tasks joined by queues. Capture, STT, diarization, and agent run as separate
asyncio tasks communicating over asyncio.Queue. A slow LLM call must never stall audio capture (backpressure/decoupling) [DIY]. This is the idiomatic-Python equivalent of Pipecat's frame pipeline and LiveKit's room model [GD] — synthesize it, don't import a framework.
- Progressive disclosure.
transcribe("meeting.wav") must work as a one-liner with smart defaults. Channel routing, engine selection, recording, and the live loop are all optional keyword-only arguments. Simple things simple; complex things possible.
- Keyword-only beyond the source. The first positional arg is the thing being transcribed (path/source). Everything else (
engine=, diarizer=, agent=, channels=, ...) is keyword-only. No magic numbers — cadence, thresholds, chunk sizes live in config dataclasses or kwargs.
The shared data model (SSOT — every hearing-* skill imports these)
Define these in hearing/types.py (module docstring required). They are the contract.
"""Shared data model for the hearing pipeline: the TranscriptSegment spine,
channel/speaker labels, and the time representation every concern speaks."""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Mapping, Optional
class Channel(str, Enum):
"""Which capture channel a segment came from. The channel *is* the
'me vs them' signal: MIC = the local user, SYSTEM = remote participants."""
MIC = "mic"
SYSTEM = "system"
MIXED = "mixed"
SpeakerLabel = str
ME: SpeakerLabel = "me"
@dataclass(frozen=True, slots=True)
class TimeSpan:
"""A half-open interval [start_ms, end_ms) in integer milliseconds.
Integer time, never float seconds — accumulation-safe and hashable."""
start_ms: int
end_ms: int
@property
def duration_ms(self) -> int:
return self.end_ms - self.start_ms
@dataclass(frozen=True, slots=True)
class TranscriptSegment:
"""The data spine. A standoff interval annotation over the audio:
it references the audio by `span` (it does not carry the samples).
Required: text + span. Everything else is optional metadata that a
concern *enriches* — STT sets text/span/confidence; diarization sets
speaker; capture sets channel. Pass it through the pipeline, don't rebuild it.
"""
text: str
span: TimeSpan
channel: Channel = Channel.MIXED
speaker: Optional[SpeakerLabel] = None
confidence: Optional[float] = None
words: tuple["Word", ...] = ()
meta: Mapping[str, object] = field(default_factory=dict)
def with_speaker(self, speaker: SpeakerLabel) -> "TranscriptSegment":
"""Return a copy carrying a speaker label (frozen → copy, don't mutate)."""
from dataclasses import replace
return replace(self, speaker=speaker)
@dataclass(frozen=True, slots=True)
class Word:
"""Optional word-level timing (WhisperX / word_timestamps=True). Lets the
agent trigger on a keyword mid-utterance instead of waiting for turn-end."""
text: str
span: TimeSpan
confidence: Optional[float] = None
Why these choices: frozen + slots=True makes segments cheap, hashable, and safe to fan out to multiple consumers; with_speaker/dataclasses.replace is the enrich-by-copy pattern; meta carries provenance (which engine produced this) per annotation-systems rule 5. Storage of segments/recordings is out of scope here — use dol stores (see python-storage); transcripts persist as standoff annotations, never floats.
The four facade Protocols (the contracts)
Define in hearing/interfaces.py. Implementers satisfy these structurally — they never import or subclass them.
"""Facade Protocols for the four pipeline concerns. Every engine/component
is dependency-injected as one of these; the pipeline depends only on these."""
from __future__ import annotations
from typing import Protocol, Iterable, Iterator, AsyncIterator, Sequence, Optional
import numpy as np
from hearing.types import TranscriptSegment, Channel
class CaptureSource(Protocol):
"""Audio source. The ONLY thing that differs batch vs live.
Yields (channel, frames) so mic and system stay on separate channels
all the way down. Batch impl reads a file; live impl reads a device/tap."""
sample_rate: int
def frames(self) -> Iterator[tuple[Channel, np.ndarray]]:
"""Batch / sync: yield (channel, frames[n_samples]) blocks to EOF."""
...
def astream(self) -> AsyncIterator[tuple[Channel, np.ndarray]]:
"""Live / async: yield (channel, frames) blocks until stopped."""
...
class STTEngine(Protocol):
"""Speech-to-text. `transcribe` is the batch one-liner; `stream_transcribe`
is the live variant. Both yield TranscriptSegments — same shape, same spine."""
def transcribe(
self, audio: np.ndarray, *, sample_rate: int, language: Optional[str] = None
) -> Sequence[TranscriptSegment]:
"""Batch: whole clip in, segments out."""
...
async def stream_transcribe(
self, frames: AsyncIterator[np.ndarray], *, sample_rate: int
) -> AsyncIterator[TranscriptSegment]:
"""Live: yields interim then finalized segments as audio arrives.
Mark interim vs final via meta (e.g. meta['final'])."""
...
class Diarizer(Protocol):
"""Labels who spoke. Enriches segments with `.speaker`. May use audio
(pyannote) or just the channel (the channel trick). Identity resolution
(spk_0 -> 'Alice') is an optional further step, same interface."""
def assign_speakers(
self,
segments: Iterable[TranscriptSegment],
*,
audio: Optional[np.ndarray] = None,
sample_rate: Optional[int] = None,
) -> Iterable[TranscriptSegment]:
"""Yield segments with `.speaker` populated (enrich-by-copy)."""
...
class AgentConsumer(Protocol):
"""Consumes the (labeled) transcript stream and does things — running
notes, suggested questions, RAG over related docs. Defaults to Claude
but stays pluggable (see claude-api for model ids/pricing)."""
async def on_segment(self, segment: TranscriptSegment) -> Optional[str]:
"""Called per finalized segment (live). Return optional surfaced insight."""
...
async def on_window(
self, window: Sequence[TranscriptSegment]
) -> Optional[str]:
"""Called on a window/turn or whole transcript (batch). Return insight."""
...
These improve on the POC's Protocols (AudioRecorder/Transcriber/AIAgent in meeting_assistant.py): the POC is mic-only, returns bare str not segments, has no diarization, no channel concept, and OpenAI-baked. Keep its clean Protocol+dataclass+DI structure; fix exactly those gaps.
The channel trick (why channel is first-class)
Because capture keeps mic on one channel and system audio on another (BlackHole + Aggregate Device, or a Core Audio tap [DIY]), you know "who's local" for free: the MIC channel = you (ME), the SYSTEM channel = remote participants. So:
- Tag every segment with its
Channel at capture time.
- Treat the MIC channel as a known speaker (
speaker=ME) — no diarization needed.
- Run real diarization only on the SYSTEM channel to separate remote speakers.
This is simpler and more reliable than voice fingerprinting/enrollment [DIY]. A channel-aware Diarizer implementation is the default; pyannote-based diarization is the upgrade. See [[hearing-audio-capture]] (channel split) and [[hearing-diarization]].
Recipe-style assembly (composition + DI)
The pipeline is a thin orchestrator that takes the four components as injected dependencies. Swapping one touches nothing else.
"""Pipeline assembly: compose the four concerns by dependency injection.
Progressive disclosure — transcribe(path) just works; everything else is kwargs."""
from hearing import transcribe
segments = transcribe("meeting.wav")
from hearing.engines import DeepgramSTT
segments = transcribe("meeting.wav", engine=DeepgramSTT())
from hearing.diarize import PyannoteDiarizer
from hearing.agents import ClaudeAgent
segments = transcribe(
"meeting.wav",
engine=DeepgramSTT(),
diarizer=PyannoteDiarizer(),
agent=ClaudeAgent(),
)
from hearing import live_transcribe
from hearing.capture import ChannelSplitCapture
async for segment in live_transcribe(
source=ChannelSplitCapture(),
engine=DeepgramSTT(),
diarizer=ChannelTrickDiarizer(),
agent=ClaudeAgent(),
trigger="vad",
):
...
transcribe/live_transcribe are the facades. Their signature is transcribe(source, *, engine=DefaultSTT(), diarizer=None, agent=None, record=None, language=None) — first positional is the source, the rest keyword-only with smart defaults (rule 7). record= (a dol store) optionally persists audio/segments; None means don't.
The orchestrator shape (batch and live share it)
async def _run_pipeline(source, *, engine, diarizer, agent, trigger):
"""One pipeline. Batch = run source.frames() once; live = loop source.astream()
with VAD/timer. Stages are decoupled async tasks joined by queues so a slow
agent never stalls capture (backpressure)."""
seg_q: asyncio.Queue[TranscriptSegment] = asyncio.Queue()
async def capture_and_stt():
async for channel, frames in source.astream():
async for seg in engine.stream_transcribe(_frames_of(channel), sample_rate=source.sample_rate):
if seg.meta.get("final"):
await seg_q.put(seg.with_channel(channel))
async def label_and_consume():
while True:
seg = await seg_q.get()
seg = next(iter(diarizer.assign_speakers([seg]))) if diarizer else seg
if agent:
asyncio.create_task(agent.on_segment(seg))
yield seg
...
The generator/streaming idioms (yield finalized segments, fan out, fire-and-forget the agent) follow python-iterables. Act on finalized segments (after VAD turn-end) to avoid churning on partial text; word-level timestamps let the agent trigger on a keyword mid-utterance [DIY].
How to add / remove / substitute a component
| Want to... | Do this | Touches |
|---|
| Use a different STT engine | write a class satisfying STTEngine, pass engine= | one assembly line |
| Run fully local (no cloud) | inject a local STTEngine (faster-whisper) + ChannelTrickDiarizer | assembly only |
| Add speaker diarization | pass diarizer=PyannoteDiarizer() | assembly only |
| Drop the agent | pass agent=None (the default) | nothing else |
| Swap Claude for a local LLM | write an AgentConsumer, pass agent= | one line (see claude-api for the default) |
| Go batch → live | change source + add trigger="vad"; reuse engine/diarizer/agent | source + trigger only |
| Persist transcripts/recordings | pass record=<dol store> | assembly only (python-storage) |
If a change forces edits in more than one concern, the interface is wrong — fix the Protocol, don't leak a concrete type across the boundary.
Interfaces to the outside (CLI / HTTP / UI)
The Python facades (transcribe, live_transcribe) are the SSOT. Surface them, don't reimplement them:
- CLI via
argh: transcribe → hearing transcribe meeting.wav --engine deepgram. See python-dispatching.
- HTTP / UI: wrap the same facades; the frontend is TS (Zod/zustand). Keep display concerns out of the data model (annotation-systems pitfall 5).
What this skill does NOT decide (delegate)
- Engine internals, model choice, WER/latency tradeoffs → [[hearing-stt]] (and claude-api for the agent LLM).
- macOS capture mechanics (BlackHole, Aggregate Device, Core Audio taps,
sounddevice channel slicing) → [[hearing-audio-capture]].
- Diarization algorithms / pyannote / identity enrollment → [[hearing-diarization]].
- Agent behaviors / RAG / tool use / prompts → [[hearing-agents]] + claude-api.
- Transcript & audio persistence (dol stores, standoff annotation storage) → python-storage + annotation-systems.
- Find existing local packages before reaching for PyPI → my-packages / local-package-ecosystem.
References
Source extracts (with citation markers and URLs) in references/from-research-reports.md. Key external repos/docs in references/key-links.md. [DIY] = the DIY macOS pipeline report; [GD] = the AI-agents architecture (GD) report — both in misc/docs/.
