Use Chunking For OpenAI API Calls

app.py

@@ -164,9 +164,10 @@ def run_complete_pipeline(
         merged_transcript=merged_transcript,
         openai_api_key=openai_api_key,
         executive_names_csv=executive_names_csv,
-        speaker_map_model="gpt-5-mini",
         cleanup_model="gpt-5",
-        timeout_seconds=1800.0,
+        timeout_seconds=600.0,
+        max_turns_per_chunk=80,
+        max_chars_per_chunk=22000,
     )
     cleaned_transcript = openai_result["cleaned_transcript"]
 

src/openai_cleanup_service.py

@@ -1,5 +1,4 @@
 import json
-import re
 from typing import Any
 
 
@@ -132,99 +131,105 @@ def _parse_executive_names(names_csv: str | None) -> list[str]:
     return deduped
 
 
-def _trim_text(value: str, max_chars: int) -> str:
-    if max_chars <= 0 or len(value) <= max_chars:
-        return value
-    return value[:max_chars]
-
-
-def _build_intro_payload(turns: list[dict[str, Any]], intro_turn_limit: int, intro_text_char_limit: int) -> list[dict[str, Any]]:
-    sampled = turns[: max(1, intro_turn_limit)]
-    payload: list[dict[str, Any]] = []
-    for idx, turn in enumerate(sampled):
-        text = str(turn.get("text", ""))
-        payload.append(
-            {
-                "turn_index": idx,
-                "speaker": turn.get("speaker"),
-                "start": turn.get("start"),
-                "end": turn.get("end"),
-                "text": _trim_text(text, intro_text_char_limit),
-            }
-        )
-    return payload
+def _build_chunk_plan(
+    turns: list[dict[str, Any]],
+    max_turns_per_chunk: int,
+    max_chars_per_chunk: int,
+) -> list[dict[str, int]]:
+    if max_turns_per_chunk <= 0:
+        max_turns_per_chunk = 1
+    if max_chars_per_chunk <= 0:
+        max_chars_per_chunk = 12000
+
+    plan: list[dict[str, int]] = []
+    n = len(turns)
+    start = 0
+    while start < n:
+        end = start
+        turns_count = 0
+        chars_count = 0
+        while end < n:
+            t = turns[end]
+            text_len = len(str(t.get("text", "")))
+            est = text_len + 60
+            if turns_count > 0 and (turns_count >= max_turns_per_chunk or chars_count + est > max_chars_per_chunk):
+                break
+            turns_count += 1
+            chars_count += est
+            end += 1
+        if end == start:
+            end = min(n, start + 1)
+        plan.append({"start": start, "end": end})
+        start = end
+    return plan
+
+
+def _normalize_final_label(final_label: str, source_label: str) -> str:
+    label = str(final_label or "").strip()
+    if not label:
+        return source_label
+    if "|" in label:
+        left = label.split("|", 1)[0].strip()
+        if left:
+            label = left
+    suffix = f"({source_label})"
+    if label.endswith(suffix):
+        label = label[: -len(suffix)].strip()
+    if not label:
+        return source_label
+    return label
+
+
+def _extract_map_updates(parsed: dict[str, Any]) -> list[dict[str, str]]:
+    candidates = parsed.get("speaker_label_map_updates")
+    if not isinstance(candidates, list):
+        candidates = parsed.get("speaker_mapping_final")
+    if not isinstance(candidates, list):
+        return []
+
+    updates: list[dict[str, str]] = []
+    for item in candidates:
+        if not isinstance(item, dict):
+            continue
+        source = str(item.get("source_label") or item.get("speaker_label") or "").strip()
+        final = str(item.get("final_label") or item.get("inferred_name") or "").strip()
+        if not source:
+            continue
+        updates.append({"source_label": source, "final_label": final})
+    return updates
 
 
-def _extract_qna_announcements(
-    turns: list[dict[str, Any]],
-    max_items: int = 80,
-    text_char_limit: int = 280,
+def _coerce_turns(
+    source_turns: list[dict[str, Any]],
+    parsed_turns: Any,
+    speaker_label_map: dict[str, str],
 ) -> list[dict[str, Any]]:
-    announcements: list[dict[str, Any]] = []
-    for idx, turn in enumerate(turns):
-        text = str(turn.get("text", "")).strip()
-        if not text:
-            continue
-        lowered = text.lower()
-        if "line of" in lowered and ("please go ahead" in lowered or "question" in lowered):
-            announcements.append(
-                {
-                    "turn_index": idx,
-                    "speaker": turn.get("speaker"),
-                    "text": _trim_text(text, text_char_limit),
-                }
-            )
-        if len(announcements) >= max_items:
-            break
-    return announcements
-
-
-def _extract_qna_name_candidates(qna_announcements: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    patterns = [
-        r"line of\s+(.+?)\s+from\s+(.+?)(?:\.|,|please go ahead|$)",
-        r"question (?:comes|is)\s+from\s+the line of\s+(.+?)\s+from\s+(.+?)(?:\.|,|please go ahead|$)",
-        r"question (?:comes|is)\s+(.+?)\s+from\s+(.+?)(?:\.|,|please go ahead|$)",
-    ]
     out: list[dict[str, Any]] = []
-    seen = set()
-    for item in qna_announcements:
-        text = str(item.get("text", ""))
-        lowered = text.lower()
-        for p in patterns:
-            match = re.search(p, lowered, flags=re.IGNORECASE)
-            if not match:
-                continue
-            raw_name = text[match.start(1) : match.end(1)].strip(" .,:;")
-            raw_firm = text[match.start(2) : match.end(2)].strip(" .,:;")
-            key = (raw_name.lower(), raw_firm.lower())
-            if key in seen:
-                break
-            seen.add(key)
-            out.append({"name": raw_name, "firm": raw_firm})
-            break
-    return out
+    parsed_list = parsed_turns if isinstance(parsed_turns, list) else []
 
+    for idx, source in enumerate(source_turns):
+        source_speaker = str(source.get("speaker", "SPEAKER_XX"))
+        mapped_default = speaker_label_map.get(source_speaker, source_speaker)
 
-def _build_speaker_label_map(turns: list[dict[str, Any]], speaker_mapping: list[dict[str, Any]] | Any) -> dict[str, str]:
-    label_map: dict[str, str] = {}
-    for turn in turns:
-        speaker = str(turn.get("speaker", "")).strip()
-        if speaker:
-            label_map.setdefault(speaker, speaker)
-
-    if isinstance(speaker_mapping, list):
-        for item in speaker_mapping:
-            if not isinstance(item, dict):
-                continue
-            source = str(item.get("speaker_label", "")).strip()
-            inferred = str(item.get("inferred_name", "")).strip()
-            if not source:
-                continue
-            if inferred:
-                label_map[source] = inferred
-            else:
-                label_map.setdefault(source, source)
-    return label_map
+        parsed_item = parsed_list[idx] if idx < len(parsed_list) and isinstance(parsed_list[idx], dict) else {}
+        candidate_speaker = _normalize_final_label(str(parsed_item.get("speaker", "")), source_speaker)
+        final_speaker = candidate_speaker or mapped_default
+        if final_speaker == source_speaker:
+            final_speaker = mapped_default
+
+        text = str(parsed_item.get("text", "")).strip() or str(source.get("text", "")).strip()
+        start = parsed_item.get("start", source.get("start"))
+        end = parsed_item.get("end", source.get("end"))
+
+        out.append(
+            {
+                "speaker": final_speaker,
+                "start": start,
+                "end": end,
+                "text": text,
+            }
+        )
+    return out
 
 
 def run_openai_cleanup_pipeline(
@@ -232,12 +237,15 @@ def run_openai_cleanup_pipeline(
     openai_api_key: str,
     executive_names_csv: str | None,
     *,
-    speaker_map_model: str = "gpt-5-mini",
     cleanup_model: str = "gpt-5",
-    timeout_seconds: float = 1800.0,
-    intro_turn_limit: int = 24,
-    intro_text_char_limit: int = 600,
+    timeout_seconds: float = 600.0,
+    max_turns_per_chunk: int = 80,
+    max_chars_per_chunk: int = 22000,
 ) -> dict[str, Any]:
+    """
+    Single-pass per chunk: each OpenAI call does both speaker naming and transcript cleanup.
+    Avoids a separate full-document speaker inference pass for long audio reliability.
+    """
     try:
         from openai import OpenAI
     except ImportError as exc:
@@ -248,129 +256,141 @@ def run_openai_cleanup_pipeline(
         raise ValueError("Merged transcript must contain a non-empty `turns` list.")
 
     executive_names = _parse_executive_names(executive_names_csv)
-    intro_turns_payload = _build_intro_payload(turns, intro_turn_limit=intro_turn_limit, intro_text_char_limit=intro_text_char_limit)
-    qna_announcements = _extract_qna_announcements(turns)
-    qna_name_candidates = _extract_qna_name_candidates(qna_announcements)
+    chunk_plan = _build_chunk_plan(
+        turns=turns,
+        max_turns_per_chunk=max_turns_per_chunk,
+        max_chars_per_chunk=max_chars_per_chunk,
+    )
 
     client = OpenAI(api_key=openai_api_key, timeout=timeout_seconds, max_retries=0)
 
-    speaker_map_system = (
-        "You map speaker labels to names. Return strict JSON only. "
-        "Infer identities from transcript context."
-    )
-    speaker_map_user = _dumps_compact(
-        {
-            "task": "Infer speaker names from intro and Q&A context snippets.",
+    # Global mapping across chunks.
+    speaker_label_map: dict[str, str] = {}
+    for turn in turns:
+        source = str(turn.get("speaker", "")).strip()
+        if source:
+            speaker_label_map.setdefault(source, source)
+
+    combined_usage = {
+        "input_tokens": 0,
+        "output_tokens": 0,
+        "total_tokens": 0,
+        "cached_input_tokens": 0,
+        "reasoning_tokens": 0,
+    }
+    per_chunk_usage: list[dict[str, Any]] = []
+    cleaned_turns: list[dict[str, Any]] = []
+    chunk_notes: list[str] = []
+    chunk_raw_responses: list[dict[str, Any]] = []
+
+    for i, chunk in enumerate(chunk_plan):
+        start = chunk["start"]
+        end = chunk["end"]
+        source_chunk_turns = turns[start:end]
+
+        payload = {
+            "task": "For this chunk only: infer speaker names and clean transcript text in one pass.",
             "rules": [
-                "Use provided intro turns and Q&A announcements only.",
-                "If a speaker is clearly queue-management/call-control voice, label as Operator.",
-                "Do not guess beyond evidence.",
-                "Prefer names from `executive_names` when they match context.",
-                "If first name matches in `executive_names` but last name is missing/uncertain, use just the first name as the speaker label.",
-                "Q&A participant names may be absent from executive list; infer only if explicit in announcement/context.",
-                "If uncertain, return null inferred_name.",
-                "If inferred_name is present, it must be only a person/role name and must not include SPEAKER_XX or separator forms like Name|SPEAKER_XX.",
+                "Keep turn order and count exactly the same as input chunk.",
+                "Keep start/end timestamps aligned to input turns.",
+                "Correct misspellings and punctuation/casing.",
+                "Only remove filler words (uh, um, you know, like) and clear false-start words/phrases.",
+                "Do not aggressively summarize, compress, or paraphrase full sentences.",
+                "Preserve substantive wording and as much original content as possible.",
+                "If uncertain whether text is filler, keep it.",
+                "Infer speaker names from this chunk context only; do not guess beyond evidence.",
+                "If first name matches in `executive_names` but last name is uncertain, first name alone is allowed.",
+                "If speaker is call-control voice, label as Operator.",
+                "If speaker name is unknown, keep generic label SPEAKER_XX.",
+                "Never output combined labels like Name|SPEAKER_XX.",
+                "Use `existing_speaker_label_map` as source of truth for labels already resolved in prior chunks.",
             ],
             "output_schema": {
-                "speaker_mapping": [
+                "speaker_label_map_updates": [
+                    {"source_label": "SPEAKER_XX", "final_label": "Name or SPEAKER_XX", "reason": "short"}
+                ],
+                "turns": [
                     {
-                        "speaker_label": "SPEAKER_XX",
-                        "inferred_name": "string|null",
-                        "confidence": "0..1",
-                        "reason": "short",
+                        "source_speaker": "SPEAKER_XX",
+                        "speaker": "Name or SPEAKER_XX",
+                        "start": "float",
+                        "end": "float",
+                        "text": "cleaned text",
                     }
                 ],
                 "notes": ["string"],
             },
             "executive_names": executive_names,
-            "intro_turns": intro_turns_payload,
-            "qna_announcements": qna_announcements,
-            "qna_name_candidates": qna_name_candidates,
+            "existing_speaker_label_map": speaker_label_map,
+            "chunk_index": i,
+            "chunk_start_turn_index": start,
+            "chunk_turns": source_chunk_turns,
         }
-    )
 
-    speaker_map_response = client.responses.create(
-        model=speaker_map_model,
-        input=[
-            {"role": "system", "content": speaker_map_system},
-            {"role": "user", "content": speaker_map_user},
-        ],
-    )
-    speaker_map_raw = _response_to_dict(speaker_map_response)
-    speaker_map_usage = _usage_from_response_dict(speaker_map_raw)
-    speaker_map_json = _extract_json_object(_response_text(speaker_map_response))
-    speaker_label_map = _build_speaker_label_map(turns, speaker_map_json.get("speaker_mapping", []))
-
-    cleanup_system = "You are a transcript cleanup and diarization refinement assistant. Return strict JSON only, no markdown."
-    cleanup_payload = {
-        "task": "Clean transcript and produce final speaker-attributed turns.",
-        "rules": [
-            "Correct likely misspellings and improve punctuation/casing.",
-            "Only remove filler words (for example: uh, um, you know, like) and clear false-start words/phrases.",
-            "Do not aggressively summarize, compress, or paraphrase full sentences.",
-            "Preserve substantive wording and as much original content as possible while cleaning.",
-            "When uncertain whether text is filler, keep the text.",
-            "Standardize executive names to the canonical forms in `executive_names` where applicable.",
-            "Use `speaker_label_map` from call 1 as the single source of truth for speaker labels.",
-            "Do not infer any new speaker identities in this call.",
-            "If a source label is not mapped to a name, keep the original generic label (for example SPEAKER_02).",
-            "If a very short mid-sentence speaker switch is likely diarization noise, merge/reassign using sentence continuity.",
-            "Preserve turn order and timing progression.",
-            "Never output combined labels like 'Name|SPEAKER_XX' or 'Name (SPEAKER_XX)'.",
-            "Do not invent facts not present in transcript context.",
-        ],
-        "output_schema": {
-            "speaker_mapping_final": [
-                {
-                    "source_label": "SPEAKER_XX",
-                    "final_label": "string (either inferred name only OR SPEAKER_XX only)",
-                    "confidence": "0..1",
-                    "reason": "short",
-                }
-            ],
-            "turns": [
+        response = client.responses.create(
+            model=cleanup_model,
+            input=[
                 {
-                    "speaker": "string (either inferred name only OR SPEAKER_XX only)",
-                    "start": "float",
-                    "end": "float",
-                    "text": "cleaned text",
-                }
+                    "role": "system",
+                    "content": "You are a transcript cleanup and speaker-label assistant. Return strict JSON only.",
+                },
+                {"role": "user", "content": _dumps_compact(payload)},
             ],
-            "summary": {"turn_count": "int", "speaker_count": "int", "notes": ["string"]},
-        },
-        "executive_names": executive_names,
-        "speaker_label_map": speaker_label_map,
-        "transcript_turns": turns,
-    }
+        )
 
-    cleanup_response = client.responses.create(
-        model=cleanup_model,
-        input=[
-            {"role": "system", "content": cleanup_system},
-            {"role": "user", "content": _dumps_compact(cleanup_payload)},
-        ],
-    )
-    cleanup_raw = _response_to_dict(cleanup_response)
-    cleanup_usage = _usage_from_response_dict(cleanup_raw)
-    cleaned_json = _extract_json_object(_response_text(cleanup_response))
-
-    token_usage = {
-        "speaker_mapping_call": speaker_map_usage,
-        "cleanup_call": cleanup_usage,
-        "combined": _sum_usage(speaker_map_usage, cleanup_usage),
+        raw = _response_to_dict(response)
+        parsed = _extract_json_object(_response_text(response))
+        usage = _usage_from_response_dict(raw)
+        for k in combined_usage:
+            combined_usage[k] += int(usage.get(k) or 0)
+        per_chunk_usage.append({"chunk_index": i, "usage": usage, "turn_range": [start, end]})
+        chunk_raw_responses.append({"chunk_index": i, "raw_response": raw})
+
+        for upd in _extract_map_updates(parsed):
+            source_label = upd["source_label"]
+            final_label = _normalize_final_label(upd["final_label"], source_label)
+            speaker_label_map[source_label] = final_label
+
+        notes = parsed.get("notes", [])
+        if isinstance(notes, list):
+            chunk_notes.extend([str(n) for n in notes if str(n).strip()])
+
+        cleaned_chunk_turns = _coerce_turns(
+            source_turns=source_chunk_turns,
+            parsed_turns=parsed.get("turns"),
+            speaker_label_map=speaker_label_map,
+        )
+        cleaned_turns.extend(cleaned_chunk_turns)
+
+    final_mapping = [
+        {"source_label": source, "final_label": final}
+        for source, final in sorted(speaker_label_map.items(), key=lambda x: x[0])
+    ]
+
+    summary = {
+        "turn_count": len(cleaned_turns),
+        "speaker_count": len({str(t.get("speaker", "")) for t in cleaned_turns}),
+        "chunk_count": len(chunk_plan),
+        "notes": chunk_notes[:200],
+    }
+    cleaned_json = {
+        "speaker_mapping_final": final_mapping,
+        "turns": cleaned_turns,
+        "summary": summary,
+        "openai_token_usage": {
+            "combined": combined_usage,
+            "per_chunk": per_chunk_usage,
+        },
     }
-    cleaned_json["openai_token_usage"] = token_usage
 
     return {
         "cleaned_transcript": cleaned_json,
         "debug": {
-            "speaker_map_model": speaker_map_model,
             "cleanup_model": cleanup_model,
             "executive_names": executive_names,
-            "speaker_mapping": speaker_map_json,
-            "speaker_label_map": speaker_label_map,
-            "speaker_mapping_raw_response": speaker_map_raw,
-            "cleanup_raw_response": cleanup_raw,
-            "openai_token_usage": token_usage,
+            "chunk_plan": chunk_plan,
+            "speaker_label_map_final": speaker_label_map,
+            "openai_token_usage": cleaned_json["openai_token_usage"],
+            "openai_raw_responses": chunk_raw_responses,
         },
     }