Files
server-26/drb-c2-core/app/internal/vocabulary_learner.py
T
Logan CusanoandClaude Opus 5 964343c819
Build & Deploy / Build & push images (push) Successful in 4m9s
Build & Deploy / Deploy to VM (push) Successful in 1m54s
Build & Deploy / Report a failed deploy (push) Skipped
area_context v2 + Maps place verification (server-26#36, #37)
#36 — the correction pass shipped in 58efdbd was right, its reference-data
shape was not. One shape now, at both scopes, every field nullable:

  area_context: { municipality?, county?, state?,
                  center?, radius_km?, resolved_from?, resolved_at?,
                  local_knowledge?: [{term, meaning}] }

`state` closes the ambiguity that made "Ossining" a national guess.
`local_knowledge` replaces roads[]/landmarks[], which could not hold
intersections, schools or nicknames and carried no meanings — `11-X-ray` is
useless alone, `11-X-ray — MTA PD patrol unit` is what a corrector can act on.
Pre-#36 roads[]/landmarks[] are read forward as bare terms so nothing an
operator already entered is lost.

Nullability is the mechanism: which scope gets filled is the operator's
declaration of how homogeneous the system is. One town — fill it once at system
level. Statewide — leave it blank and fill each talkgroup.

The backend owns the derived anchor. PUT /systems/{id} merges config.talkgroups[]
against what is stored instead of writing the client's blob verbatim, which
would have erased the anchor and the pending queue — the same defect as the
ten_codes wipe.

#37 — Maps as a verifier, not as prompt stuffing. The corrector emits its
location nouns; each is geocoded against the talkgroup's anchor, and on a miss
we look for a sound-alike that does resolve there, correct to it, and propose
{term, meaning} to that talkgroup. Cost scales with location nouns, not calls.

No anchor means SKIP. An area too wide to discriminate stores no anchor at all,
because a statewide radius would confirm anything inside it — verification that
passes everything is worse than none, since it reads as a check in the data.

Also re-anchors _geocode_location, which rejected results >40km from the NODE
(server-26#6). An antenna is not a jurisdiction; distance-from-node was always
a stand-in for the anchor and is now only the fallback.

The induction loop proposes at talkgroup level and never promotes. Blast
radius: a wrong term on a channel misleads that channel, the same term
system-wide misleads one 400km away on a statewide system.

38 new tests; 240 pass. Frontend typechecks clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-23 16:43:59 -04:00

532 lines
22 KiB
Python

"""
Per-system vocabulary learning for STT accuracy improvement.
Three mechanisms:
1. Bootstrap — one-shot GPT-4o call generates local knowledge at system setup:
agencies + abbreviations, unit naming, streets, acronyms.
2. Correction — diffs admin transcript edits, extracts corrected tokens → vocabulary.
3. Induction — background loop samples N tokens of transcripts per system,
asks GPT-4o-mini to propose new terms → queued as pending for review.
Firestore schema additions on system documents:
vocabulary: list[str] — approved terms; injected into Whisper + GPT prompts
vocabulary_pending: list[dict] — induction suggestions awaiting admin review
each: {term, source, added_at}
vocabulary_bootstrapped: bool — bootstrap has been run at least once
"""
import asyncio
import difflib
import json
import random
import re
from datetime import datetime, timezone, timedelta
from typing import Any, Optional
from app.internal.logger import logger
from app.internal import area_context
from app.internal import firestore as fstore
from app.config import settings
# ─────────────────────────────────────────────────────────────────────────────
# Prompt templates
# ─────────────────────────────────────────────────────────────────────────────
_BOOTSTRAP_PROMPT = """\
You are building a radio vocabulary dictionary to improve speech-to-text accuracy for a P25 \
public-safety radio monitoring system in a specific area. The STT model has no local knowledge, \
so common terms like "YVAC" get transcribed as "why vac", "5-baker" as "5 acre", etc.
System name: {system_name}
System type: {system_type}
Area context: {area_hint}
Return ONLY a JSON object:
{{"vocabulary": [list of strings]}}
Include terms you are confident about for this area:
- Agency names and their radio abbreviations (e.g. "YVAC" = Yorktown Volunteer Ambulance Corps)
- Unit ID examples using the local naming convention (e.g. "5-baker", "5-charlie", "1-david";
many departments use APCO phonetics: adam, baker, charles, david, edward, frank, george,
henry, ida, john, king, lincoln, mary, nora, ocean, paul, queen, robert, sam, tom, union,
victor, william, x-ray, young, zebra)
- Major routes, roads, and key intersections
- Local landmarks and geographic references dispatchers use
- Agency-specific codes that differ from standard APCO
Return a flat list of strings — abbreviations, proper names, unit IDs, street names.
Do NOT include common English words. Max 80 terms. Only include what you are confident is \
accurate for this specific area; return fewer terms rather than guessing."""
_INDUCTION_PROMPT = """\
You are analyzing P25 emergency radio transcripts from ONE talkgroup (a single radio channel) \
to find local terms that should be added to improve future speech-to-text accuracy for that \
channel.
System: {system_name}
Channel: {talkgroup_name}
Area: {area_hint}
Terms this channel already knows (do not re-propose these): {existing_vocab}
Sampled transcripts:
{transcript_block}
Find terms that are LIKELY STT errors or local terms missing from the list:
- Unit IDs that appear garbled (e.g. "5 acre" → "5-baker")
- Agency acronyms spelled out phonetically (e.g. "why vac" → "YVAC")
- Street names or locations that look misspelled or oddly transcribed
- Callsigns or local codes not yet known
Return ONLY a JSON object:
{{"new_terms": [{{"term": "YVAC", "meaning": "Yorktown Volunteer Ambulance Corps"}}, ...]}}
`meaning` is what the term refers to — an agency, a road, a unit type. Omit it or use null \
when you genuinely do not know; a term with no meaning is still worth proposing.
Only propose what is specific to THIS channel and this area. Do not propose a term just \
because it appears often. Return {{"new_terms": []}} if nothing new is found."""
# ─────────────────────────────────────────────────────────────────────────────
# Public API
# ─────────────────────────────────────────────────────────────────────────────
async def bootstrap_system_vocabulary(system_id: str) -> list[str]:
"""
One-shot GPT-4o bootstrap: generate local-knowledge vocabulary for a system.
Merges generated terms into system.vocabulary and sets vocabulary_bootstrapped=True.
Returns the list of newly generated terms.
"""
system_doc = await fstore.doc_get("systems", system_id)
if not system_doc:
logger.warning(f"Vocabulary bootstrap: system {system_id} not found")
return []
system_name = system_doc.get("name", "Unknown")
system_type = system_doc.get("type", "P25")
# Prefer the place an operator actually set. Guessing the area from talkgroup
# names is thin for a single-municipality system and close to useless for a
# multi-county one (server-26#36), so it is only the fallback now.
area = system_doc.get("area_context") or {}
place = ", ".join(str(area[f]) for f in area_context.PLACE_FIELDS if area.get(f))
if place:
area_hint = place
else:
talkgroups = system_doc.get("config", {}).get("talkgroups", [])
tg_names = [tg.get("name", "") for tg in talkgroups if tg.get("name")][:8]
area_hint = f"Talkgroups include: {', '.join(tg_names)}" if tg_names else "Unknown area"
terms = await asyncio.to_thread(_sync_bootstrap, system_name, system_type, area_hint)
if not terms:
return []
existing = system_doc.get("vocabulary") or []
existing_lower = {t.lower() for t in existing}
to_add = [t for t in terms if t.lower() not in existing_lower]
merged = list(dict.fromkeys(existing + to_add))
await fstore.doc_set("systems", system_id, {
"vocabulary": merged,
"vocabulary_bootstrapped": True,
})
logger.info(
f"Vocabulary bootstrap: {len(to_add)} term(s) generated for system {system_id} "
f"({system_name})"
)
return to_add
async def learn_from_correction(system_id: str, original: str, corrected: str) -> None:
"""
Diff original and corrected transcripts; append new tokens to the approved vocabulary.
Called automatically when an admin saves a transcript correction.
"""
if not system_id or not original or not corrected:
return
new_terms = _diff_new_terms(original, corrected)
if not new_terms:
return
system_doc = await fstore.doc_get("systems", system_id)
if not system_doc:
return
existing = system_doc.get("vocabulary") or []
existing_lower = {t.lower() for t in existing}
to_add = [t for t in new_terms if t.lower() not in existing_lower]
if not to_add:
return
merged = list(dict.fromkeys(existing + to_add))
await fstore.doc_set("systems", system_id, {"vocabulary": merged})
logger.info(
f"Vocabulary: learned {len(to_add)} term(s) from correction on system {system_id}: "
f"{to_add}"
)
async def approve_pending_term(system_id: str, term: str) -> None:
"""Move a pending term into the approved vocabulary."""
system_doc = await fstore.doc_get("systems", system_id)
if not system_doc:
return
pending = [p for p in (system_doc.get("vocabulary_pending") or []) if p["term"] != term]
vocab = system_doc.get("vocabulary") or []
if term.lower() not in {t.lower() for t in vocab}:
vocab = list(dict.fromkeys(vocab + [term]))
await fstore.doc_set("systems", system_id, {
"vocabulary": vocab,
"vocabulary_pending": pending,
})
async def dismiss_pending_term(system_id: str, term: str) -> None:
"""Remove a pending term without adding it to vocabulary."""
system_doc = await fstore.doc_get("systems", system_id)
if not system_doc:
return
pending = [p for p in (system_doc.get("vocabulary_pending") or []) if p["term"] != term]
await fstore.doc_set("systems", system_id, {"vocabulary_pending": pending})
async def add_term(system_id: str, term: str) -> None:
"""Manually add a term to the approved vocabulary."""
system_doc = await fstore.doc_get("systems", system_id)
if not system_doc:
return
vocab = system_doc.get("vocabulary") or []
if term.lower() not in {t.lower() for t in vocab}:
vocab = list(dict.fromkeys(vocab + [term.strip()]))
await fstore.doc_set("systems", system_id, {"vocabulary": vocab})
async def remove_term(system_id: str, term: str) -> None:
"""Remove a term from the approved vocabulary."""
system_doc = await fstore.doc_get("systems", system_id)
if not system_doc:
return
vocab = [t for t in (system_doc.get("vocabulary") or []) if t.lower() != term.lower()]
await fstore.doc_set("systems", system_id, {"vocabulary": vocab})
async def get_vocabulary(system_id: str) -> dict:
"""Return vocabulary and pending terms for a system (TTL-cached, 5 min)."""
doc = await fstore.doc_get_cached("systems", system_id)
if not doc:
return {"vocabulary": [], "vocabulary_pending": [], "vocabulary_bootstrapped": False}
return {
"vocabulary": doc.get("vocabulary") or [],
"vocabulary_pending": doc.get("vocabulary_pending") or [],
"vocabulary_bootstrapped": doc.get("vocabulary_bootstrapped", False),
}
# ─────────────────────────────────────────────────────────────────────────────
# Prompt-injection helpers (called by transcription.py and intelligence.py)
# ─────────────────────────────────────────────────────────────────────────────
def build_whisper_vocab_prompt(vocabulary: list[str]) -> str:
"""
Format vocabulary for Whisper prompt injection.
Whisper's prompt field acts as a context prior with a ~224-token limit.
The base _WHISPER_PROMPT uses ~70 tokens; we budget ~150 tokens (≈550 chars) here.
"""
if not vocabulary:
return ""
char_budget = 550
terms: list[str] = []
used = 0
for term in vocabulary:
cost = len(term) + 2 # ", "
if used + cost > char_budget:
break
terms.append(term)
used += cost
return ", ".join(terms) + ". " if terms else ""
def build_gpt_vocab_block(vocabulary: list[str]) -> str:
"""Format vocabulary for injection into GPT extraction prompts."""
if not vocabulary:
return ""
return f"Known local terms: {', '.join(vocabulary)}\n"
# ─────────────────────────────────────────────────────────────────────────────
# Background induction loop
# ─────────────────────────────────────────────────────────────────────────────
async def vocabulary_induction_loop() -> None:
from app.internal.feature_flags import get_flags
interval = settings.vocabulary_induction_interval_hours * 3600
logger.info(
f"Vocabulary induction loop started — "
f"interval: {settings.vocabulary_induction_interval_hours}h, "
f"sample budget: {settings.vocabulary_induction_sample_tokens} tokens"
)
await asyncio.sleep(30) # short startup grace period before first pass
while True:
try:
flags = await get_flags()
if flags["vocabulary_learning_enabled"]:
await _run_induction_pass()
else:
logger.info("Vocabulary learning disabled — skipping induction pass")
except Exception as e:
logger.error(f"Vocabulary induction pass failed: {e}")
await asyncio.sleep(interval)
async def _run_induction_pass() -> None:
systems = await fstore.collection_list("systems")
if not systems:
return
logger.info(f"Vocabulary induction: processing {len(systems)} system(s)")
for system in systems:
system_id = system.get("system_id")
if system_id:
try:
await _induct_system(system_id, system)
except Exception as e:
logger.warning(f"Induction failed for system {system_id}: {e}")
async def _induct_system(system_id: str, system_doc: dict) -> None:
"""
Sample recent transcripts per TALKGROUP and propose local knowledge there.
Proposals used to land at system level, which is the wrong blast radius
(server-26#37). A wrong term on a talkgroup misleads one channel; the same
term at system level misleads every channel on that system — including one
400km away on a statewide system, which is exactly the context poisoning the
scope rule exists to prevent. If a term really does apply system-wide,
carrying it on several talkgroups costs almost nothing, while auto-promoting
a wrong one is expensive to notice. So: talkgroup-level pending terms only,
and nothing here ever promotes upward or approves itself.
"""
system_name = system_doc.get("name", "Unknown")
system_area = system_doc.get("area_context") or {}
# Fetch calls from the last 7 days only — avoids scanning the entire history.
# Active calls have ended_at=None and are excluded by the range filter automatically.
# Needs a composite index on (system_id ASC, ended_at ASC).
cutoff = datetime.now(timezone.utc) - timedelta(days=7)
all_calls = await fstore.collection_where("calls", [
("system_id", "==", system_id),
("ended_at", ">=", cutoff),
])
if not all_calls:
return
by_tg: dict[Any, list[dict]] = {}
for call in all_calls:
tgid = call.get("talkgroup_id")
if tgid is None:
continue
by_tg.setdefault(tgid, []).append(call)
# The sample budget is per system, split across the talkgroups that have
# traffic — a channel with 400 calls should not starve one with 12.
char_budget = max(
(settings.vocabulary_induction_sample_tokens * 4) // max(len(by_tg), 1), 800
)
for talkgroup_id, calls in by_tg.items():
try:
await _induct_talkgroup(
system_id, system_doc, system_name, system_area,
talkgroup_id, calls, char_budget,
)
except Exception as e:
logger.warning(
f"Induction failed for talkgroup {talkgroup_id} on system {system_id}: {e}"
)
async def _induct_talkgroup(
system_id: str,
system_doc: dict,
system_name: str,
system_area: dict,
talkgroup_id: Any,
calls: list[dict],
char_budget: int,
) -> None:
tg_entry = area_context.talkgroup_entry(system_doc, talkgroup_id)
tg_area = tg_entry.get("area_context") or {}
area = area_context.effective(system_area, tg_area)
talkgroup_name = (
tg_entry.get("name")
or calls[0].get("talkgroup_name")
or f"TGID {talkgroup_id}"
)
known = area_context._known_terms(tg_entry, system_doc)
random.shuffle(calls)
transcript_block = ""
sampled_call_docs: list[dict] = []
for call in calls:
text = call.get("transcript_corrected") or call.get("transcript") or ""
if not text:
continue
if len(transcript_block) + len(text) > char_budget:
break
transcript_block += f"{text}\n"
sampled_call_docs.append(call)
if len(sampled_call_docs) < 3:
return # not enough data on this channel to learn from yet
place = ", ".join(str(area[f]) for f in area_context.PLACE_FIELDS if area.get(f))
proposed = await asyncio.to_thread(
_sync_induct,
system_name, talkgroup_name, place or "not set",
sorted(known)[:80], transcript_block,
)
if not proposed:
return
entries = [
{
"term": p["term"],
"meaning": p.get("meaning"),
"source": "induction",
"source_call_ids": _find_source_calls(p["term"], sampled_call_docs),
}
for p in proposed
if p.get("term") and p["term"].lower() not in known
]
if entries:
await area_context.add_pending(system_id, talkgroup_id, entries)
# ─────────────────────────────────────────────────────────────────────────────
# Internal sync helpers
# ─────────────────────────────────────────────────────────────────────────────
def _find_source_calls(term: str, sampled_calls: list[dict], max_results: int = 3) -> list[str]:
"""
Find which sampled calls most likely produced this induction suggestion.
Splits the proposed term into tokens and searches call transcripts for overlap.
Falls back to the first two sampled calls when no token match is found
(e.g. fully garbled terms like "why vac" → "YVAC" have no word overlap).
"""
tokens = [t.lower() for t in re.split(r"[^a-zA-Z0-9]+", term) if len(t) >= 2]
matched: list[str] = []
if tokens:
for call in sampled_calls:
call_id = call.get("call_id")
if not call_id:
continue
text = (call.get("transcript_corrected") or call.get("transcript") or "").lower()
if any(tok in text for tok in tokens):
matched.append(call_id)
if len(matched) >= max_results:
break
if not matched:
matched = [c["call_id"] for c in sampled_calls[:2] if c.get("call_id")]
return matched
_STOP_WORDS = {
"the", "and", "for", "are", "was", "were", "this", "that", "with",
"have", "has", "had", "but", "not", "from", "they", "will", "what",
"can", "all", "been", "one", "two", "three", "four", "five", "six",
"you", "out", "who", "get", "her", "him", "his", "its", "our", "my",
"via", "per", "any", "now", "got", "she", "let", "did", "may", "yes",
"sir", "say", "see", "too", "off", "how", "put", "set", "try", "back",
"just", "like", "into", "than", "them", "then", "some", "also", "onto",
"went", "over", "copy", "okay", "unit", "post", "road", "lane", "going",
"being", "doing", "there", "their", "about", "would", "could", "should",
"route", "north", "south", "east", "west", "avenue", "street", "drive",
}
def _diff_new_terms(original: str, corrected: str) -> list[str]:
"""
Token-level diff: find tokens in `corrected` that replaced or were inserted
relative to `original`. These are the admin's intended spellings — good
candidates for vocabulary.
"""
orig_tokens = original.split()
corr_tokens = corrected.split()
matcher = difflib.SequenceMatcher(None,
[t.lower() for t in orig_tokens],
[t.lower() for t in corr_tokens],
)
new_terms: list[str] = []
for tag, _i1, _i2, j1, j2 in matcher.get_opcodes():
if tag in ("insert", "replace"):
for tok in corr_tokens[j1:j2]:
clean = tok.strip(".,!?;:()'\"").strip("-")
if len(clean) >= 3 and clean.lower() not in _STOP_WORDS:
new_terms.append(clean)
return list(dict.fromkeys(new_terms))
def _sync_bootstrap(system_name: str, system_type: str, area_hint: str) -> list[str]:
from app.config import settings as cfg
from openai import OpenAI
if not cfg.openai_api_key:
return []
prompt = _BOOTSTRAP_PROMPT.format(
system_name=system_name,
system_type=system_type,
area_hint=area_hint,
)
try:
client = OpenAI(api_key=cfg.openai_api_key)
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
)
data = json.loads(response.choices[0].message.content)
terms = data.get("vocabulary") or []
return [str(t).strip() for t in terms if str(t).strip()]
except Exception as e:
logger.warning(f"Vocabulary bootstrap GPT call failed: {e}")
return []
def _sync_induct(
system_name: str,
talkgroup_name: str,
area_hint: str,
existing_vocab: list[str],
transcript_block: str,
) -> list[dict]:
"""Returns [{term, meaning}] — a bare string is still accepted from the model."""
from app.config import settings as cfg
from openai import OpenAI
if not cfg.openai_api_key:
return []
vocab_str = ", ".join(existing_vocab[:80]) if existing_vocab else "(none yet)"
prompt = _INDUCTION_PROMPT.format(
system_name=system_name,
talkgroup_name=talkgroup_name,
area_hint=area_hint,
existing_vocab=vocab_str,
transcript_block=transcript_block[:8000],
)
try:
client = OpenAI(api_key=cfg.openai_api_key)
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
)
data = json.loads(response.choices[0].message.content)
return area_context.normalize_local_knowledge(data.get("new_terms") or [])
except Exception as e:
logger.warning(f"Vocabulary induction GPT call failed: {e}")
return []