Add EasyCom interaction weak-label pipeline

2026-08-28 00:07:54 +08:00
parent d34725778f
commit 5f45e7c115
5 changed files with 1661 additions and 0 deletions
--- a/adjudicate_easycom_interaction_events.py
+++ b/adjudicate_easycom_interaction_events.py
@@ -0,0 +1,461 @@
+#!/usr/bin/env python3
+"""Optionally adjudicate ambiguous EasyCom interaction proxies with DeepSeek.
+
+This is a second-stage *weak-label adjudicator*, not a replacement for the
+timestamp rules in ``derive_easycom_interaction_events.py``.  By default it
+only reviews ambiguous training candidates and promotes a semantic label when
+two passes agree and both report high confidence.
+
+Transcript excerpts are sent to an external API only when the caller passes
+``--allow-external-transcript-upload``.  The API key is read exclusively from
+``DEEPSEEK_API_KEY`` and is never written to output.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import random
+import time
+import urllib.error
+import urllib.request
+from collections import Counter, defaultdict
+from pathlib import Path
+from typing import Any, Iterable, Sequence
+
+
+PROMPT_VERSION = "easycom-interaction-adjudicator-v1"
+OUTPUT_SCHEMA_VERSION = 1
+DEFAULT_BASE_URL = "https://api.deepseek.com"
+DEFAULT_MODEL = "deepseek-v4-flash"
+DEFAULT_EVENT_TYPES = ("backchannel_candidate", "interruption_candidate")
+ALLOWED_LABELS = {
+    "backchannel",
+    "short_answer_or_response",
+    "interruption",
+    "cooperative_overlap_takeover",
+    "other",
+    "uncertain",
+}
+ALLOWED_CONFIDENCE = {"high", "medium", "low"}
+
+SYSTEM_PROMPT = """You are adjudicating weak labels for conversational turn-taking research.
+Use only the supplied transcript and timing evidence. Do not infer prosody, gaze, or intention that is not observed.
+
+Definitions:
+- backchannel: a brief acknowledgement/feedback that does not take the conversational floor.
+- short_answer_or_response: a brief but content-bearing answer/response, not merely feedback.
+- interruption: the incoming speaker competitively cuts off the previous floor holder and takes the floor.
+- cooperative_overlap_takeover: overlapping floor transfer without enough evidence of competitive interruption.
+- other: none of the above.
+- uncertain: the available text/timing cannot reliably distinguish the labels.
+
+Be conservative. Timing overlap alone does not prove interruption. A short word alone does not prove backchannel.
+Return one JSON object exactly in this form:
+{"items":[{"uid":"...","label":"uncertain","confidence":"low","rationale":"brief evidence-based sentence"}]}
+Return every input uid exactly once. The response must be JSON, with no markdown or extra text."""
+
+
+def _load_jsonl(path: Path) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+    with path.open("r", encoding="utf-8") as handle:
+        for line_number, line in enumerate(handle, start=1):
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                row = json.loads(line)
+            except json.JSONDecodeError as exc:
+                raise ValueError(f"Invalid JSON at {path}:{line_number}: {exc}") from exc
+            if not isinstance(row, dict) or not isinstance(row.get("uid"), str):
+                raise ValueError(f"Expected object with uid at {path}:{line_number}")
+            rows.append(row)
+    return rows
+
+
+def load_candidates(
+    input_dir: Path,
+    *,
+    splits: set[str],
+    event_types: set[str],
+) -> list[dict[str, Any]]:
+    paths = sorted(input_dir.glob("Session_*.events.jsonl"))
+    if not paths:
+        raise FileNotFoundError(f"No Session_*.events.jsonl files in {input_dir}")
+    candidates: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    for path in paths:
+        for row in _load_jsonl(path):
+            uid = row["uid"]
+            if uid in seen:
+                raise ValueError(f"Duplicate event uid: {uid}")
+            seen.add(uid)
+            if row.get("split") in splits and row.get("event_type") in event_types:
+                candidates.append(row)
+    return sorted(candidates, key=lambda row: row["uid"])
+
+
+def stratified_limit(
+    rows: Sequence[dict[str, Any]],
+    limit: int | None,
+    *,
+    seed: int,
+) -> list[dict[str, Any]]:
+    """Select a deterministic pilot across sessions and candidate types."""
+
+    if limit is None or limit >= len(rows):
+        return list(rows)
+    if limit < 0:
+        raise ValueError("--max-events must be non-negative")
+    rng = random.Random(seed)
+    buckets: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list)
+    for row in rows:
+        buckets[(str(row.get("event_type")), str(row.get("session")))].append(row)
+    keys = sorted(buckets)
+    for key in keys:
+        rng.shuffle(buckets[key])
+    rng.shuffle(keys)
+    selected: list[dict[str, Any]] = []
+    while len(selected) < limit:
+        made_progress = False
+        for key in keys:
+            if buckets[key]:
+                selected.append(buckets[key].pop())
+                made_progress = True
+                if len(selected) == limit:
+                    break
+        if not made_progress:
+            break
+    return sorted(selected, key=lambda row: row["uid"])
+
+
+def _candidate_for_prompt(row: dict[str, Any]) -> dict[str, Any]:
+    fields = (
+        "uid",
+        "event_type",
+        "participant_role",
+        "participant_id",
+        "previous_floor_participant_id",
+        "retained_floor_participant_id",
+        "target_matches_previous_floor",
+        "duration_seconds",
+        "onset_overlap_seconds",
+        "gap_from_previous_seconds",
+        "speaker_after_previous_end_seconds",
+        "previous_transcript_has_cutoff",
+        "previous_transcription",
+        "transcription",
+        "target_of_speech",
+    )
+    return {field: row.get(field) for field in fields}
+
+
+def build_user_prompt(rows: Sequence[dict[str, Any]], pass_index: int) -> str:
+    if pass_index == 0:
+        review = "Primary pass: classify conservatively from the supplied evidence."
+    else:
+        review = (
+            "Skeptical review pass: actively look for alternative explanations. "
+            "Use uncertain unless the semantic label remains well supported."
+        )
+    payload = [_candidate_for_prompt(row) for row in rows]
+    return (
+        review
+        + "\nReturn JSON for these candidates:\n"
+        + json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
+    )
+
+
+def prompt_hash(rows: Sequence[dict[str, Any]], passes: int) -> str:
+    payload = {
+        "version": PROMPT_VERSION,
+        "passes": passes,
+        "system": SYSTEM_PROMPT,
+        "rows": [_candidate_for_prompt(row) for row in rows],
+    }
+    encoded = json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _validate_decisions(
+    payload: Any,
+    expected_uids: Sequence[str],
+) -> list[dict[str, str]]:
+    if not isinstance(payload, dict) or not isinstance(payload.get("items"), list):
+        raise ValueError("Response must be a JSON object containing an items list")
+    expected = set(expected_uids)
+    actual: dict[str, dict[str, str]] = {}
+    for item in payload["items"]:
+        if not isinstance(item, dict):
+            raise ValueError("Every response item must be an object")
+        uid = item.get("uid")
+        label = item.get("label")
+        confidence = item.get("confidence")
+        rationale = item.get("rationale")
+        if uid not in expected or uid in actual:
+            raise ValueError(f"Unexpected or duplicate uid in response: {uid!r}")
+        if label not in ALLOWED_LABELS:
+            raise ValueError(f"Invalid label for {uid}: {label!r}")
+        if confidence not in ALLOWED_CONFIDENCE:
+            raise ValueError(f"Invalid confidence for {uid}: {confidence!r}")
+        if not isinstance(rationale, str) or not rationale.strip():
+            raise ValueError(f"Missing rationale for {uid}")
+        actual[uid] = {
+            "uid": uid,
+            "label": label,
+            "confidence": confidence,
+            "rationale": rationale.strip(),
+        }
+    if set(actual) != expected:
+        raise ValueError(f"Response uid mismatch; missing={sorted(expected-set(actual))}")
+    return [actual[uid] for uid in expected_uids]
+
+
+def _post_chat_completion(
+    *,
+    api_key: str,
+    base_url: str,
+    model: str,
+    user_prompt: str,
+    timeout_seconds: float,
+    max_retries: int,
+) -> tuple[dict[str, Any], dict[str, Any]]:
+    endpoint = base_url.rstrip("/") + "/chat/completions"
+    body = {
+        "model": model,
+        "messages": [
+            {"role": "system", "content": SYSTEM_PROMPT},
+            {"role": "user", "content": user_prompt},
+        ],
+        "response_format": {"type": "json_object"},
+        "thinking": {"type": "disabled"},
+        "temperature": 0,
+        "max_tokens": 2048,
+        "stream": False,
+    }
+    request = urllib.request.Request(
+        endpoint,
+        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
+        headers={
+            "Authorization": f"Bearer {api_key}",
+            "Content-Type": "application/json",
+        },
+        method="POST",
+    )
+    last_error: Exception | None = None
+    for attempt in range(max_retries + 1):
+        try:
+            with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
+                envelope = json.loads(response.read().decode("utf-8"))
+            content = envelope["choices"][0]["message"]["content"]
+            if not isinstance(content, str) or not content.strip():
+                raise ValueError("DeepSeek returned empty JSON content")
+            return json.loads(content), envelope.get("usage", {})
+        except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError, KeyError, json.JSONDecodeError, ValueError) as exc:
+            last_error = exc
+            if attempt >= max_retries:
+                break
+            time.sleep(min(30.0, 2.0**attempt))
+    raise RuntimeError(f"DeepSeek request failed after {max_retries + 1} attempts: {last_error}")
+
+
+def adjudicate_batch(
+    rows: Sequence[dict[str, Any]],
+    *,
+    api_key: str,
+    base_url: str,
+    model: str,
+    passes: int,
+    timeout_seconds: float,
+    max_retries: int,
+) -> tuple[list[dict[str, Any]], Counter[str]]:
+    expected_uids = [row["uid"] for row in rows]
+    by_uid: dict[str, list[dict[str, str]]] = {uid: [] for uid in expected_uids}
+    usage: Counter[str] = Counter()
+    for pass_index in range(passes):
+        payload, pass_usage = _post_chat_completion(
+            api_key=api_key,
+            base_url=base_url,
+            model=model,
+            user_prompt=build_user_prompt(rows, pass_index),
+            timeout_seconds=timeout_seconds,
+            max_retries=max_retries,
+        )
+        for key, value in pass_usage.items():
+            if isinstance(value, int):
+                usage[key] += value
+        for decision in _validate_decisions(payload, expected_uids):
+            by_uid[decision["uid"]].append(decision)
+
+    request_prompt_sha256 = prompt_hash(rows, passes)
+    results: list[dict[str, Any]] = []
+    for source in rows:
+        decisions = by_uid[source["uid"]]
+        labels = {decision["label"] for decision in decisions}
+        unanimous = len(labels) == 1
+        final_label = decisions[0]["label"] if unanimous else "uncertain"
+        all_high = all(decision["confidence"] == "high" for decision in decisions)
+        expected_label = {
+            "backchannel_candidate": "backchannel",
+            "interruption_candidate": "interruption",
+        }.get(source.get("event_type"))
+        accepted = unanimous and all_high and final_label == expected_label
+        promoted = (
+            f"{final_label}_proxy_llm_consensus" if accepted else None
+        )
+        results.append(
+            {
+                "schema_version": OUTPUT_SCHEMA_VERSION,
+                "uid": source["uid"],
+                "session": source.get("session"),
+                "split": source.get("split"),
+                "source_event_type": source.get("event_type"),
+                "model": model,
+                "prompt_version": PROMPT_VERSION,
+                "request_prompt_sha256": request_prompt_sha256,
+                "passes": passes,
+                "decisions": decisions,
+                "unanimous": unanimous,
+                "final_label": final_label,
+                "all_high_confidence": all_high,
+                "accepted_for_auxiliary": accepted,
+                "promoted_event_type": promoted,
+            }
+        )
+    return results, usage
+
+
+def _validate_resume_rows(
+    rows: Sequence[dict[str, Any]],
+    *,
+    model: str,
+    passes: int,
+) -> None:
+    """Prevent an append-only output from silently mixing experiments."""
+
+    seen: set[str] = set()
+    for row in rows:
+        uid = row["uid"]
+        if uid in seen:
+            raise ValueError(f"Duplicate uid in existing output: {uid}")
+        seen.add(uid)
+        expected = {
+            "schema_version": OUTPUT_SCHEMA_VERSION,
+            "model": model,
+            "prompt_version": PROMPT_VERSION,
+            "passes": passes,
+        }
+        mismatches = {
+            key: (row.get(key), value)
+            for key, value in expected.items()
+            if row.get(key) != value
+        }
+        if mismatches:
+            raise ValueError(
+                f"Existing output {uid} was produced by a different adjudication "
+                f"configuration: {mismatches}. Use a new --output path."
+            )
+
+
+def _chunks(rows: Sequence[dict[str, Any]], size: int) -> Iterable[Sequence[dict[str, Any]]]:
+    for start in range(0, len(rows), size):
+        yield rows[start : start + size]
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--input-dir", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--splits", nargs="+", default=["train"])
+    parser.add_argument("--event-types", nargs="+", default=list(DEFAULT_EVENT_TYPES))
+    parser.add_argument("--model", default=os.environ.get("DEEPSEEK_MODEL", DEFAULT_MODEL))
+    parser.add_argument("--base-url", default=os.environ.get("DEEPSEEK_BASE_URL", DEFAULT_BASE_URL))
+    parser.add_argument("--batch-size", type=int, default=8)
+    parser.add_argument("--passes", type=int, default=2)
+    parser.add_argument("--max-events", type=int)
+    parser.add_argument("--selection-seed", type=int, default=2027)
+    parser.add_argument("--timeout-seconds", type=float, default=120.0)
+    parser.add_argument("--max-retries", type=int, default=3)
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--allow-external-transcript-upload", action="store_true")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    if args.batch_size <= 0 or args.passes <= 0 or args.max_retries < 0:
+        raise ValueError("batch-size/passes must be positive and max-retries non-negative")
+    candidates = load_candidates(
+        args.input_dir.resolve(),
+        splits=set(args.splits),
+        event_types=set(args.event_types),
+    )
+    candidates = stratified_limit(
+        candidates,
+        args.max_events,
+        seed=args.selection_seed,
+    )
+
+    completed: set[str] = set()
+    if args.output.exists():
+        resume_rows = _load_jsonl(args.output)
+        _validate_resume_rows(resume_rows, model=args.model, passes=args.passes)
+        completed = {row["uid"] for row in resume_rows}
+    pending = [row for row in candidates if row["uid"] not in completed]
+    print(
+        f"Candidates={len(candidates)} completed={len(completed & {row['uid'] for row in candidates})} "
+        f"pending={len(pending)} model={args.model} passes={args.passes}"
+    )
+    if not pending:
+        return
+
+    if args.dry_run:
+        sample = pending[: args.batch_size]
+        print("DRY RUN: no API request will be made. First request prompt:\n")
+        print(build_user_prompt(sample, 0))
+        print("\nprompt_sha256:", prompt_hash(sample, args.passes))
+        return
+    if not args.allow_external_transcript_upload:
+        raise PermissionError(
+            "Refusing to upload transcript excerpts. Review --dry-run, then pass "
+            "--allow-external-transcript-upload if authorized."
+        )
+    api_key = os.environ.get("DEEPSEEK_API_KEY")
+    if not api_key:
+        raise RuntimeError("Set DEEPSEEK_API_KEY in the environment")
+
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    usage: Counter[str] = Counter()
+    accepted: Counter[str] = Counter()
+    processed = 0
+    with args.output.open("a", encoding="utf-8") as handle:
+        for batch_number, batch in enumerate(_chunks(pending, args.batch_size), start=1):
+            results, batch_usage = adjudicate_batch(
+                batch,
+                api_key=api_key,
+                base_url=args.base_url,
+                model=args.model,
+                passes=args.passes,
+                timeout_seconds=args.timeout_seconds,
+                max_retries=args.max_retries,
+            )
+            usage.update(batch_usage)
+            for result in results:
+                handle.write(json.dumps(result, ensure_ascii=False) + "\n")
+                if result["accepted_for_auxiliary"]:
+                    accepted[result["final_label"]] += 1
+            handle.flush()
+            processed += len(results)
+            print(
+                f"batch={batch_number} processed={processed}/{len(pending)} "
+                f"accepted={dict(accepted)} total_tokens={usage.get('total_tokens', 0)}"
+            )
+    print(f"Saved adjudications to {args.output.resolve()}")
+    print("Usage:", dict(usage))
+    print("Accepted high-confidence consensus:", dict(accepted))
+
+
+if __name__ == "__main__":
+    main()