correlator: LLM tier reads the scene transcript, not the whole call (#102) #112

Merged
logan merged 2 commits from fix/llm-correlator-scene-transcript into main 2026-09-07 04:40:35 -04:00
7 changed files with 143 additions and 6 deletions
@@ -670,6 +670,7 @@ async def correlate_call(
reassignment: bool = False, reassignment: bool = False,
embedding: Optional[list] = None, embedding: Optional[list] = None,
severity: Optional[str] = None, severity: Optional[str] = None,
transcript: Optional[str] = None,
) -> Optional[str]: ) -> Optional[str]:
""" """
Link call_id to an existing incident or create a new one. Link call_id to an existing incident or create a new one.
@@ -686,7 +687,7 @@ async def correlate_call(
system_id=system_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name, system_id=system_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name,
tags=tags, incident_type=incident_type, location=location, tags=tags, incident_type=incident_type, location=location,
reassignment=reassignment, create_if_new=create_if_new, reassignment=reassignment, create_if_new=create_if_new,
embedding=embedding, severity=severity, embedding=embedding, severity=severity, transcript=transcript,
) )
decision = _run_decision(ctx) decision = _run_decision(ctx)
return await _apply_and_log(decision, ctx) return await _apply_and_log(decision, ctx)
@@ -710,6 +711,7 @@ async def preview_correlation(
reassignment: bool = False, reassignment: bool = False,
embedding: Optional[list] = None, embedding: Optional[list] = None,
severity: Optional[str] = None, severity: Optional[str] = None,
transcript: Optional[str] = None,
) -> dict: ) -> dict:
""" """
Run the rules engine and return the decision WITHOUT committing to Firestore. Run the rules engine and return the decision WITHOUT committing to Firestore.
@@ -730,7 +732,7 @@ async def preview_correlation(
system_id=system_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name, system_id=system_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name,
tags=tags, incident_type=incident_type, location=location, tags=tags, incident_type=incident_type, location=location,
reassignment=reassignment, create_if_new=create_if_new, reassignment=reassignment, create_if_new=create_if_new,
embedding=embedding, severity=severity, embedding=embedding, severity=severity, transcript=transcript,
) )
decision = _run_decision(ctx) decision = _run_decision(ctx)
return {"decision": decision, "ctx": ctx} return {"decision": decision, "ctx": ctx}
@@ -765,6 +767,7 @@ async def _build_context(
create_if_new: bool, create_if_new: bool,
embedding: Optional[list] = None, embedding: Optional[list] = None,
severity: Optional[str] = None, severity: Optional[str] = None,
transcript: Optional[str] = None,
) -> dict: ) -> dict:
now = reference_time or datetime.now(timezone.utc) now = reference_time or datetime.now(timezone.utc)
window = timedelta(hours=settings.correlation_window_hours) window = timedelta(hours=settings.correlation_window_hours)
@@ -804,6 +807,13 @@ async def _build_context(
call_vehicles = vehicles if vehicles is not None else (call_doc.get("vehicles") or []) call_vehicles = vehicles if vehicles is not None else (call_doc.get("vehicles") or [])
call_cleared = cleared_units if cleared_units is not None else (call_doc.get("cleared_units") or []) call_cleared = cleared_units if cleared_units is not None else (call_doc.get("cleared_units") or [])
call_severity = severity or "routine" call_severity = severity or "routine"
# The transcript the LLM correlation tier reasons over. Prefer the SCENE's
# own words (server-26#102) — passed by upload.py's scene loop — and fall
# back to the call doc only when no scene text was supplied (the
# recorrelation sweep, and single-scene calls where the two are identical).
# Without this, every non-primary scene of a multi-scene call was judged by
# the LLM against a transcript containing the OTHER scenes.
scene_transcript = transcript or call_doc.get("transcript_corrected") or call_doc.get("transcript")
# A string that is not a place is not a location anywhere downstream — not # A string that is not a place is not a location anywhere downstream — not
# in the fit tests, not in the thin-call test, not in the LLM prompt, and # in the fit tests, not in the thin-call test, not in the LLM prompt, and
# not on the incident. Its coordinates go with it: coords are geocoded # not on the incident. Its coordinates go with it: coords are geocoded
@@ -826,6 +836,7 @@ async def _build_context(
return { return {
"call_id": call_id, "org_id": org_id, "all_active": all_active, "recent": recent, "call_id": call_id, "org_id": org_id, "all_active": all_active, "recent": recent,
"call_doc": call_doc, "call_embedding": call_embedding, "call_doc": call_doc, "call_embedding": call_embedding,
"scene_transcript": scene_transcript,
"call_units": call_units, "call_vehicles": call_vehicles, "call_units": call_units, "call_vehicles": call_vehicles,
"call_cleared": call_cleared, "call_severity": call_severity, "call_cleared": call_cleared, "call_severity": call_severity,
"coords": coords, "is_thin_call": is_thin_call, "now": now, "coords": coords, "is_thin_call": is_thin_call, "now": now,
+45 -2
View File
@@ -172,7 +172,7 @@ async def extract_scenes(
Each scene dict contains: Each scene dict contains:
tags, incident_type, location, location_coords, resolved, tags, incident_type, location, location_coords, resolved,
severity, vehicles, units, transcript_corrected, severity, vehicles, units, transcript, transcript_corrected,
segment_indices, embedding segment_indices, embedding
Side-effect: updates calls/{call_id} in Firestore with merged tags, Side-effect: updates calls/{call_id} in Firestore with merged tags,
@@ -337,6 +337,10 @@ async def extract_scenes(
) )
embedding = await asyncio.to_thread(_sync_embed, scene_text) embedding = await asyncio.to_thread(_sync_embed, scene_text)
scene_transcript = _scene_transcript_text(
transcript, segments, segment_indices, transcript_corrected
)
processed.append({ processed.append({
"tags": tags, "tags": tags,
"incident_type": incident_type, "incident_type": incident_type,
@@ -348,6 +352,7 @@ async def extract_scenes(
"severity": severity, "severity": severity,
"resolved": resolved, "resolved": resolved,
"reassignment": reassignment, "reassignment": reassignment,
"transcript": scene_transcript,
"transcript_corrected": transcript_corrected, "transcript_corrected": transcript_corrected,
"segment_indices": segment_indices, "segment_indices": segment_indices,
"embedding": embedding, "embedding": embedding,
@@ -571,11 +576,49 @@ def _municipality_from_tg(tg_name: Optional[str]) -> Optional[str]:
def _build_transcript_block(transcript: str, segments: Optional[list[dict]]) -> str: def _build_transcript_block(transcript: str, segments: Optional[list[dict]]) -> str:
"""Format transcript as numbered transmissions if segments are available.""" """Format transcript as numbered transmissions if segments are available."""
if segments and len(segments) > 1: if segments and len(segments) > 1:
lines = [f"{i+1}. [{s['start']}s] {s['text']}" for i, s in enumerate(segments)] # 0-based labels, matching the prompt's "0-based indices into the
# numbered transmissions" — the model echoes these back as
# `segment_indices`, which _build_scene_embed_text and the per-scene
# `transcript` (server-26#102) then slice with directly.
lines = [f"{i}. [{s['start']}s] {s['text']}" for i, s in enumerate(segments)]
return f"Transmissions ({len(segments)}):\n" + "\n".join(lines) return f"Transmissions ({len(segments)}):\n" + "\n".join(lines)
return f"Transcript:\n{transcript}" return f"Transcript:\n{transcript}"
def _scene_transcript_text(
transcript: str,
segments: Optional[list[dict]],
segment_indices: Optional[list[int]],
transcript_corrected: Optional[str],
) -> str:
"""
This scene's own words, unprefixed — the segments it owns, joined.
server-26#102: the correlator's LLM tier reads this per scene instead of
the call doc's whole-call transcript, so on a multi-scene call scene N is
no longer judged against scenes 1..N-1's text.
Never returns "". Anything that would leave the slice empty — no
`segment_indices` (a single-segment call is never numbered by
`_build_transcript_block`), or indices that are out of range / not ints —
falls back to the whole-call transcript, which for a single-scene call is
the same text and for a mis-sliced multi-scene call is at least this
call's own words. `_sync_extract`'s prompt documents 0-based indices and
`_build_transcript_block` numbers to match, so no base normalisation here.
"""
if transcript_corrected:
return transcript_corrected
if segments and segment_indices:
joined = " ".join(
segments[i]["text"]
for i in segment_indices
if isinstance(i, int) and 0 <= i < len(segments)
)
if joined:
return joined
return transcript
def _build_scene_embed_text( def _build_scene_embed_text(
transcript: str, transcript: str,
segments: Optional[list[dict]], segments: Optional[list[dict]],
+7 -1
View File
@@ -61,7 +61,13 @@ def _inc_summary(inc: dict, now: datetime) -> str:
def _call_block(ctx: dict) -> str: def _call_block(ctx: dict) -> str:
lines = [] lines = []
call_doc = ctx["call_doc"] call_doc = ctx["call_doc"]
transcript = call_doc.get("transcript_corrected") or call_doc.get("transcript") # The SCENE's own transcript, resolved in _build_context (server-26#102).
# Falls back to the call doc for a ctx built without a scene (tests, sweep).
transcript = (
ctx.get("scene_transcript")
or call_doc.get("transcript_corrected")
or call_doc.get("transcript")
)
if transcript: if transcript:
lines.append(f"Transcript: {transcript[:700]}") lines.append(f"Transcript: {transcript[:700]}")
if ctx["tags"]: if ctx["tags"]:
@@ -108,6 +108,7 @@ async def _recorrelate_orphan(call: dict) -> bool:
cleared_units = call.get("cleared_units") or [], cleared_units = call.get("cleared_units") or [],
embedding = call.get("embedding"), embedding = call.get("embedding"),
severity = call.get("severity"), severity = call.get("severity"),
transcript = call.get("transcript_corrected") or call.get("transcript"),
reference_time = started_at, # anchor window to when the call happened reference_time = started_at, # anchor window to when the call happened
create_if_new = False, # never create — link-only create_if_new = False, # never create — link-only
) )
+4 -1
View File
@@ -116,6 +116,7 @@ async def _correlate_with_consensus(
reassignment: bool = False, reassignment: bool = False,
embedding: Optional[list] = None, embedding: Optional[list] = None,
severity: Optional[str] = None, severity: Optional[str] = None,
transcript: Optional[str] = None,
) -> Optional[str]: ) -> Optional[str]:
""" """
Consensus correlator: runs the rules engine and the cheap LLM in sequence. Consensus correlator: runs the rules engine and the cheap LLM in sequence.
@@ -133,7 +134,7 @@ async def _correlate_with_consensus(
tags=tags, incident_type=incident_type, location=location, tags=tags, incident_type=incident_type, location=location,
location_coords=location_coords, units=units, vehicles=vehicles, location_coords=location_coords, units=units, vehicles=vehicles,
cleared_units=cleared_units, reassignment=reassignment, cleared_units=cleared_units, reassignment=reassignment,
embedding=embedding, severity=severity, embedding=embedding, severity=severity, transcript=transcript,
) )
ctx = preview["ctx"] ctx = preview["ctx"]
rules_decision = preview["decision"] rules_decision = preview["decision"]
@@ -226,6 +227,7 @@ async def _run_extraction_pipeline(
reassignment=is_reassignment, reassignment=is_reassignment,
embedding=scene.get("embedding"), embedding=scene.get("embedding"),
severity=scene.get("severity"), severity=scene.get("severity"),
transcript=scene.get("transcript"),
) )
if incident_id and incident_id not in incident_ids: if incident_id and incident_id not in incident_ids:
incident_ids.append(incident_id) incident_ids.append(incident_id)
@@ -343,6 +345,7 @@ async def _run_intelligence_pipeline(
reassignment=is_reassignment, reassignment=is_reassignment,
embedding=scene.get("embedding"), embedding=scene.get("embedding"),
severity=scene.get("severity"), severity=scene.get("severity"),
transcript=scene.get("transcript"),
) )
if incident_id and incident_id not in incident_ids: if incident_id and incident_id not in incident_ids:
incident_ids.append(incident_id) incident_ids.append(incident_id)
@@ -304,6 +304,39 @@ async def test_a_scene_is_judged_on_its_own_embedding_and_severity():
assert ctx["call_severity"] == "major" assert ctx["call_severity"] == "major"
@pytest.mark.asyncio
async def test_the_llm_tier_reads_the_scene_transcript_not_the_whole_call():
"""
server-26#102. intelligence.py writes only the primary scene's corrected
text to calls/{id}. _call_block (the LLM correlation prompt) must reason
over the SCENE being correlated, not a whole-call transcript that also
contains the other scenes. _build_context threads the scene's text in;
with no scene text it falls back to the call doc (sweep / single-scene).
"""
with patch("app.internal.incident_correlator.fstore") as mock_fstore:
mock_fstore.doc_get = AsyncMock(return_value={
"transcript": "scene one about a fire. scene two about a traffic stop.",
})
mock_fstore.collection_list = AsyncMock(return_value=[])
scene = await _build_context(
call_id="call-1", units=None, vehicles=None, cleared_units=None,
location_coords=None, reference_time=NOW,
system_id="sys-1", talkgroup_id=383, talkgroup_name=DISPATCH_TG,
tags=[], incident_type="police", location=None,
reassignment=False, create_if_new=True,
transcript="scene two about a traffic stop.",
)
fallback = await _build_context(
call_id="call-1", units=None, vehicles=None, cleared_units=None,
location_coords=None, reference_time=NOW,
system_id="sys-1", talkgroup_id=383, talkgroup_name=DISPATCH_TG,
tags=[], incident_type="police", location=None,
reassignment=False, create_if_new=True,
)
assert scene["scene_transcript"] == "scene two about a traffic stop."
assert fallback["scene_transcript"] == "scene one about a fire. scene two about a traffic stop."
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_a_bare_number_never_becomes_an_incident_location_or_title(): async def test_a_bare_number_never_becomes_an_incident_location_or_title():
inc = await _create(tags=["flames"], location="49", coords=None, inc = await _create(tags=["flames"], location="49", coords=None,
@@ -0,0 +1,40 @@
"""
server-26#102 — a scene is correlated on its OWN transcript, not the whole call.
_scene_transcript_text slices the segments a scene owns. It must never return
"" (an empty slice would let incident_correlator._build_context fall back to
the call doc's whole-call transcript, re-opening the leak in exactly the case
— bad indices — where it matters).
"""
from app.internal.intelligence import _scene_transcript_text
SEGS = [
{"text": "structure fire, 12 Main"},
{"text": "engine 4 responding"},
{"text": "traffic stop, plate ABC"},
{"text": "one occupant"},
]
WHOLE = "structure fire, 12 Main engine 4 responding traffic stop, plate ABC one occupant"
def test_scene_owns_a_subset_of_segments():
assert _scene_transcript_text(WHOLE, SEGS, [0, 1], None) == "structure fire, 12 Main engine 4 responding"
assert _scene_transcript_text(WHOLE, SEGS, [2, 3], None) == "traffic stop, plate ABC one occupant"
def test_corrected_text_wins_when_present():
assert _scene_transcript_text(WHOLE, SEGS, [0], "cleaned up text") == "cleaned up text"
def test_no_segment_indices_falls_back_to_whole_call():
# single-segment calls are never numbered by _build_transcript_block → null indices
assert _scene_transcript_text(WHOLE, SEGS, None, None) == WHOLE
assert _scene_transcript_text(WHOLE, None, [0, 1], None) == WHOLE
def test_out_of_range_or_nonint_indices_fall_back_never_empty():
assert _scene_transcript_text(WHOLE, SEGS, [9, 10], None) == WHOLE # all out of range
assert _scene_transcript_text(WHOLE, SEGS, ["1", "2"], None) == WHOLE # 1-based strings, rejected
assert _scene_transcript_text(WHOLE, SEGS, [-1], None) == WHOLE # negative
# partial validity: keep what's in range
assert _scene_transcript_text(WHOLE, SEGS, [3, 99], None) == "one occupant"