hearing

hearing — pluggable meeting transcription & context-aware AI agents.

A small, composable toolkit for capturing, transcribing, and reasoning over meeting audio. Four swappable concerns compose into one pipeline by dependency injection: audio capture (mic + system audio on separate channels) -> STT engine facade (local or cloud) -> optional diarization / speaker-id (the mic-vs-system channel split gives “me vs them” for free) -> an agent layer that consumes the transcript (batch or live).

Progressive disclosure — the one-liner just works:

from hearing import transcribe
transcript = transcribe("meeting.wav")
print(transcript.formatted())

…while engine selection, diarization, agents, channel routing, and the live loop are all optional keyword-only arguments.

The architecture and the per-concern guidance live in the project’s agent skills (.claude/skills/hearing*); start with the hearing skill.

class hearing.Channel(value)

Which capture channel a segment came from.

The channel is the “me vs them” signal, for free: MIC is the local user, SYSTEM is everyone else (audio that came out of the speakers). MIXED is a single-channel / unknown source.

hearing.SpeakerLabel

alias of str

class hearing.TimeSpan(start_ms: int, end_ms: int)

A half-open interval [start_ms, end_ms) in integer milliseconds.

Integer time, never float seconds — accumulation-safe and hashable.

>>> TimeSpan(0, 1500).duration_ms
1500

property duration_ms: int

Length of the interval in milliseconds.

classmethod from_seconds(start: float, end: float) → TimeSpan

Build a span from float seconds (e.g. from an STT engine).

class hearing.Transcript(segments: tuple[~hearing.types.TranscriptSegment, ...]=(), sample_rate: int | None = None, meta: Mapping[str, object]=<factory>)

A finished transcript: an ordered collection of segments plus metadata.

Iterable and len-able, so it stands in for a Sequence of segments wherever the architecture’s facades return “segments”. Adds convenience accessors the CLI/agents need.

>>> t = Transcript((TranscriptSegment("a", TimeSpan(0, 1000), speaker=ME),
...                  TranscriptSegment("b", TimeSpan(1000, 2000), speaker=THEM)))
>>> t.text
'a b'
>>> len(t)
2
>>> sorted(t.speakers)
['me', 'them']

property duration_ms: int

End of the last segment, or 0 if empty.

formatted(*, speakers: bool = True, timestamps: bool = True) → str

Render as a human-readable transcript, one line per segment.

property speakers: set[str]

The distinct speaker labels present (ignoring None).

property text: str

All segment texts joined by a single space.

to_jsonable() → list[dict]

A plain list[dict] ready for json.dumps (segments only).

with_meta(**kw: object) → Transcript

Return a copy with extra metadata merged in.

class hearing.TranscriptSegment(text: str, span: TimeSpan, channel: Channel = Channel.MIXED, speaker: str | None = None, confidence: float | None = None, words: tuple[~hearing.types.Word, ...]=(), meta: Mapping[str, object]=<factory>)

The data spine — a standoff interval annotation over the audio.

Required fields are text and span. Everything else is optional metadata that a concern enriches: STT sets text/span/ confidence; capture sets channel; diarization sets speaker. Segments are frozen — enrich by copying (with_speaker/with_channel), never mutate.

>>> seg = TranscriptSegment("hello", TimeSpan(0, 500))
>>> seg.with_speaker(ME).speaker
'me'
>>> seg.text
'hello'

with_channel(channel: Channel) → TranscriptSegment

Return a copy carrying a channel label (frozen -> copy, don’t mutate).

with_speaker(speaker: str) → TranscriptSegment

Return a copy carrying a speaker label (frozen -> copy, don’t mutate).

class hearing.Word(text: str, span: TimeSpan, confidence: float | None = None)

Optional word-level timing (WhisperX / word_timestamps=True).

Lets an agent trigger on a keyword mid-utterance instead of waiting for the end of a turn. Purely optional metadata on a TranscriptSegment.

async hearing.live_transcribe(*, source: CaptureSource, engine: STTEngine | None = None, diarizer: Diarizer | None = None, agent: AgentConsumer | None = None, audio_queue_max: int = 50, segment_queue_max: int = 100)

Stream FINALIZED segments as a meeting unfolds (the live path).

Same components as transcribe() — only the source (a streaming CaptureSource instead of a file) and the trigger cadence (VAD utterance turn-ends instead of run-once) change. The STTEngine and AgentConsumer interfaces are unchanged.

Each channel is demuxed onto its own bounded queue and driven through engine.stream_transcribe as an independent async task, so a slow agent never stalls capture (backpressure) and the channel “me vs them” label rides every segment. Diarization defaults to the free channel trick; the agent’s on_segment is fired fire-and-forget. Yields segments as utterances finalize.

See the hearing-live-pipeline skill.

hearing.summarize(source: str | Path | CaptureSource | Transcript, *, agent: AgentConsumer | None = None, context: str | None = None, model: str | None = None, **transcribe_kwargs) → str | None

Transcribe (if needed) then run an agent to produce meeting notes.

Accepts an audio source or an already-built Transcript. Builds the default agent (Claude if available, else the offline extractive fallback) when none is injected.

hearing.transcribe(source: str | Path | CaptureSource, *, engine: STTEngine | None = None, diarizer: Diarizer | None = None, agent: AgentConsumer | None = None, language: str | None = None, split: bool = True, mic_channels: Sequence[int] = (0,), system_channels: Sequence[int] = (1,), record=None) → Transcript

Transcribe an audio source to a Transcript.

The one-liner transcribe("meeting.wav") works with all defaults: the default local engine (faster-whisper), automatic channel splitting when the file is multi-channel, and the free “me vs them” channel-trick diarizer.

Parameters:

• source – a file path or any CaptureSource.

• engine – STT engine (default: faster-whisper). Inject any STTEngine.

• diarizer – speaker labeller. None -> the channel trick is applied automatically when channel info is present (mic=”me”, system=”them”). Inject a PyannoteDiarizer to separate individual remote speakers.

• agent – optional AgentConsumer run over the finished transcript (batch). Its output is stored under transcript.meta['insight'].

• language – force a language code (e.g. “en”); None = auto-detect.

• split – split mic/system channels when the source is multi-channel.

• system_channels (mic_channels /) – column indices for the channel split.

• record – optional MutableMapping store (e.g. a dol store) to persist the transcript JSON under key "<source>.transcript.json".

Returns:

A Transcript (iterable/len-able, so it stands in for a sequence of segments). Segments carry text, span, channel, and speaker.