From 2215414282f52198067fea10b514bf05f1a50907 Mon Sep 17 00:00:00 2001 From: Martin Date: Thu, 17 Sep 2026 19:16:57 +0200 Subject: [PATCH] Add post-diarization speaker review workflow --- PROJECT_KNOWLEDGE.md | 8 ++- README.md | 19 +++++- run_north.sh | 57 ++++++++++++++++ src/mka/application/meeting_service.py | 32 +++++++++ src/mka/ui/streamlit_app.py | 41 ++++++----- tests/test_alpha_pair.py | 3 + tests/test_meeting_service.py | 94 +++++++++++++++++++++++++- tests/test_speaker_mapping_ui.py | 10 +-- 8 files changed, 240 insertions(+), 24 deletions(-) create mode 100755 run_north.sh diff --git a/PROJECT_KNOWLEDGE.md b/PROJECT_KNOWLEDGE.md index 43f35cc..c10ed39 100644 --- a/PROJECT_KNOWLEDGE.md +++ b/PROJECT_KNOWLEDGE.md @@ -81,7 +81,8 @@ Imported audio (optionally normalize loudness; default on) -> transcribe with whisper.cpp and large-v3-turbo -> optionally diarize with pyannote.audio Community-1 - -> generate a protocol directly from the full transcript + -> when diarized, wait for optional speaker review before explicit protocol generation + -> otherwise generate a protocol directly from the full transcript -> review and edit by a human ``` @@ -121,7 +122,10 @@ Speaker selectors filter participants using a snapshot of all current widget selections, falling back to saved mappings before first interaction. Filtering retains existing assignments; clearing a selection releases the participant. Completeness counts and warnings cover detected speakers only and do not gate -protocol regeneration or change mapping persistence. +protocol generation or change mapping persistence. A diarized run persists the +`awaiting_speaker_review` checkpoint after diarization; an empty, partial or +complete mapping can then explicitly generate a protocol from the existing +artifacts. Failed generation preserves that checkpoint and saved mappings for retry. ## Progress Contract diff --git a/README.md b/README.md index 4a803e9..744c7d7 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,8 @@ Audio file -> FFmpeg preparation (mono, 16 kHz PCM WAV; normalization optional) -> whisper.cpp transcription (large-v3-turbo) -> optional pyannote.audio Community-1 diarization - -> direct full-transcript protocol generation + -> optional speaker review and explicit protocol generation (when diarized) + or direct full-transcript protocol generation (without diarization) -> human review and export ``` @@ -212,6 +213,22 @@ From the Meeting Assistant repository, start the UI with: PYTHONPATH=src:../meeting-lab .venv/bin/streamlit run src/mka/ui/streamlit_app.py ``` +### North launcher + +In the post-diarization Assistant worktree on North, use the checked-in +launcher instead of creating another virtual environment: + +```bash +./run_north.sh +``` + +It reuses North's validated Assistant virtual environment and external +Whisper/Ollama/Docker-ROCm runtime while loading Meeting Assistant from this +worktree and Meeting Lab from +`/opt/git-projekts/meeting-lab`. `HF_TOKEN`, when present, is +passed through to the disposable diarization container without being stored by +the launcher. + Streamlit opens `http://localhost:8501` by default. Installing the sibling Meeting Lab project supplies its runtime requirements such as PyYAML and Requests. Optional diarization dependencies are needed only when diarization diff --git a/run_north.sh b/run_north.sh new file mode 100755 index 0000000..5eb6ee8 --- /dev/null +++ b/run_north.sh @@ -0,0 +1,57 @@ +#!/usr/bin/env bash +set -euo pipefail + +assistant_root="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)" +assistant_src="$assistant_root/src" +lab_root=/opt/git-projekts/meeting-lab +north_venv=/opt/git-projekts/meeting-assistant/.venv +whisper_executable=/home/martin/whisper.cpp/whisper.cpp/build/bin/whisper-cli +whisper_model=/home/martin/whisper.cpp/whisper.cpp/models/ggml-large-v3-turbo.bin + +require_directory() { + if [[ ! -d "$1" ]]; then + printf 'Required directory is missing: %s\n' "$1" >&2 + exit 1 + fi +} + +require_file() { + if [[ ! -f "$1" ]]; then + printf 'Required file is missing: %s\n' "$1" >&2 + exit 1 + fi +} + +require_executable() { + if [[ ! -x "$1" ]]; then + printf 'Required executable is missing or not executable: %s\n' "$1" >&2 + exit 1 + fi +} + +require_directory "$assistant_src" +require_file "$assistant_src/mka/ui/streamlit_app.py" +require_directory "$lab_root" +require_file "$lab_root/src/meeting_lab/__init__.py" +require_directory "$north_venv" +require_executable "$north_venv/bin/streamlit" +require_executable "$whisper_executable" +require_file "$whisper_model" +if ! command -v docker >/dev/null 2>&1; then + printf 'Required executable is not on PATH: docker\n' >&2 + exit 1 +fi + +export MKA_WHISPER_EXECUTABLE="$whisper_executable" +export MKA_WHISPER_MODEL="$whisper_model" +export MKA_PROTOCOL_MODEL=qwen3.8:27b +export MKA_OLLAMA_ENDPOINT=http://127.0.0.1:11434 +export MKA_DIARIZATION_MODE=auto +export MKA_DIARIZATION_RUNTIME=container +export MKA_DIARIZATION_CONTAINER_IMAGE=rocm/pytorch:rocm7.2.1_ubuntu24.04_py3.12_pytorch_release_2.9.1 +export MKA_DIARIZATION_CONTAINER_ARGS='["--device=/dev/kfd","--device=/dev/dri","--group-add","video"]' + +# Keep feature sources ahead of the reused venv's editable installations. +export PYTHONPATH="$assistant_src:$lab_root" + +exec "$north_venv/bin/streamlit" run "$assistant_src/mka/ui/streamlit_app.py" diff --git a/src/mka/application/meeting_service.py b/src/mka/application/meeting_service.py index 28c95a3..da82542 100644 --- a/src/mka/application/meeting_service.py +++ b/src/mka/application/meeting_service.py @@ -92,6 +92,8 @@ class ProcessingOutcome: failed_stage: str | None = None error_message: str | None = None speaker_attribution_available: bool | None = None + awaiting_speaker_review: bool = False + detected_speaker_count: int | None = None @dataclass(frozen=True) @@ -249,6 +251,7 @@ class MeetingProcessingService: "diarization_runtime": self.settings.diarization_runtime, "diarization_container_image": (self.settings.diarization_container_image), "diarization_container_args": self.settings.diarization_container_args, + "stop_after_diarization": options.diarization_enabled, } ) current_stage: str | None = None @@ -299,6 +302,15 @@ class MeetingProcessingService: failed_stage=failed_stage or current_stage or "preparing", error_message=failure_message, ) + if result.protocol_path is None: + return ProcessingOutcome( + succeeded=True, + run_dir=result.run_dir, + original_protocol=None, + protocol_path=None, + awaiting_speaker_review=True, + detected_speaker_count=self._detected_speaker_count(result.run_dir), + ) protocol_path = Path(result.protocol_path) return ProcessingOutcome( succeeded=True, @@ -401,6 +413,13 @@ class MeetingProcessingService: context_data["speaker_mappings"] = dict(sorted(speaker_mappings.items())) self._apply_glossary(context_data) context = self.meeting_lab.create_context(context_data) + # A checkpoint mapping is user input, not a derived protocol artifact. Save it + # before inference so a failed generation remains retryable with the same review. + if context_path.is_symlink(): + context_path.unlink() + context_path.write_text( + yaml.safe_dump(context.data, allow_unicode=True, sort_keys=False), encoding="utf-8" + ) def relay(event: Any) -> None: self._update_timer(event) @@ -432,6 +451,7 @@ class MeetingProcessingService: self.timer.finish_run() raise self.timer.finish_run() + self._persist_timing(Path(result.run_dir)) protocol_path = Path(result.protocol_path) return ProcessingOutcome( succeeded=True, @@ -473,6 +493,18 @@ class MeetingProcessingService: value = metadata.get("speaker_attribution_available") return value if isinstance(value, bool) else None + @staticmethod + def _detected_speaker_count(run_dir: Path | None) -> int | None: + if run_dir is None: + return None + metadata_path = Path(run_dir) / "run_metadata.json" + try: + metadata = json.loads(metadata_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + value = metadata.get("diarization", {}).get("speaker_count") + return value if isinstance(value, int) else None + @staticmethod def _read_failure(run_dir: Path | None) -> dict[str, str]: if run_dir is None: diff --git a/src/mka/ui/streamlit_app.py b/src/mka/ui/streamlit_app.py index 96bdf54..ac731ea 100644 --- a/src/mka/ui/streamlit_app.py +++ b/src/mka/ui/streamlit_app.py @@ -554,7 +554,15 @@ def _render_result() -> None: st.caption("Intermediate artifacts and run metadata were preserved here.") return - st.success("Processing completed. Review the generated protocol before use.") + awaiting_review = getattr(outcome, "awaiting_speaker_review", False) + if awaiting_review: + count = getattr(outcome, "detected_speaker_count", None) + detected = f"{count} speakers detected" if count is not None else "speakers detected" + st.info( + f"Diarization complete — {detected}. Assign speakers if desired, then generate the protocol." + ) + else: + st.success("Processing completed. Review the generated protocol before use.") st.caption(f"Run artifacts: {outcome.run_dir}") if outcome.speaker_attribution_available is False: st.warning( @@ -563,17 +571,18 @@ def _render_result() -> None: ) if message := st.session_state.pop("speaker_mapping_message", None): st.success(message) - with st.expander("Original generated protocol", expanded=False): - st.markdown(outcome.original_protocol or "") - edited = st.text_area( - "Editable protocol", - key="edited_protocol", - height=500, - help="The original protocol.md remains unchanged.", - ) - if st.button("Save edited protocol", type="primary"): - path = MeetingProcessingService.save_edited_protocol(outcome.run_dir, edited) - st.success(f"Saved edited protocol to {path}") + if not awaiting_review: + with st.expander("Original generated protocol", expanded=False): + st.markdown(outcome.original_protocol or "") + edited = st.text_area( + "Editable protocol", + key="edited_protocol", + height=500, + help="The original protocol.md remains unchanged.", + ) + if st.button("Save edited protocol", type="primary"): + path = MeetingProcessingService.save_edited_protocol(outcome.run_dir, edited) + st.success(f"Saved edited protocol to {path}") try: service = MeetingProcessingService(AppSettings.from_environment(), MeetingLabGateway()) @@ -590,7 +599,7 @@ def _render_result() -> None: if duplicate_assignments: st.error("Each participant can be assigned to only one detected speaker.") if st.button( - "Regenerate protocol with confirmed speakers", + "Generate protocol" if awaiting_review else "Regenerate protocol with confirmed speakers", disabled=duplicate_assignments, type="primary", ): @@ -610,7 +619,7 @@ def _render_result() -> None: performance_profile=selected_profile, ), states, - "Starting protocol regeneration", + "Starting protocol generation" if awaiting_review else "Starting protocol regeneration", ) except (OSError, RuntimeError, ValueError) as exc: _remember_regeneration_timing( @@ -618,13 +627,13 @@ def _render_result() -> None: "Protocol regeneration failed", service.timer.snapshot(), ) - st.error(f"Protocol regeneration failed: {exc}") + st.error(f"Protocol generation failed: {exc}") else: _remember_regeneration_timing(states, message, timing) st.session_state.outcome = regenerated _queue_edited_protocol(regenerated.original_protocol or "") st.session_state.speaker_mapping_message = ( - "Speaker mappings saved and protocol regenerated." + "Speaker mappings saved and protocol generated." ) st.rerun() diff --git a/tests/test_alpha_pair.py b/tests/test_alpha_pair.py index acea1dd..82c14bd 100644 --- a/tests/test_alpha_pair.py +++ b/tests/test_alpha_pair.py @@ -104,9 +104,12 @@ def test_paired_anonymous_generation_and_mapped_regeneration( ProcessingOptions(diarization_enabled=True, performance_profile=profile), ) assert outcome.succeeded + assert outcome.awaiting_speaker_review assert prepare.call_args.kwargs["normalization_enabled"] is True assert transcribe.call_args.args[3] == language root = outcome.run_dir + assert calls == [] + service.regenerate_protocol(root, {}, performance_profile=profile) first = root / "protocol/generations/001" before = {p.name: p.read_bytes() for p in first.iterdir()} original = (root / "diarization/transcript_diarized.json").read_bytes() diff --git a/tests/test_meeting_service.py b/tests/test_meeting_service.py index ce5054f..98499c4 100644 --- a/tests/test_meeting_service.py +++ b/tests/test_meeting_service.py @@ -6,6 +6,7 @@ from types import SimpleNamespace from typing import Any import pytest +import yaml from mka.application.config import AppSettings from mka.application.meeting_service import ( @@ -99,6 +100,24 @@ class FakeMeetingLab: ) ) return SimpleNamespace(exit_code=2, run_dir=self.run_dir, protocol_path=None) + if config["stop_after_diarization"]: + write_speaker_review_artifacts(self.run_dir) + (self.run_dir / "run_metadata.json").write_text( + json.dumps( + { + "status": "awaiting_speaker_review", + "diarization": {"speaker_count": 2}, + } + ), + encoding="utf-8", + ) + progress_sink( + SimpleNamespace( + stage="diarization", status="completed", elapsed_seconds=0.3, + progress=None, message=None, + ) + ) + return SimpleNamespace(exit_code=0, run_dir=self.run_dir, protocol_path=None) protocol = self.run_dir / "protocol.md" protocol.write_text("# Generated protocol\n", encoding="utf-8") return SimpleNamespace(exit_code=0, run_dir=self.run_dir, protocol_path=protocol) @@ -387,7 +406,7 @@ def test_process_propagates_enabled_diarization_and_progress(tmp_path: Path) -> audio.write_bytes(b"audio") events = [] - service.process( + outcome = service.process( audio, meeting(), participants(), @@ -397,6 +416,9 @@ def test_process_propagates_enabled_diarization_and_progress(tmp_path: Path) -> assert gateway.config_values is not None assert gateway.config_values["diarization"] == "gpu" + assert gateway.config_values["stop_after_diarization"] is True + assert outcome.awaiting_speaker_review is True + assert outcome.detected_speaker_count == 2 assert [(event.stage, event.status) for event in events[:2]] == [ ("preparing", "started"), ("preparing", "completed"), @@ -404,6 +426,33 @@ def test_process_propagates_enabled_diarization_and_progress(tmp_path: Path) -> assert events[2].progress == 0.25 +def test_diarization_checkpoint_defers_protocol_but_disabled_run_does_not(tmp_path: Path) -> None: + service, gateway = make_service(tmp_path) + audio = tmp_path / "meeting.wav" + audio.write_bytes(b"audio") + + checkpoint = service.process( + audio, meeting(), participants(), ProcessingOptions(diarization_enabled=True) + ) + assert checkpoint.succeeded + assert checkpoint.awaiting_speaker_review + assert checkpoint.original_protocol is None + assert checkpoint.protocol_path is None + + normal_root = tmp_path / "without-diarization" + normal_root.mkdir() + normal_service, normal_gateway = make_service(normal_root) + normal_audio = normal_root / "meeting.wav" + normal_audio.write_bytes(b"audio") + normal = normal_service.process( + normal_audio, meeting(), participants(), ProcessingOptions(diarization_enabled=False) + ) + assert normal.succeeded + assert not normal.awaiting_speaker_review + assert normal.protocol_path is not None + assert normal_gateway.config_values["stop_after_diarization"] is False + + def test_process_propagates_ordered_diarization_container_args(tmp_path: Path) -> None: service, gateway = make_service(tmp_path) service.settings = replace( @@ -537,6 +586,49 @@ def test_protocol_regeneration_allows_unmapped_and_rejects_duplicate_participant ) +@pytest.mark.parametrize( + "mappings", + [ + {"SPEAKER_00": "martin"}, + {"SPEAKER_00": "martin", "SPEAKER_01": "anna"}, + ], +) +def test_protocol_generation_uses_partial_and_complete_confirmed_mappings( + tmp_path: Path, mappings: dict[str, str] +) -> None: + service, gateway = make_service(tmp_path) + write_speaker_review_artifacts(gateway.run_dir) + + outcome = service.regenerate_protocol(gateway.run_dir, mappings) + + assert outcome.succeeded + assert gateway.regeneration is not None + assert gateway.regeneration["meeting_context"].data["speaker_mappings"] == mappings + + +def test_failed_protocol_generation_keeps_checkpoint_mapping_for_retry(tmp_path: Path) -> None: + service, gateway = make_service(tmp_path) + source = write_speaker_review_artifacts(gateway.run_dir) + original_source = source.read_bytes() + original_regeneration = gateway.regenerate_protocol + + def fail_regeneration(*args, **kwargs): + raise RuntimeError("model unavailable") + + gateway.regenerate_protocol = fail_regeneration + with pytest.raises(RuntimeError, match="model unavailable"): + service.regenerate_protocol(gateway.run_dir, {"SPEAKER_00": "martin"}) + + persisted = yaml.safe_load( + (gateway.run_dir / "context" / "meeting_context.yaml").read_text(encoding="utf-8") + ) + assert persisted["speaker_mappings"] == {"SPEAKER_00": "martin"} + gateway.regenerate_protocol = original_regeneration + retry = service.regenerate_protocol(gateway.run_dir, {"SPEAKER_00": "martin"}) + assert retry.succeeded + assert source.read_bytes() == original_source + + def test_fallback_attribution_loss_is_read_from_runtime_metadata(tmp_path: Path) -> None: run_dir = tmp_path / "run" protocol_dir = run_dir / "protocol" diff --git a/tests/test_speaker_mapping_ui.py b/tests/test_speaker_mapping_ui.py index 74b52fc..3a88a1b 100644 --- a/tests/test_speaker_mapping_ui.py +++ b/tests/test_speaker_mapping_ui.py @@ -80,7 +80,7 @@ def test_widget_reruns_filter_release_and_update_status(): assert not app.exception -def test_anonymous_regeneration_remains_allowed(monkeypatch): +def test_checkpoint_allows_anonymous_protocol_generation(monkeypatch): from mka.application.meeting_service import SpeakerMappingReview, SpeakerReview from mka.ui import streamlit_app as ui @@ -89,6 +89,8 @@ def test_anonymous_regeneration_remains_allowed(monkeypatch): run_dir=Path("/tmp/mapping-test"), speaker_attribution_available=True, original_protocol="Anonymous protocol", + awaiting_speaker_review=True, + detected_speaker_count=1, ) review = SpeakerMappingReview((SpeakerReview("SPEAKER_00", ()),), (("a", "A"),), {}) calls = [] @@ -124,9 +126,9 @@ def test_anonymous_regeneration_remains_allowed(monkeypatch): _render_result() app = AppTest.from_function(result_app, args=(outcome,)).run() - regenerate = app.button[1] - assert not regenerate.disabled + generate = next(button for button in app.button if button.label == "Generate protocol") + assert not generate.disabled assert app.warning - regenerate.click().run() + generate.click().run() assert calls == [{}] assert not app.exception