49 changed files with 6156 additions and 19 deletions
+57
View File
@@ -1,5 +1,18 @@
# Project Knowledge # Project Knowledge
## Meeting language in direct protocols
Direct protocol prompts derive their explicit output language from persisted
Meeting Context `meeting.language`, including regeneration and diarized fallback.
Missing context/language explicitly defaults to `de`; omitted language is accepted
without modifying context data. Protocol runtime `output_language` is derived
provenance, not a separate setting. No transcript/context translation is performed.
Prompt evolution (2026-09-11): replaced unconditional German output with the
meeting-language instruction in the shared prompt builder. Names remain verbatim.
Validation uses mocked generation for German/English, plain/diarized inputs,
regeneration and legacy contexts; no live model or extraction gold run is involved.
This is a compact operational summary of the current Meeting Lab state. This is a compact operational summary of the current Meeting Lab state.
## Objective ## Objective
@@ -25,7 +38,28 @@ Implemented:
- Prompt loading from `src/meeting_lab/llm/prompts.py`. - Prompt loading from `src/meeting_lab/llm/prompts.py`.
- Meeting Context V1 loading, validation and optional extraction prompt - Meeting Context V1 loading, validation and optional extraction prompt
injection with minimal extraction JSON provenance. injection with minimal extraction JSON provenance.
- FFmpeg-backed WAV, FLAC and M4A preparation into a per-run canonical mono
16 kHz signed PCM16 WAV artifact before transcription or diarization. Audio
preparation always runs. Optional loudness normalization defaults to on and
currently uses the isolated FFmpeg filter
`loudnorm=I=-16:LRA=11:TP=-1.5`. This is a conservative speech-recording
default and may be revisited after empirical comparison without changing the
orchestration API.
- Interim Markdown protocol generation in `src/meeting_lab/protocol/`. - Interim Markdown protocol generation in `src/meeting_lab/protocol/`.
- Direct protocol prompt input protection: diarized transcripts are rendered as
compact adjacent-speaker blocks without per-segment timestamps. Every source
segment remains represented in order. A deterministic heuristic enforces a
configurable safe input budget, falls back to complete plain transcript text
when necessary, and fails before any Ollama request if even that input is too
large. Silent head/tail truncation is prohibited.
- The `qwen3.8:27b` direct-protocol stage explicitly requests `num_ctx=32768`
and `think=false`; the practical prompt target is approximately 29,000 tokens.
A 31,038-token synthetic prompt passed, but larger prompts are not assumed safe
from the model's advertised 262,144-token native context alone.
- `regenerate_mvp_protocol` updates the run's validated Meeting Context and
regenerates protocol artifacts from the existing diarized transcript when
available. It never reruns audio preparation, Whisper or Pyannote, and it
preserves anonymous speaker labels in the source transcript.
- Non-LLM unit tests for chunking, extraction helpers, protocol rendering and - Non-LLM unit tests for chunking, extraction helpers, protocol rendering and
gold-test runner validation. gold-test runner validation.
- Meeting Context V1 scaffold and documentation for manually maintained - Meeting Context V1 scaffold and documentation for manually maintained
@@ -139,6 +173,9 @@ departments only when they are explicitly supplied as metadata. It must not be
used to infer responsibilities. In the current implementation this context can used to infer responsibilities. In the current implementation this context can
be injected into chunk extraction prompts as authoritative metadata, and only be injected into chunk extraction prompts as authoritative metadata, and only
minimal provenance is written to extraction JSON. minimal provenance is written to extraction JSON.
The implemented MVP statuses are exactly `present` and `mentioned_only`.
Legacy entries without a status receive collection-appropriate defaults. Only
present participants may be targets of explicit `SPEAKER_XX` mappings.
A `responsible` or future `owner` / `assignee` value may be recorded only when A `responsible` or future `owner` / `assignee` value may be recorded only when
source evidence explicitly assigns, accepts or confirms responsibility. If the source evidence explicitly assigns, accepts or confirms responsibility. If the
@@ -258,3 +295,23 @@ The next recommended evaluation step is to use the consolidated V0 result as
input for the unchanged Working Protocol renderer and compare that output input for the unchanged Working Protocol renderer and compare that output
against the Working Protocol Synthesizer V0 baseline and the human reference against the Working Protocol Synthesizer V0 baseline and the human reference
protocol. protocol.
Protocol calls accept an optional positive `protocol_num_thread` setting through
both initial processing and regeneration. With `None`, Ollama receives no
`num_thread` override and selects its own thread configuration.
Configured glossary aliases are diagnostic metadata only. Meeting Context supplies
terminology guidance; direct-protocol transcript text is never alias-substituted.
See `docs/protocol-generation-regression.md` for the frozen-input regression.
Successful protocols are retained in `protocol/generations/NNN/` with their exact
prompt, transcript input, response, model/runtime metadata and Meeting Context.
Relative compatibility symlinks resolve through `protocol/current`, which is
replaced atomically only after a complete record is written. Failed regeneration
keeps the previous protocol, diagnostics and context. Legacy regular files are
snapshotted before conversion. Publication is serialized with a local file lock.
Read `current` once when inspecting a consistent multi-file snapshot. Copy whole
run directories preserving relative symlinks. Process interruption may leave an
unreferenced record or temporary directory, never a partial current generation.
This local POSIX-filesystem contract does not promise power-loss durability or
network-filesystem transaction semantics.
+19
View File
@@ -1,5 +1,12 @@
# Meeting Lab # Meeting Lab
The direct protocol uses saved Meeting Context `meeting.language`: `de` requests
German output and `en` requests English output, for plain and diarized transcripts
and protocol-only regeneration. Meeting Assistant supplies the same selection to
Whisper. Missing context/language retains German output for older runs. Effective
output language is recorded as `output_language` in protocol runtime metadata.
Transcript text, authored context, names and speaker mappings are not translated.
Experimentierumgebung zur Entwicklung eines lokalen Diskussionsanalyzers für Meetingtranskripte. Experimentierumgebung zur Entwicklung eines lokalen Diskussionsanalyzers für Meetingtranskripte.
## Ziel ## Ziel
@@ -271,3 +278,15 @@ Artefakten, Windows-Beispiel und Vergleich mit dem AI-PC stehen in
## Lizenz ## Lizenz
Noch nicht festgelegt. Noch nicht festgelegt.
Successful protocols are retained in `protocol/generations/NNN/` with their exact
prompt, transcript input, response, model/runtime metadata and Meeting Context.
Relative compatibility symlinks resolve through `protocol/current`, which is
replaced atomically only after a complete record is written. Failed regeneration
keeps the previous protocol, diagnostics and context. Legacy regular files are
snapshotted before conversion. Publication is serialized with a local file lock.
Read `current` once when inspecting a consistent multi-file snapshot. Copy whole
run directories preserving relative symlinks. Process interruption may leave an
unreferenced record or temporary directory, never a partial current generation.
This local POSIX-filesystem contract does not promise power-loss durability or
network-filesystem transaction semantics.
+67
View File
@@ -0,0 +1,67 @@
# Optional Speaker Diarization
The direct-protocol MVP keeps speaker diarization disabled by default. Enable
anonymous Community-1 speaker labels with `--diarization auto`, `gpu`, or
`cpu`:
```bash
python3 scripts/run_mvp_meeting.py meeting.wav \
--whisper-model /path/to/ggml-model.bin \
--diarization auto
```
Native mode (the default runtime) requires a compatible local PyTorch and
`pyannote.audio==4.0.7`. For isolated ROCm/CUDA environments, select the
container runtime and provide its image and hardware arguments explicitly:
```bash
python3 scripts/run_mvp_meeting.py meeting.wav \
--whisper-model /path/to/ggml-model.bin \
--diarization gpu \
--diarization-runtime container \
--diarization-container-image IMAGE \
--diarization-container-arg=--device=/dev/kfd \
--diarization-container-arg=--device=/dev/dri \
--diarization-container-arg=--group-add \
--diarization-container-arg=video
```
The container receives `HF_TOKEN` by environment-variable name only. It mounts
the source audio and repository read-only and writes diarization artifacts into
the current run directory. Meeting Lab loads mono 16 kHz PCM16 WAV with
Python's `wave` module and sends an in-memory tensor to pyannote, avoiding its
torchcodec file decoder.
Anonymous `SPEAKER_XX` labels are aligned to Whisper segments by maximum
temporal overlap with Community-1 exclusive diarization. The original Whisper
transcript is preserved; the derived transcript under `diarization/` is used as
the direct-protocol generator's source input.
The full diarized JSON and timestamped text remain immutable audit artifacts,
but their per-segment formatting is too verbose for a full-meeting LLM prompt:
timestamps and repeated speaker labels can more than double input size. For
protocol generation, Meeting Lab deterministically groups only adjacent
segments assigned to the same anonymous speaker and omits timestamps. A later
return by the same speaker starts a new block, and unassigned segments remain
under `SPEAKER_UNASSIGNED`. `protocol/transcript_input.txt` preserves the exact
derived representation sent to prompt construction.
Before contacting Ollama, Meeting Lab conservatively estimates prompt tokens
from UTF-8 byte count without adding a model tokenizer dependency. The default safe
budget is 29,000 estimated tokens within the explicitly configured 32,768-token
Ollama context. The estimate is calibrated against the currently validated
German BPD input and is configurable through
`MvpMeetingConfig.protocol_safe_input_token_budget` or
`--protocol-safe-input-token-budget`.
If compact diarized input exceeds the budget, the generator deterministically
uses the complete plain segment transcript and records the fallback. If that
also exceeds the budget, generation fails before model lookup or generation;
it never truncates, chunks, summarizes, retries, or makes multiple protocol
calls implicitly. Full diarization artifacts are never overwritten by this
selection.
Glossary aliases are recorded as configuration provenance, never applied as
deterministic replacements to the compact/plain protocol input. Canonical
terminology guides generation through Meeting Context. Raw Whisper and
diarization artifacts remain unchanged. `glossary_replacements` is always empty.
+29
View File
@@ -2259,3 +2259,32 @@ the basis of this prototype. Further work should first analyze whether the
failure comes from the schema/prompt representation, the model's sparse-output failure comes from the schema/prompt representation, the model's sparse-output
reliability, or the boundary between subject grouping and semantic synthesis. reliability, or the boundary between subject grouping and semantic synthesis.
It should not proceed through repeated prompt tuning against these nine cases. It should not proceed through repeated prompt tuning against these nine cases.
## EXP-0027 — GTM-Hub real-world protocol regression case
Status: Accepted as a qualitative regression case
Date: 2026-09-09
The private `samples/real_live/gtm_hub_2026-09-07/` case preserves an existing
78-minute German GTM-Hub discussion, its exact protocol input transcript, a
human colleague's concise reference, and two existing Meeting Assistant
protocols (with anonymous speaker labels and with five confirmed mappings).
It is an evaluation dataset, not a prompt/model experiment; no inference was
run to create it.
The evaluation establishes product and architecture learning without changing
the current architecture:
- the detailed/working protocol remains the primary product artifact; a short
distribution protocol is a later separate renderer, not a replacement;
- diarization has particular value for long, multi-speaker,
disagreement-heavy meetings because it improves discourse attribution;
- diarization does not solve semantic commitment extraction: speaker
attribution and normative interpretation are separate problems;
- responsibility and action extraction require stricter evidence than ordinary
discussion summarization, especially when a named speaker is involved.
Evidence and explicit qualitative regression expectations are recorded in
`samples/real_live/gtm_hub_2026-09-07/evaluation.md`. The case does not define
a synthetic gold protocol or a benchmark metric.
+7 -3
View File
@@ -107,8 +107,10 @@ or aliases are corrected.
`department`: Organizational unit. Optional and nullable. `department`: Organizational unit. Optional and nullable.
`attendance_status`: `present` for participants. This distinguishes attendees `attendance_status`: exactly `present` for participants or `mentioned_only` for
from mentioned people. people who are relevant but did not attend. For backward compatibility, a
missing status defaults to `present` in `participants` and `mentioned_only` in
`mentioned_people`.
`mentioned_people`: People discussed or referenced but not present. They are `mentioned_people`: People discussed or referenced but not present. They are
not participants and must not be treated as speakers. not participants and must not be treated as speakers.
@@ -153,7 +155,7 @@ mentioned_people:
aliases: [] aliases: []
role: null role: null
department: null department: null
attendance_status: "not_present" attendance_status: "mentioned_only"
notes: "Wurde erwaehnt, war aber nicht anwesend." notes: "Wurde erwaehnt, war aber nicht anwesend."
organization: organization:
@@ -282,6 +284,8 @@ The current validator checks that:
- participant ids and mentioned-person ids do not collide - participant ids and mentioned-person ids do not collide
- referenced departments exist in `organization.departments` - referenced departments exist in `organization.departments`
- `attendance_status` values are valid - `attendance_status` values are valid
- speaker mappings reference present participants only; mentioned-only people
cannot be diarized speakers
- participants are marked `present` - participants are marked `present`
- mentioned people are not marked `present` - mentioned people are not marked `present`
+25
View File
@@ -0,0 +1,25 @@
# Direct-protocol regression reproduction
The August 2026 glossary regression was isolated with frozen inputs. Condition
A used the original derived transcript; condition B differed only by these
seven deterministic substitutions:
- `Carbofool` to `Carbofol` (two occurrences)
- `Bento Fix` to `Bentofix` (two occurrences)
- `Sikirgut-Heistlöse` to `Secugrid HS` (one occurrence)
- `Lumini` to `Luminy` (two occurrences)
To repeat the manual comparison, copy the investigated run to a new temporary
directory, retain its Meeting Context and speaker mapping, and invoke
`regenerate_mvp_protocol` through the same parameters used by Meeting
Assistant. Never run the comparison in the historical run directory. Preserve
the model, `num_ctx`, `num_predict`, temperature, think setting, thread setting,
and Meeting Context. Compare the new generation's `exact_prompt.txt` and
`transcript_input.txt` with the frozen A and B artifacts before comparing model
output.
The required fixed condition is A: configured glossary aliases remain visible
in Meeting Context terminology guidance, while `transcript_input.txt` retains
the original seven source spellings. Automated tests mock the model and enforce
that invariant; this full historical experiment remains an explicit manual LLM
validation so normal tests do not depend on a local model.
+25
View File
@@ -0,0 +1,25 @@
# Meeting Assistant — First functional alpha
Version: `0.1.0a1`
Tag: `v0.1.0-alpha.1`
This paired Meeting Lab revision supports the first functional end-to-end
Meeting Assistant alpha: reusable transcription and optional diarization,
explicit speaker mapping and regeneration, German and English protocol
generation, glossary provenance without transcript mutation, configurable
protocol thread profiles, generation history with atomic publication, and the
GTM-Hub qualitative regression case.
Validated candidate revisions:
- Assistant: `b50b87c32b3941310c719f459fd45faa87658b68`
- Meeting Lab: `d2e3636b949280402728308022c99bdee6ea1066`
The immutable release revisions are the commits targeted by the paired
`v0.1.0-alpha.1` tags in the two repositories.
Accepted limitations are documented in the Assistant release notes, including
the requirement for local runtime/model/container setup, human review of
generated protocols, imperfect diarization, context limits, local POSIX
publication assumptions, and the historical ignored fixtures needed by some
old experiment tests.
+1 -1
View File
@@ -1,6 +1,6 @@
[project] [project]
name = "meeting-lab" name = "meeting-lab"
version = "0.1.0" version = "0.1.0a1"
requires-python = ">=3.11" requires-python = ">=3.11"
dependencies = [ dependencies = [
"PyYAML>=6.0", "PyYAML>=6.0",
@@ -0,0 +1,69 @@
# GTM-Hub meeting — 2026-09-07
Durable real-world regression case for a German GTM-Hub meeting. This is a
private real-meeting sample; do not publish or redistribute it outside the
intended development context.
## Case metadata
- **Case ID:** `gtm_hub_2026-09-07`
- **Meeting:** GTM-Hub Meeting, 2026-09-07
- **Duration:** about 78 min 36 s (4,715.52 s from existing diarization metadata)
- **Context:** a biweekly cross-functional board meeting about coordinating
projects for new markets, products, and applications; this discussion focused
initially on administrative questions and process definition.
- **Participants:** five confirmed participants: Malte Schnau, Henning
Ehrenberg, Björn-Erik Falkenau, Martin Tazl, and Jovana Husemann.
## Contents and provenance
`transcript/transcript.txt` is the exact compact diarized transcript
representation supplied to both comparable Meeting Assistant protocol calls;
it is the evidentiary source for evaluation. It is byte-identical to both
source runs' `protocol/transcript_input.txt` files (SHA-256
`42ac52223cac36eeeb7c01e6d6673b2fb50973c4365b5ddaafdd12927b2f12eb`).
The source plain Whisper transcripts were also identical between the two runs,
but are not copied because they were not the exact representation supplied to
protocol generation.
| Case file | Meeting Assistant source | Generation | Speaker status |
| --- | --- | --- | --- |
| `references/meeting_assistant_anonymous_speakers.md` | `data/meetings/gtm-hub-meeting-2026-09-07/runs/b895cda2_2026-09-07_11-18-54_GTM-HUB_20260909_025239/protocol.md` | 2026-09-09 01:00:46 UTC; `qwen3.8:27b` | Diarized with anonymous `SPEAKER_XX` labels; zero confirmed speaker-to-person mappings. |
| `references/meeting_assistant_speaker.md` | `data/meetings/gtm-hub-meeting-2026-09-07/runs/284a47be_2026-09-07_11-18-54_GTM-HUB_20260909_031106/protocol.md` | 2026-09-09 01:30:44 UTC; `qwen3.8:27b` | Diarized with the same anonymous labels plus five confirmed speaker-to-person mappings in the prompt/context. |
| `references/human_protocol_malte.txt` | copied human colleague reference supplied for this case | not generated | author identified as Malte by filename; preserve unchanged. |
Both runs have meeting ID `gtm-hub-meeting-2026-09-07`, title/date
`GTM-Hub Meeting 2026-09-07`, the same source recording name and size
(154,620,897-byte FLAC), the same 4,715.52-second prepared-audio duration, the
same Whisper model (`ggml-large-v3-turbo.bin`), and the same protocol model,
temperature (0), context window (32,768), and compact diarized transcript
selection. Their Meeting Context V1 files identify the same meeting and four
shared confirmed participants; the speaker-aware context additionally includes
Jovana Husemann and the five confirmed mappings. Exact run IDs, hashes, flags,
and paths are in `manifest.json`.
The source runs were separate full-pipeline executions: their audio,
transcription, diarization, and exact protocol-input artifacts are
byte-identical, but the mapped-speaker run is not operational evidence that it
reused the anonymous-speaker run's Whisper or diarization artifacts. This case
therefore compares equivalent diarized input conditions with different speaker
identity mappings only.
## Evaluation role
The transcript is the evidentiary source. The human colleague protocol is
**not** a gold protocol: it is a human reference showing what an experienced
participant considered worth preserving in a concise working protocol. The two
Meeting Assistant files are comparison variants. This case must not force a
system to imitate the human protocol verbatim.
This is a valuable regression case because a long, five-speaker discussion
contains competing mental models, repeated disagreement and clarification, and
only partial convergence. It distinguishes strategic collection/evaluation and
coordination from QMS/process documentation and operational project work. It
also discusses responsibility repeatedly without always creating a concrete
personal commitment.
See `evaluation.md` for the qualitative comparison and explicit regression
expectations. No audio, raw model response, container log, or other unrelated
run artifacts are intentionally included.
@@ -0,0 +1,113 @@
# Qualitative evaluation — GTM-Hub 2026-09-07
## Method and status
This is a qualitative regression case, not a measured accuracy benchmark.
Assess future outputs against `transcript/transcript.txt`, the evidentiary
source. The human reference is a concise colleague-authored view, not gold
labels. The two Meeting Assistant protocols are comparison variants; their
wording is not itself evidence.
The transcript shows a long attempt to align several interpretations of the
GTM-Hub: a cross-functional place to collect, evaluate, coordinate, and then
assign ideas; separate QMS/process-description work; and later operational
project execution. It records repeated requests for clarification and limited
common understanding, not immediate detailed closure.
## Comparison
| Dimension | Human reference | Assistant with anonymous speaker labels | Assistant with confirmed speaker mappings |
| --- | --- | --- | --- |
| Core meeting intent | Captures the organizational outcome and the three aims: responsibility clarity, information flow, described project flow. | Strong GTM-Hub coverage as strategic collection, evaluation, coordination. | Equally strong, with competing framings retained. |
| Topic coverage | Selective and highly compressed. | Strong: purpose, process/QMS, criteria, roles. | Strong; additionally separates threads and proponents. |
| Discussion reconstruction | Loses much of why discussion evolved. | Reconstructs broad process logic but smooths argumentative turns. | Best reconstruction of structure and repeated clarification. |
| Differing viewpoints | Notes different perspectives, not their content or ownership. | Weak at retaining who held which position. | Best preservation: differing positions remain tied to confirmed speakers. |
| Speaker/person attribution | Initials only; not a complete attribution record. | No confirmed person mapping in input context. | Valuable where mappings are confirmed; names make unsupported interpretation more consequential. |
| Position / proposal / agreement / consensus / decision | Compresses them into a basic shared understanding. | Sometimes promotes discussion to “es wurde vereinbart” or “die Teilnehmer waren sich einig.” | Preserves positions better, but still overstates some proposals/open details as agreement, decision, or obligation. |
| Action-item precision | One concise genuine next step: comment on/give feedback on the BD presentation. | Creates broadly attributed implied actions despite absent explicit individual assignment. | Turns suggestions into named commitments, including delegation by Martin Tazl. |
| Responsibility attribution | Focuses on later project-responsibility clarity rather than naming it prematurely. | Describes future assignment reasonably, but implies responsibility in action wording. | Better attribution does not establish responsibility; the named false action is high risk. |
| Open-question preservation | Retains only the core unresolved process issue. | Lists open points, some synthesized rather than clearly preserved as open. | Keeps several open themes but turns others into settled next steps. |
| Useful detail | Good concise working/distribution reference; insufficient alone for later argument reconstruction. | Detailed enough for overall logic, though attribution and restraint are weak. | Most useful detailed reference for complex multi-speaker discussion, provided commitments are evidence-checked. |
| Risk of overinterpretation | Low through editorial compression, at cost of omitted context. | Moderate: consensus and implied collective actions strengthened. | Highest impact when a semantic overinterpretation is attached to a named person. |
| Product suitability | Strong concise working/distribution protocol; not detailed evidentiary reconstruction. | Better detailed reference than short distribution protocol, but needs restraint. | Best detailed reference of the comparison outputs; a separate later renderer should create the concise distribution protocol. |
### Overall conclusion
The human reference is very strongly editorially compressed. It captures the
main organizational outcome well and works as a concise working/distribution
protocol, but loses much argumentative context and can be insufficient weeks
later when reconstructing why the discussion evolved.
The anonymous-speaker Assistant output has strong topic coverage and
reconstructs overall process logic well. It tends to smooth disagreement into
stronger consensus than the transcript supports, is weaker at retaining who
held which position, and can overstate phrases such as “es wurde vereinbart”
or “die Teilnehmer waren sich einig.”
The mapped-speaker Assistant output best preserves differing viewpoints and
discussion structure. It is substantially better for reconstructing who argued
what in this complex multi-speaker discussion; speaker information increases
the value of a detailed protocol. It does not solve semantic interpretation of
commitments/responsibility. A false action is more dangerous when confidently
attached to a named person.
## Explicit regression expectations
These are qualitative expectations, not a synthetic gold protocol.
### R1 — Core purpose
The detailed protocol should preserve that the GTM-Hub is a cross-functional
collection, evaluation, and coordination point for ideas, markets, products,
or initiatives.
### R2 — Different perspectives
Preserve that participants approached the subject from different
perspectives/mental models; do not claim immediate detailed agreement.
### R3 — Strategic vs. operational distinction
Preserve the distinction among strategic evaluation/coordination, formal
process description/QMS, and operational project execution.
### R4 — Responsibility timing
Do not imply concrete project responsibility before group evaluation and
assignment. The transcript explicitly says responsibility does not yet exist at
that stage and is assigned later.
### R5 — Consensus strength
It is valid to report a basic/common understanding around collection,
discussion, and assignment of ideas. Do not turn unresolved process details
into final consensus.
### R6 — Speaker-aware value
For diarized processing, differing positions should remain attributable to
separate speakers/persons where mappings are confirmed.
### R7 — No responsibility invention
Speaker-aware processing must not turn an organizational suggestion into a
personal commitment. In particular, the statement that formal completeness
checking could be done by an administrative office/secretariat must **not**
become “Martin Tazl will delegate the formal review to an administrative
office” without explicit commitment evidence.
### R8 — Proposal vs. action
A suggestion to structure discussion into two parts must not automatically
become a personal action item for the person proposing it.
### R9 — Human-reference role
Detailed-protocol mode must not be forced to become as short as the human
reference.
### R10 — Detailed-protocol product direction
The main protocol should retain enough detail for a participant to reconstruct
the substance several weeks later. A future short/distribution protocol may
deliberately compress much more strongly.
@@ -0,0 +1,42 @@
{
"case_id": "gtm_hub_2026-09-07",
"meeting_date": "2026-09-07",
"meeting_id": "gtm-hub-meeting-2026-09-07",
"duration_seconds": 4715.52,
"transcript": {
"case_path": "transcript/transcript.txt",
"source_paths": [
"/opt/git-projekts/meeting-assistant/data/meetings/gtm-hub-meeting-2026-09-07/runs/b895cda2_2026-09-07_11-18-54_GTM-HUB_20260909_025239/protocol/transcript_input.txt",
"/opt/git-projekts/meeting-assistant/data/meetings/gtm-hub-meeting-2026-09-07/runs/284a47be_2026-09-07_11-18-54_GTM-HUB_20260909_031106/protocol/transcript_input.txt"
],
"sha256": "42ac52223cac36eeeb7c01e6d6673b2fb50973c4365b5ddaafdd12927b2f12eb",
"representation": "exact compact diarized protocol input"
},
"human_reference": { "path": "references/human_protocol_malte.txt" },
"comparison_provenance": {
"source_runs_are_separate_full_pipeline_executions": true,
"relevant_audio_transcription_diarization_and_protocol_input_artifacts_are_byte_identical": true,
"comparison": "equivalent diarized input conditions with anonymous versus confirmed speaker identity mappings",
"does_not_establish_operational_reuse_of_whisper_or_diarization_between_generations": true
},
"anonymous_speaker_reference": {
"case_path": "references/meeting_assistant_anonymous_speakers.md",
"source_path": "/opt/git-projekts/meeting-assistant/data/meetings/gtm-hub-meeting-2026-09-07/runs/b895cda2_2026-09-07_11-18-54_GTM-HUB_20260909_025239/protocol.md",
"source_run_id": "b895cda2_2026-09-07_11-18-54_GTM-HUB_20260909_025239",
"generation_timestamp": "2026-09-09T01:00:46+00:00",
"model": "qwen3.8:27b",
"diarization_enabled": true,
"speaker_mapping_count": 0,
"sha256": "134fe2caa0c798f46248f039a4e179fc5a1024f0469ca09f7f9ca176e5e30264"
},
"speaker_aware_reference": {
"case_path": "references/meeting_assistant_speaker.md",
"source_path": "/opt/git-projekts/meeting-assistant/data/meetings/gtm-hub-meeting-2026-09-07/runs/284a47be_2026-09-07_11-18-54_GTM-HUB_20260909_031106/protocol.md",
"source_run_id": "284a47be_2026-09-07_11-18-54_GTM-HUB_20260909_031106",
"generation_timestamp": "2026-09-09T01:30:44+00:00",
"model": "qwen3.8:27b",
"diarization_enabled": true,
"speaker_mapping_count": 5,
"sha256": "18c77411a7ecf869940e56308db258f177c1d813dbf7495f9d865a72a1fadf3d"
}
}
@@ -0,0 +1,21 @@
Teilnehmer:
EH, MT, BEF, JH, JMS
Thema 1: Prozessaufbau nach Mission 2030
Der Aufbau der Prozesse nach Mission 2030 erfolgt unabhängig von dem im GtM-Hub besprochenen Prozess.
Mission 2030 wird dafür in mehrere Abschnitte unterteilt und neu beschrieben. Die Abbildung dient der Zertifizierung nach ISO 9001.
EH hat den ersten Entwurf vorgestellt und den Unterschied zum GtM-Hub-Prozess erläutert.
Ziel ist es einen Großteil so aufzubauen, wie wir möchten, dass es bearbeitet wird. Es wird natürlich weiterhin Ausnahmen geben.
Thema 2: Verantwortlichkeiten und Projektablauf
Wir haben lange diskutiert und festgestellt, dass wir das Thema aus unterschiedlichen Perspektiven und mit unterschiedlichen Zielsetzungen angegangen sind.
Zielsetzung ist:
Verantwortung: Projektverantwortliche klar benennen.
Informationsfluss: Sicherstellen, dass Themen nicht in einzelnen Abteilungen bearbeitet werden, ohne andere betroffene Bereiche zu informieren.
Projektablauf: Den Projektablauf genauer beschreiben. Systematische Bewertung etc..
Dabei wurde angemerkt, dass es Kriterien braucht, wann dieser Projektablauf greift – nicht jede sinnvolle Idee lässt sich in ein einheitliches Raster pressen.
Grundsätzlich besteht Einigkeit darüber, wie Ideen ins Gremium getragen und dort weiterverarbeitet werden sollen.
Als nächster Schritt wurde festgehalten, dass wir die Präsentation von BD kommentieren und Feedback dazu geben.
@@ -0,0 +1,53 @@
# Meeting Protocol
## Einordnung und Zielsetzung des Gremiums
Die Teilnehmer diskutierten die Rolle und den Zweck des „GTM-Hub“ (Go-To-Market Hub). Es wurde klargestellt, dass das Gremium nicht dazu dient, einzelne Projekte im Detail zu steuern oder als Projektmanagement-Instanz zu fungieren, sondern als strategischer Filter und Koordinationspunkt für neue Ideen, Märkte und Initiativen im Rahmen der „Mission 2030“.
Ein zentraler Punkt der Diskussion war die Abgrenzung zwischen dem strategischen Prozess (Sammeln, Bewerten und Zuordnen von Ideen) und der operativen Projektumsetzung. Es wurde betont, dass die Geschäftsführung die Diskussion über neue Märkte und strategische Innovationen in diesem Gremium wünscht, um abteilungsübergreifende Transparenz zu schaffen und Doppelarbeit zu vermeiden. Die Teilnehmer waren sich einig, dass der Fokus zunächst auf der Definition des gemeinsamen Verständnisses und der groben Prozessschritte liegen muss, bevor in Details der einzelnen Projekttypen (z. B. Zulassung, Produktentwicklung) eingetaucht wird.
## Prozessbeschreibung und Integration in das Prozessmanagement
Es wurde vereinbart, den GTM-Hub-Prozess im bestehenden Prozessmanagement-System (im Kontext der ISO 9000-Anforderungen) abzubilden. Der Prozess soll als Unterprozess oder eigenständiger Prozess beschrieben werden, der die Schritte von der Idee bis zur Zuordnung an einen Verantwortlichen umfasst.
Die Diskussion umfasste folgende Aspekte:
* **Eingangsstufe:** Ideen können aus verschiedenen Quellen (Vertrieb, F&E, Marketing, externe Quellen) kommen. Es wurde vorgeschlagen, eine formale Eingangsstufe (z. B. eine E-Mail-Adresse oder ein Formular) zu definieren, die von einer neutralen Stelle (z. B. Sekretariat oder F&E als Sammelstelle) auf formale Vollständigkeit geprüft wird, um sicherzustellen, dass ausreichend Informationen vorliegen, bevor das Gremium tagt.
* **Entscheidungsstufe:** In der GTM-Hub-Runde wird diskutiert, ob die Idee strategisch passt, welche Abteilung federführend ist und ob weitere Informationen benötigt werden.
* **Zuordnung:** Nach der Entscheidung wird die Idee an den zuständigen Bereich (z. B. Business Development, Produktmanagement, F&E) zur weiteren Bearbeitung übergeben.
Es wurde darauf hingewiesen, dass die Beschreibung im Prozessmanagement so gestaltet sein muss, dass sie für alle Beteiligten nachvollziehbar ist und die Mindestanforderungen der ISO 9000 erfüllt, ohne unnötige bürokratische Hürden zu schaffen. Die Teilnehmer waren sich einig, dass die Beschreibung lebendig bleiben muss und bei Bedarf angepasst werden kann.
## Strategische Ausrichtung und Kriterien
Ein wiederkehrender Punkt war die Notwendigkeit eines strategischen Checks, bevor in die operative Bearbeitung eingestiegen wird. Es wurde kritisiert, dass in früheren Diskussionen zu schnell in Details gegangen wurde, ohne die strategische Passung (z. B. Marktfit, Deckungsbeitrag, strategische Priorität) ausreichend zu prüfen.
Die Teilnehmer diskutierten, welche Kriterien für die Bewertung von Ideen relevant sind. Dabei wurden unterschiedliche Ansätze genannt:
* **Strategische Kriterien:** Passung zur Mission 2030, Marktchancen, Wettbewerbslage.
* **Operative Kriterien:** Technische Machbarkeit, Ressourcenbedarf, Zeitplan.
Es wurde vereinbart, dass die Kriterien nicht für alle Projekttypen identisch sein müssen. Für neue Märkte und strategische Innovationen sind andere Bewertungskriterien relevant als für Produktverbesserungen oder Zulassungen. Die Teilnehmer waren sich einig, dass die Definition dieser Kriterien Teil der Prozessbeschreibung sein muss, aber nicht im Detail im GTM-Hub-Prozess selbst festgelegt werden sollte, sondern in den jeweiligen Fachprozessen.
## Rollen und Verantwortlichkeiten
Die Diskussion um die Rollen und Verantwortlichkeiten ergab folgende Punkte:
* **Sammelstelle:** Es wurde vorgeschlagen, eine neutrale Stelle (z. B. F&E oder Sekretariat) zu benennen, die für die formale Prüfung und das Einsammeln der Ideen verantwortlich ist. Diese Stelle trifft keine inhaltlichen Entscheidungen, sondern stellt sicher, dass die Ideen ausreichend beschrieben sind.
* **GTM-Hub-Runde:** Die Runde besteht aus Vertretern der relevanten Abteilungen (F&E, Marketing, Business Development, Qualitätssicherung, Produktmanagement). Sie trifft die Entscheidung über die Weiterverfolgung und die Zuordnung.
* **Projektverantwortliche:** Nach der Zuordnung liegt die Verantwortung für die operative Umsetzung bei der jeweiligen Fachabteilung. Der GTM-Hub behält die Übersicht und kann bei Bedarf Feedback einholen, ist aber nicht für die Projektsteuerung verantwortlich.
Es wurde betont, dass die Zuordnung an einen Projektverantwortlichen (oder eine Abteilung) ein zentraler Schritt ist, um Transparenz zu schaffen und Doppelarbeit zu vermeiden. Die Teilnehmer waren sich einig, dass die Definition der Verantwortlichkeiten im Prozess klar sein muss, aber die konkrete Person erst im Einzelfall bestimmt wird.
## Offene Punkte und nächste Schritte
* **Präsentation und Feedback:** Es wurde vereinbart, dass die aktuelle Fassung des GTM-Hub-Prozesses (bzw. die Präsentation mit den vorgeschlagenen Schritten) an alle Teilnehmer versendet wird. Die Teilnehmer sollen bis zu einem bestimmten Termin (nicht explizit genannt, aber implizit kurzfristig) Feedback geben.
* **Strategische Kriterien:** Die Definition der strategischen Kriterien für die Bewertung von Ideen ist noch offen. Es wurde vorgeschlagen, dies in einem separaten Schritt oder in Zusammenarbeit mit der Geschäftsführung zu klären.
* **Prozessbeschreibung:** Die formale Beschreibung des GTM-Hub-Prozesses im Prozessmanagement-System ist noch nicht abgeschlossen. Die Teilnehmer werden ihre jeweiligen Inputs einbringen, um die Beschreibung zu finalisieren.
* **Abgrenzung zu anderen Prozessen:** Die Abgrenzung zwischen dem GTM-Hub-Prozess und den bestehenden Prozessen (z. B. Vertriebsprozess, Entwicklungsprozess) muss noch klarer definiert werden. Es wurde vereinbart, dass der GTM-Hub-Prozess als Eingangspunkt für neue Ideen dient und die Zuordnung an die bestehenden Prozesse erfolgt.
## Maßnahmenliste
* **Verantwortlich:** [Nicht explizit zugewiesen, implizit die Teilnehmer des GTM-Hubs]
* **Maßnahme:** Versendung der aktuellen Fassung des GTM-Hub-Prozesses an alle Teilnehmer.
* **Frist:** [Nicht explizit genannt]
* **Verantwortlich:** [Nicht explizit zugewiesen, implizit die Teilnehmer des GTM-Hubs]
* **Maßnahme:** Feedback zur aktuellen Fassung des GTM-Hub-Prozesses geben.
* **Frist:** [Nicht explizit genannt]
* **Verantwortlich:** [Nicht explizit zugewiesen, implizit die Teilnehmer des GTM-Hubs]
* **Maßnahme:** Definition der strategischen Kriterien für die Bewertung von Ideen.
* **Frist:** [Nicht explizit genannt]
* **Verantwortlich:** [Nicht explizit zugewiesen, implizit die Teilnehmer des GTM-Hubs]
* **Maßnahme:** Finalisierung der formalen Beschreibung des GTM-Hub-Prozesses im Prozessmanagement-System.
* **Frist:** [Nicht explizit genannt]
@@ -0,0 +1,45 @@
# Meeting Protocol
## Einordnung und Zielsetzung des GTM-Hubs
Die Teilnehmer haben die ursprüngliche Intention und den aktuellen Stand des „GTM-Hub“-Prozesses (Go-To-Market Hub) diskutiert. Der Prozess wurde im Kontext der Strategie „Mission 2030“ eingeführt, um abteilungsübergreifende Ideen, neue Märkte und Produktinnovationen zu bündeln, zu bewerten und zuzuordnen.
**Diskussionsverlauf und Positionen:**
* **Jovana Husemann** betonte, dass die Runde primär strategischen Charakter hat. Sie verwies darauf, dass die Geschäftsführung die Diskussion über neue Märkte und strategische Innovationen in diesem Gremium gewünscht habe. Sie kritisierte, dass in den letzten Sitzungen zu viel Zeit für prozessuale Details aufgewendet wurde, während der strategische Teil (die Bewertung, ob eine Idee zur Strategie passt) noch nicht abschließend geklärt ist. Sie forderte, dass vor der detaillierten Prozessbeschreibung die strategischen Kriterien und die Zustimmung der Abteilungen zu den definierten Stufen (Stages) erfolgen müssen.
* **Malte Schnau** stellte fest, dass der Fokus bisher zu stark auf der reinen Prozessbeschreibung lag, während die Kernanforderung – die transparente Sammlung von Ideen, die Zuordnung an einen Projektverantwortlichen und die Vermeidung von Doppelarbeit – im Vordergrund stehen sollte. Er argumentierte, dass der Prozess lebendig bleiben muss und nicht im Detail für jede Projektart (z. B. Zulassung vs. Produktentwicklung) vordefiniert werden kann, da die Anforderungen stark variieren.
* **Henning Ehrenberg** ordnete die Diskussion in den Kontext des Qualitätsmanagementsystems (ISO 9000) ein. Er betonte, dass die Beschreibung der Prozesse nicht nur der internen Abstimmung dient, sondern auch der Nachweisführung gegenüber Auditoren (Rezertifizierungsaudit) dienen muss. Er plädierte dafür, die Prozesse so zu beschreiben, dass sie zu 90 % der Realität entsprechen, um Abweichungen im Audit zu minimieren.
* **Björn-Erik Falkenau** hob hervor, dass die Runde dazu dienen soll, die „Geschichte“ und den strategischen Check vor der operativen Umsetzung sicherzustellen. Er kritisierte, dass oft zu schnell in operative Details eingetaucht wird, ohne vorher zu prüfen, ob die Idee zur Unternehmensstrategie passt oder ob bereits andere Abteilungen an ähnlichen Themen arbeiten. Er forderte, den Prozess global in wenigen Hauptschritten zu definieren, bevor Details geklärt werden.
* **Martin Tazl** stellte klar, dass die Forschung & Entwicklung (F&E) nicht die operative Verantwortung für die strategische Bewertung neuer Märkte oder Business-Entwicklung trägt. Die Rolle der F&E beschränkt sich auf die formale Prüfung der Eingangsideen (Vollständigkeit) und die technische Machbarkeit, falls die Idee in den Bereich der Produktentwicklung fällt. Er betonte, dass die Zuordnung der Verantwortung (wer „den Hut aufhat“) erst nach der strategischen Bewertung in der Runde erfolgt.
**Entscheidungsgrundlagen und Konsens:**
* Es besteht Einigkeit, dass der GTM-Hub als zentraler Sammel- und Bewertungspunkt für neue Ideen, Märkte und Produkte dient.
* Die strategische Bewertung (Passt die Idee zur Mission 2030?) muss vor der operativen Zuordnung an eine Fachabteilung erfolgen.
* Die Prozessbeschreibung muss die Anforderungen des Qualitätsmanagements (ISO 9000) erfüllen, darf aber keine unnötigen Fesseln für die operative Arbeit anlegen.
* Die Verantwortung für die inhaltliche Bewertung neuer Märkte liegt nicht primär bei der F&E, sondern erfordert die Einbindung von Business Development, Marketing und Vertrieb.
## Prozessbeschreibung und Integration in das Qualitätsmanagement
Die Teilnehmer diskutierten die technische Umsetzung der Prozessbeschreibung im Rahmen des Qualitätsmanagementsystems.
**Diskussionsverlauf und Positionen:**
* **Henning Ehrenberg** erläuterte, dass die Beschreibung der Prozesse im Qualitätsmanagementsystem (QMS) bereits seit über einem Jahr läuft und unabhängig von der GTM-Hub-Diskussion fortgeführt wird. Er zeigte auf, wie der GTM-Hub als Unterprozess in die bestehende Prozesslandschaft (z. B. Vertriebsprozess, F&E-Prozess) integriert werden kann. Er betonte, dass die Beschreibung so erfolgen muss, dass sie für alle Beteiligten nachvollziehbar ist und die Mindestanforderungen der ISO 9000 erfüllt.
* **Jovana Husemann** äußerte Bedenken, dass die reine Prozessbeschreibung im QMS von den Mitarbeitern nicht gelesen oder beachtet wird. Sie argumentierte, dass die strategische Ausrichtung und die Zustimmung der Abteilungen zur Strategie wichtiger sind als die formale Prozessbeschreibung. Sie forderte, dass die Präsentation der strategischen Stufen (Stages) von allen Abteilungen kommentiert und bestätigt werden muss, bevor die Prozessbeschreibung finalisiert wird.
* **Malte Schnau** schlug vor, die Diskussion in zwei Teile zu gliedern: 1) Die grobe Definition des Ablaufs (Sammlung, Bewertung, Zuordnung) und 2) die detaillierte Definition der Kriterien für verschiedene Projektarten. Er betonte, dass der erste Teil (der Ablauf) schnell abgeschlossen werden sollte, um Fortschritt zu erzielen.
* **Martin Tazl** wies darauf hin, dass die formale Prüfung der Eingangsideen (z. B. Vollständigkeit der Informationen) eine einfache Aufgabe ist, die nicht zwingend von der F&E, sondern von einer administrativen Stelle (z. B. Sekretariat) übernommen werden kann. Die F&E sollte sich nur auf die technische Machbarkeit konzentrieren, wenn die Idee in den Bereich der Produktentwicklung fällt.
**Entscheidungsgrundlagen und Konsens:**
* Die Prozessbeschreibung im QMS muss die strategischen Ziele des GTM-Hubs abbilden und die Anforderungen der ISO 9000 erfüllen.
* Die formale Prüfung der Eingangsideen kann von einer administrativen Stelle übernommen werden, um die F&E zu entlasten.
* Die strategische Bewertung und die Zuordnung der Verantwortung erfolgen in der GTM-Hub-Runde.
## Nächste Schritte und offene Punkte
Die Teilnehmer haben die nächsten Schritte zur Klärung der offenen Punkte besprochen.
**Maßnahmen und Verantwortlichkeiten:**
* **Jovana Husemann** wird die Präsentation der strategischen Stufen (Stages) an die Teilnehmer senden und um Kommentare bitten. Sie erwartet, dass die Abteilungen die strategische Ausrichtung bestätigen oder kommentieren.
* **Malte Schnau** wird die Diskussion in zwei Teile gliedern: 1) Die grobe Definition des Ablaufs und 2) die detaillierte Definition der Kriterien. Er wird vorschlagen, den ersten Teil (den Ablauf) schnell abzuschließen.
* **Henning Ehrenberg** wird die Prozessbeschreibung im QMS so anpassen, dass sie die strategischen Ziele des GTM-Hubs abbildet und die Anforderungen der ISO 9000 erfüllt.
* **Martin Tazl** wird die formale Prüfung der Eingangsideen an eine administrative Stelle delegieren, um die F&E zu entlasten.
**Offene Punkte:**
* Die strategische Bewertung der Ideen muss vor der operativen Zuordnung an eine Fachabteilung erfolgen.
* Die Prozessbeschreibung im QMS muss die Anforderungen der ISO 9000 erfüllen.
* Die formale Prüfung der Eingangsideen muss von einer administrativen Stelle übernommen werden.
@@ -0,0 +1,193 @@
SPEAKER_03: wie das mit Off-Handing-Warten lohnt. Okay. Vielleicht könnt ihr mich noch mal auf Stand holen, Giovanna. Wir hatten das letzte Mal vor drei Wochen zusammengesessen bezüglich der... Also beziehungsweise da war ich nicht mit da, das war im Urlaub. Lars hat da einmal ein Protokoll geschrieben mit der Namensänderung und wie der Ablauf der Projekte... Das war mal ein bisschen... Ah, hallo. Hi. Dass da nochmal über den Ablauf gesprochen wurde. Ich habe ja mit Vorplag gemacht. Darüber wurde dann ja noch mal mit ausgiebig drüber diskutiert, über die möglichen Abläufe. Und mein letzter Stand war jetzt, dass ihr am 28. noch mal sitzen wolltet, August, bezüglich eines Vorschlags. Da stand ein Protokoll von Lars.
SPEAKER_02: Ja, wir haben am 28. zusammengesessen, aber das hatte ein anderes Thema. Da ging es darum, dass wir generell das ganze Thema Mission 2030 im Prozessmanagement abbilden müssen. Und wir sind angefangen mit dem Vertriebsprozess oder haben an dem Vertriebsprozess gearbeitet, wie wir den abbilden wollen. Wir haben das auch ganz bewusst erstmal nicht in einer Riesenrunde gemacht. Es geht auch wirklich nur um den Vertriebsprozess. Also es kommt irgendwas bei Commercial Sales an. Ne, stimmt gar nicht. Pre-Sales. Und wie geht es dann weiter?
SPEAKER_04: Ja, und grundsätzlich zum Termin, als du nicht da warst, also wir haben tatsächlich zwei Themen, meiner Meinung nach. Einmal das sehr strategische Thema, wie wollen wir, dass die Geschäftsführung geklärt hat, die Idee kommt rein, wir wollen die Themen einmal zu klären. Und das haben wir als Development erstmal eine Präsentation gemacht, als du das mitbekommst.
SPEAKER_03: Ja, ich habe es gesehen.
SPEAKER_04: Genau, und das andere Thema ist an diesem Prozessmanagement, das ist eigentlich, also ich kann das wenig zu sagen, da gibt es schon einen Weg, wie du das machst, Henning, oder also, ich glaube, ihr braucht mir da gar nicht so ein Business Development dafür.
SPEAKER_02: Also wir werden die mit Sicherheit brauchen, aber nicht zum jetzigen Zeitpunkt, weil, was wir vorhaben, ist ja, dass wir, also wir haben dieses Mission 2030, was irgendwann mit Business Development losgeht, was über mehrere Phasen dann irgendwann hoffentlich darin endet, dass wir Ware ausgeliefert haben. Und das ist ja ein mehrstufiger Prozess und wir sind jetzt so weit, dass wir gesagt haben, okay, wir werden das zerlegen in Business Development, in Pre-Sales, bis zu dem Zeitpunkt, wo eine Ausschreibung rauskommt, dann ist es im Wesentlichen der Commercial Sales, wenn die Ausschreibung raus ist. Und dann haben wir den Lieferprozess, den Lieferprozess, den haben wir schon, den brauchen wir nicht ändern. Und die anderen drei Prozesse müssen wir beschreiben. Also wir müssen schon beschreiben, wir müssen beschreiben, was dort passiert. Und wir müssen auch gewisse Regeln einhalten. Also das ist jetzt der ISO 9000-Gedanke, da wird es sicherlich die ein oder andere Anforderung geben, warum man sagt, das muss man jetzt so und so machen und man muss es auch so und so, man müsste es in irgendeiner Form auch dokumentieren können, zeigen können, also nicht nur verbal erzählen, wir arbeiten so, sondern es auch zeigen können. Aber das können wir alles relativ light halten und wir haben jetzt beim letzten Mal wirklich angefangen und haben gesagt, okay, wir nehmen uns mal den zweiten Block, wie arbeitet eigentlich Technik mit Precels, wo geht das ganze Spielchen los, bis hin, da kommt jetzt irgendwann die Ausschreibung. Und da sind wir auch beim letzten Mal, wir sind nicht fertig geworden, wir haben ja einfach angefangen. Und genauso werden wir das für die anderen Teilprozesse auch beschreiben müssen, das wird nicht anders gehen.
SPEAKER_04: Ja, das habe ich jetzt gesagt, weil ich sehe, das sind zwei Projekte meiner Meinung nach, weil eine Sache ist diese technikischen Teil, wir wollen die Projekte einmal, also hier in diese, ich weiß nicht, dieses Akademium erstmal, also dokumentieren, auswerten und dann umsetzen und die andere ist dann, wie begleiten wir das für den Prozessmanagement. So sehe ich das.
SPEAKER_02: Ja, also das, also ich habe beim letzten Mal beschrieben, dass wir an dem Prozessmanagement schon seit über einem Jahr arbeiten. Das hat Lars im Protokoll etwas anders dargestellt. Das machen wir sowieso. Das ist auch überhaupt keine Frage, das hat auch nichts damit zu tun, ob wir jetzt diese Runde hier haben oder nicht. Wir müssen beschreiben, wie unsere Prozesse sind.
SPEAKER_04: Aber das läuft schon.
SPEAKER_02: Genau, das läuft schon und das läuft schon seit einem Jahr und das werden wir sukzessive abarbeiten und wir werden sicherlich auch noch, ich war ja letzte Woche bei einem Seminar, was durch die neue ISO 9000 auf uns zukommt, da wird es auch noch ein paar neue Forderungen geben, die müssen wir auch noch berücksichtigen und da werden wir auch alleine durch unsere organisatorischen Änderungen ein paar Sachen etwas detaillierter beschreiben müssen. Eine BBG gab es früher nicht. Das, was früher die alte BBG war, müssen wir in irgendeiner Form auch mit beschreiben. Also das hat aber alles hiermit nichts zu tun.
SPEAKER_04: Ja, und das hat Sie, wenn wir schon da sind, das ist schon vorbereitet für das BBG. Ich kann Sie das zukommen lassen.
SPEAKER_02: Ne, ich meine nicht die neue BBG, ich meine die alte BBG. Die neue BBG hat für uns, für mich jetzt hier mit dem Thema ISO 9000 und Prozessmanagement nichts zu tun, weil das ist ja eine separate Firma. Das, was früher BBG war, was jetzt EDD ist, das haben wir bisher in unseren Prozessen wenig beschrieben. Das werden wir in Zukunft etwas detaillierter beschreiben müssen.
SPEAKER_03: Okay. Aber dann, wie du sagtest, das heißt, das ist einmal das Thema Mission 2030 allgemein nach ISO 9000, dann muss ich gestehen, habe ich jetzt für mich noch nicht verstanden, wo und bei wem gerade der Ball liegt und wie der Stand ist bei dem Part, bei dem anderen Part, den du gerade befinden hast. Also bei dem, wie, dem Ablauf, wie kommen die Ideen hier rein? Ich meine, das fing ja an, das war jetzt nur einmal gerade aus meiner Perspektive. Wir haben das hier besprochen vor, ich glaube, mittlerweile sechs Wochen ungefähr, haben uns doch geeinigt, dass ich einen Vorschlag mache. Da hatten wir dann ja auch noch zusammengesessen. Den Vorschlag, der hat nochmal zur Diskussion geführt, zwei, drei Wochen später und da weiß ich gerade nicht, genau, dann habt ihr die Mail fertig gemacht mit eurem Vorschlag und dann war ich im Urlaub und ab jetzt ist das da für mich halt einmal der Punkt, wo ich gerade nicht weiß, wo das liegt, wie das ist, was da besprochen wurde.
SPEAKER_02: Wir haben hier, da weißt du glaube ich, den Urlaub, da haben wir hier drüber diskutiert und das Ganze ist ja losgegangen an der Diskussion. Wir haben ja im F&E haben wir einen sogenannten Ding-Pool und ich habe dann gesagt, naja, aus meiner Sicht ist das jetzt relativ einfach. Man könnte das, was wir hier vorhaben, einfach dort mit beschreiben als Prozess, als Unterprozess und eigentlich ist es in meinem Augen auch gar nicht viel. Ich habe angefangen zu versuchen, das zu beschreiben. Das können wir uns jetzt gerne angucken. wo wir da jetzt gerade stehen. auf einen kleinen Augenblick.
SPEAKER_03: Ja, weil in dem Protokoll ließ es sich wirklich ein bisschen vermischt von den zwei Punkten, muss ich gestehen.
SPEAKER_04: also deswegen habe ich so angefangen zu erklären, also wir haben tatsächlich, meiner Meinung nach, das sind zwei getrennte Projekte. Und wir haben am Anfang gesagt, dieses Gremium trifft sich einfach um das Projekt auszuwerten und das hat mit Prozessmanagement, meiner Meinung nach, nichts zu tun.
SPEAKER_03: Klar ist das verbunden.
SPEAKER_04: Ja, genau. Also aber die letzten zwei Termine, es ging mehr um Prozessmanagement statt diese...
SPEAKER_02: Also es ging nicht um Prozessmanagement, es ging darum, wo und wie beschreiben wir, was wir tun wollen. Und das gehört ins Prozessmanagement, aber ich kann im Prozessmanagement oder sollte im Prozessmanagement halt das beschreiben, wie wir arbeiten. Punkt. Und das, wie wir arbeiten, sollte gewisse Mindeststandards erfüllen und dann ist alles gut. Vergiss Prozessmanagement, das ist einfach nur das Cool, womit ich es beschreibe. Wir müssen uns darüber einig sein und im Klaren sein, wie wir es eigentlich abbilden, wie wir es leben wollen. Und dann geht es nur noch darum, das, was wir leben wollen, vernünftig abzubilden.
SPEAKER_04: Gut, und das, wie wir das leben wollen, das hat Business Development Vorbereitung, das habe ich ja mal präsentiert
SPEAKER_UNASSIGNED: und das habe ich ja mal präsentiert
SPEAKER_04: und das habe ich mitbekommen.
SPEAKER_03: Okay. Aber ich wusste nicht, ob das jetzt nochmal besprochen wurde oder ob das jetzt für alle so in Ordnung war.
SPEAKER_04: Das haben wir ja nicht besprochen, aber es war ja bei allen anderen hier. Und also das ist jetzt unser Input. Also ich habe auch keinen Antwort bekommen. Ich weiß es auch nicht. Das ist wohl mein letzter Stand. Also ich kann nichts mehr sagen. Weil innerhalb dieser Präsentation habe ich auch klar nochmal deutlich gemacht, dass wir ja tatsächlich jemand brauchen, der dann diese Umsetzung übernimmt. Wer kümmert sich von Projektmanagementseite für die Umsetzung? strategisch, wie wir das machen wollen. Das ist einfach nur, was die Geschäftsführung gesagt hat, einfach nur mal dokumentiert und mit Ideen, welches geht und so weiter. Aber einer muss das umfassen, oder einer muss sich kümmern, dass das wirklich stattfindet, zu koordinieren, zu managen. Wir brauchen einen Projektmanager dafür.
SPEAKER_02: Für was jetzt? Bin ich nicht ganz sicher, was du meinst.
SPEAKER_04: Also das ist jetzt ein bisschen das Problem, weil du warst in die ersten zwei Runden nicht dabei. Also diese Gremium müssen jetzt...
SPEAKER_02: Das sollte kein Problem sein, weil im Idealfall haben wir eine Beschreibung, die so ist, dass jeder sich sofort damit zurechtfinden kann. Wenn das nicht so ist, dann ist die Beschreibung nicht gut. Ja, sorry. Also ich denke jetzt mal im Sinne vom Prozessmanagement.
SPEAKER_04: Ja gut, aber ich habe das nicht beschrieben, weil also das ist... Also wir haben die Protokolle. Hast du die Protokolle gelesen?
SPEAKER_02: Ich habe die Protokolle gelesen, ja.
SPEAKER_04: Also diese Gremium müssen...
SPEAKER_02: Ich habe die nicht mehr alle vor Augen, also müsste ich dann nochmal wieder neu lesen.
SPEAKER_04: Also diese Gremium ist hier, der Stand von Geschäftsführung hat gesagt, wir...
SPEAKER_02: Du brauchst nicht immer betonen, dass das Stand von der Geschäftsführung ist. Wir arbeiten hier zusammen.
SPEAKER_04: Also wenn du willst, ich kann dir die Geschichte von Anfang an erzählen. Wenn du das nicht möchtest, ich das erzähle, dann vielleicht kannst du selber die Protokolle lesen. Das muss ich gar nicht sagen. Ich kann dir nur sagen, warum ich hier sitze.
SPEAKER_02: Das ist der Satz, deswegen der Geschäftsführung. Ich dachte, wir arbeiten hier zusammen. Habe ich da irgendwas falsch verstanden?
SPEAKER_04: Also ich habe es nicht falsch verstanden, aber du hast... Nun, wir würden folgst und was ich sagen. Ich kann das sagen, wie ich das möchte. Ich habe recht dafür, dass du sagen, wie ich das möchte. Ja. Aber ich habe nicht einmal gesagt, ich akzeptiere, was du sagst.
SPEAKER_02: Also, mein Kenntnisstand ist, dass wir beschreiben wollten, wie wir die Projekte hier reinfliegen und wie wir sie dann zuordnen. Nicht, wie wir einzelne Projekte abwickeln. Habe ich dann irgendwas falsch verstanden? Das war mein Kenntnisstand. Nochmal vom Wording. Also, es ging doch darum, dass von irgendwo aus verschiedenen Stellen im Unternehmen oder aus den Prozessbeteiligten Ideen hier eingebracht werden, wo dann diskutiert wird, macht es Sinn, die weiter zu verfolgen oder nicht?
SPEAKER_00: Ja.
SPEAKER_02: Und wenn wir der Meinung sind, dass es Sinn macht, die weiter zu verfolgen, auch zu definieren, wo sie weiter verfolgt werden. Genau.
SPEAKER_03: Eine Zuordnung an einem Projektleiter oder wer noch immer. Das war mein Verstand.
SPEAKER_02: Das war das, was ich verstanden habe, warum es hier jetzt eigentlich gehen soll, dass wir das beschreiben beziehungsweise definieren. Genau.
SPEAKER_03: Und ich hatte mich, weil es einen alten Prozess gab, über den wir gesprochen haben, hatte ich mich im ersten Schritt an diesem alten Prozess, nicht um jetzt einen Prozess abzubilden, aber einfach, da war ja schon mal ganz schön beschrieben, wie man Ideen so durch ein Unternehmen eben durchschleusen kann. Habe den als Aufschlag genommen und was ergänzt, was wir in der Diskussion besprochen hatten. Also das halt einfach, BD-Projekte haben wir fehlt, die mussten dann noch kurz beschrieben werden. Aber ich habe es nicht in dem tiefsten Detailgrad gemacht, wohlwissend, dass wir auch einfach ein Gefühl dafür kriegen müssen, je nachdem was für Projekte da auf den Tisch kommen. Das konnte ja wirklich von wir brauchen einen neuen Aufsatz für den Stapler sein, bis hin zu wir wollen mit Produkt XY auf den neuen Markt oder mit System XY oder mit einem ganz neuen Produktidee in der Region, wo wir noch nicht waren, wie auch immer, das ist jetzt die Funktionssache. Das muss ja alles irgendwo abbildbar sein und das kann man natürlich nicht im Detail beschreiben. Das heißt, ich habe versucht, die Filterkategorien anzupassen an die Sachen, die sich in den letzten zwei Jahren verändert haben und das so zu beschreiben. Genau. Und dann kam halt die Diskussion und das, genau, die Präsentation von dir, was von dem, was ich das gelesen habe, eigentlich in eine ziemlich ähnliche Richtung ging, wie was das andere auch beschrieben hat, mit ein, zwei Punkten, die vom Detailgrad tiefer waren. Aber im Großen und Ganzen ist die Frage, die ich mir jetzt stelle, wir sitzen hier jetzt sechs Wochen ungefähr zusammen seit dem ersten Aufschlag. Ich denke, dass mit dem Abbilden müssen wir meiner Meinung nach einfach zum Ende kommen, uns irgendwie einigen und dann eine Lösung zu finden, dass wir das einfach anfangen können. Das ist zumindest meine Meinung. Ich meine, unabhängig davon, ob man es jetzt als Präsentation oder als Prozess abbildet, das ist ja lebend. Also wenn wir in zwei Monaten merken, dass etwas überhaupt nicht passt, dann müssen wir es ja wieder anpassen, grundsätzlich. Zumindest meine Meinung.
SPEAKER_02: Also ich habe versucht, warte mal, wir machen das jetzt am besten, sollte ich mich über Teams einreden. Ja, ich denke, das ist ein Kastellon. Ja.
SPEAKER_03: Ich habe mich hier das Mikro ein, und das ist ein Kastellon.
SPEAKER_UNASSIGNED: Das ist ein Kastellon. Das ist ein Kastellon.
SPEAKER_03: Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Und das ist ein Kastellon.
SPEAKER_04: Und das ist ein Kastellon. Und das ist ein Kastellon.
SPEAKER_03: Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon.
SPEAKER_UNASSIGNED: Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon.
SPEAKER_02: Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon.
SPEAKER_01: Das ist ein Kastellon. Das ist ein Kastellon.
SPEAKER_UNASSIGNED: Das ist ein Kastellon.
SPEAKER_02: Das ist ein Kastellon. Das ist ein Kastellon. Das ist ein Kastellon.
SPEAKER_UNASSIGNED: Das ist ein Kastellon.
SPEAKER_02: Das ist ein Kastellon. Das ist ein Kastellon.
SPEAKER_UNASSIGNED: Das ist ein Kastellon. Das ist ein Kastellon.
SPEAKER_02: Ich habe versucht, in dem F&G-Prozess, das ist der Prozess, so wie wir ihn heute fast haben. Ein paar kleine Änderungen schon. Also wir hatten bisher hier vorne einfach das Thema "Ideenpool".
SPEAKER_03: Ja. Willst du mit rüber kommentieren mit rauf gucken, das kann man deutlich besser sehen, Johanna. Willst du mit rüber kommentieren mit rauf gucken, das kann man deutlich besser sehen.
SPEAKER_02: Also statt dass wir hier den Ideenpool haben, habe ich hier jetzt einfach einen Unterprozess eingeführt. Den habe ich genannt "Go-To-Market Hub", so wie wir das ja auch als Namen besprochen haben. Und hier drin in der Beschreibung habe ich im Wesentlichen das reingenommen, was Lars auch schon als Texte verteilt hatte. Sorry, das ist ein bisschen klein. Alle zwei bis vier Wochen findet eine Abstimmung, einen Austausch über neue Themen, neue Initiativen, neue Produktideen, neue Marktideen. Hieran nehmen mindestens teil, F&E, WM, Business Development, Marketing, Qualitätssicherung. In dieser Runde werden Themen besprochen, die perspektivisch marktrelevant sind. Wir haben in der Verbesserung der Kommunikation der vertriebsunterstützenden Abteilungen ein Meeting eingeführt. Koordinierung, vertriebsvorbereiten der vertriebsunterstützenden Prozesse, Koordinierung, Priorisierung. Im Prinzip ist es das, ich sage mal, ich habe das einfach rauskopiert aus dem, was Lars beim letzten Mal verteilt hat. Und daraus waren letztendlich diese vier Themen, wo ich, ich habe es jetzt einfach mal rüberkopiert. Ich bin mir inhaltlich momentan nicht sicher, ob das so sinnvoll ist und so passiv ist. Ich sage mal, als Beispiel, neue Märkte, weiß ich nicht, ist neue Märkte im Entwicklungsprozess, den wir als Entwicklungsprozess unter F&E sehen. Da war ich mir jetzt, nachdem ich das gelesen habe oder neu gelesen habe, nicht mehr sicher, ob das so an der Stelle richtig ist. Also ja, klar, neue Produkte ist mal auf jeden Fall F&E-Prozess, neue Märkte, weiß ich nicht.
SPEAKER_04: Deswegen habe ich das letztes Mal auch gesprochen, deswegen habe ich das zweimal auch gesagt. Das ist kein F&E-Prozess mehr, sondern wir haben gesagt, das bleibt nur in der F&E, wo das war. Weil ich muss nochmal sagen, pünktmal, die gewünscht von Geschäftsführung war, dass die neuen Märkte auch hier diskutiert werden in der Runde. Und du warst auf den ersten Termin nicht dabei und die Geschäftsführung hat gesagt, es gibt die bestehenden Märkte und die neuen Märkte. Bestehende Märkte betreibt weiter Betrieb und für die neuen Märkte, alles, was wir über die neuen Märkte diskutieren, da können wir hier auf den Tisch legen und zusammen diskutieren. Und deswegen ist das jetzt nicht ein F&E-Prozess, sondern ist ein Prozess für die gesamten Unternehmen. Aber dann hat Martin gesagt, und ich habe damit leben, das war für mich in Ordnung, weil das schon war in F&E, sollte in F&E bleiben. Aber grundsätzlich macht das nicht in F&E, das ist was anderes, weil es geht um die Geschäftsentwicklung und Betrieb, dass es hier anfängt, so mit dieser Gremie. Es liegt nicht um die Erweiterung und bestehenden Prozess, das schon existiert und das funktioniert, meiner Meinung nach, sondern es geht um die Umsetzung von Strategie oder von Mission 2030. Und deswegen war der strategische Teil von Geschäftsführung und Erweiterung basierend auf dessen, was wir auf der Strategiesitzung gehört haben.
SPEAKER_02: Ja, so jetzt ist die Frage, ich sag mal, Björn, Martin, wie seht ihr das vom Verständnis her? Also zumindest mal, Björn und ich stellen das zumindest in Frage, ob das hier überhaupt mit Entwicklungsprozess richtig beschrieben ist oder geht es uns? Und das wäre jetzt die Frage, geht es uns nur da, nur in Anführungsstrichen darum, hier in der Runde das sozusagen zuzuordnen, das auf die Liste zu packen von allen möglichen Prozessen, wo dann an der Stelle eben steht, wird nicht in der Entwicklung, sondern wird dann so anders bearbeitet. Also das ist für mich die Grundsache. Also einfach nur, dass wir überhaupt ein gemeinsames Verständnis dafür kriegen.
SPEAKER_01: Genau, so sehe ich das. Also ich will mich überhaupt gar nicht im Entwicklungsprozess, im Entwicklungsprozess im Sinne einer F&E-Entwicklung mit Business Developing beschäftigen. Das ist nicht meine Aufgabe, ist nicht meine Kompetenz, kann ich nicht, brauche ich auch nicht. Nur wir waren uns ja einig, dass der Eingang unter Umständen nicht, beim Eingang von irgendetwas unter Umständen noch nicht klar ist, in welchem Trichter es dann weiterfällt. Und deswegen haben wir gesagt, okay, wir nehmen halt einen bestehenden Prozess, wo bereits Ideen gesammelt werden. Und dann, ob es dann in Swimlanes aufgeteilt wird, wie du das damals genannt hast, glaube ich, oder ob wir dann in Unterprozesse zerfallen, das ist von meiner Seite egal. Oder ob wir den Prozess namentlich auch so verallgemeinern, dass wir eben alles in diesem Prozess machen können, was irgendwie eine Entwicklung im weitesten Sinne ist. Eine Marktentwicklung ist ja auch eine Entwicklung. Es muss ja nicht ein Produkt sein oder ein Design oder so. Also da kann ich mit beiden leben. Aber nein, es ist nicht originär im Entwicklungsprozess, wie er bisher war, richtig angeordnet, aus meiner Sicht.
SPEAKER_03: Ich verstehe die Bezeichnung jetzt auch so. Sorry, nicht, dass ich, das ist immer blöd, wenn man bemerkt, wo man nicht dabei war. Aber ich verstehe das jetzt einfach so, dass es bedeuten soll, wir führen die drei Punkte durch diesen, wie hieß er, auch 0,8. Und alles andere, bestehende Märkte, Anwendungen, Produkte durch diesen Neubau, den wir, wo wir gerade eben drüber gesprochen haben, diesen Punkt 1, durch den Prozess. So habe ich das jetzt verstanden. Das teilt sich halt auch, aber letztendlich, ja, es ist doch einfach nur ein Trichter. Und danach verteilen wir es ja hier gemeinsam an den Projektverantwortlichen. Das, was, so wie ich es verstanden habe, eine der Anforderungen war. Also einmal, dass wir untereinander informiert sind und jeder weiß, welche Projekte unabhängig, wo er im Unternehmen sitzt und wo es bearbeitet wird, weiß, dass was passiert, dass nicht Dinge doppelt bearbeitet werden. Und dass Leuten gut aufgesetzt wird, um dieses Projekt wirklich durchzuführen. Und das war ja jetzt hier sehr zu sagen, das machen wir doch auch.
SPEAKER_04: Ja, aber um das zu kommen, so Prozess zu kommen, müssen wir uns erstmal einigen, sind wir im strategischen Teil einverstanden. Weil Henning hat jetzt gesagt, ist es so, bestimmte Märkte, neue Märkte, neue Anwendungen, strategische Innovation. Also für mich kommt Prozessmanagement am Ende, wenn alles schon hier abgestimmt ist, wie wollen wir weiter zum Unternehmen machen. Dann kann man weiter, weil die kann ich jetzt zum Beispiel Betriebe klären, am Ende mit Prozessmanagement, also was wir hier teilen. Wie kann jetzt Betriebe mitmachen, wenn ich jetzt komme und sage, deswegen war ich tatsächlich nicht einverstanden, warum das Betriebeforschung bleibt. Weil ich finde, es ist irgendwo, weil ich jetzt denke, für mich ist Research und Gewalt, das ist Wissenschaft. Und jetzt reden wir hier, so muss ich die strategische Innovation, das ist Forschung und Entwicklung, aber mit dem Ziel, dass ich Betriebeaktivität optimieren kann. Deswegen würde ich sagen, wir sollten den Prozess kopieren und unter anderem Namen nennen und dann nur strategische Innovation unter Forschung und Entwicklung probieren. Aber bevor wir zum Prozess kommen, muss ich sagen, wir müssen erstmal einig, von strategischer Zeit, der einverstanden, dass der Weg nicht so weit ausgetürt wird.
SPEAKER_02: Aber dadurch, da steht ja nur F&E, weil die Verantwortlichkeit für den Ablauf in der, also ich habe die Lösung, das ist relativ einfach. Also, Johanna sagt, dass im Prinzip, du störst dich daran, dass das jetzt im F&E-Prozess ist. Das ist aber inhaltlich überhaupt kein Problem, weil das, was ich jetzt hier eingefügt habe, ich mache es mal hier auf, dieses Kästchen, was auch immer wieder reinsteigt, Das kann ich zum Beispiel an zwölf verschiedenen Stellen in den Vision 2030 Prozess einbinden, das kann ich an drei Stellen im PM-Prozess einbinden, an fünf Stellen, ich kann das überall, wo ich sage, da können Ideen kommen, bla bla bla und so weiter und so fort. Und nach meinem Verständnis geht es hier jetzt ausschließlich darum, und wir bilden das jetzt gerade hier ab, das können wir hinterher auch genauso woanders machen, geht es darum, dass wir sagen, wie treffen wir die Entscheidung. Hier vorne, hier vorne kommt irgendwas rein, ich wäre jetzt mittlerweile so weit, dass ich sagen würde, wir schmeißen dieses Kästchen hier komplett raus, perspektivisch, aber ich muss ja immer leider ein bisschen rum springen, also dieses Kästchen schmeißen wir hier komplett raus, das ist nur irreführend und ist eigentlich auch in meinen Augen gar nicht hilfreich. So, wir haben dann hier das Kästchen Ideenpool, da kommt irgendwas rein, wir diskutieren das hier in der Runde, wir fragen eventuell, ergänzen wir Informationen an, weil wir sagen, sorry, das ist nicht ausreichend, da können wir noch gar keine Entscheidung treffen. Und dann wird in dieser Runde eine Entscheidung getroffen, wo federführend, in welchem Bereich dieses Thema weiterverfolgt wird.
SPEAKER_03: Genau, das heißt die Verantwortung zu dem Zeitpunkt, die gibt es noch gar nicht. Ja, die wird leider definiert, genau, für das Projekt, genau. Die Verantwortung, die im R&D liegt, ist nur das Einsammeln der Ideen, aber wir können ja genauso in den, also ich als Person, als PM-Person oder du als PD-Person, du hast ja deine eigenen Ideen, aus welchen Inputs auch immer. Es geht ja nur darum, die hier zusammenzutragen und dafür ist das R&D verantwortlich und danach geht es weiter.
SPEAKER_04: Aber ich bin immer noch zur Meinung, dass das nicht der richtige Weg ist, weil ich kann überhaupt nicht mehr vorstellen, dass zum Beispiel jemand von zum Beispiel Vertrieb hat Toll-Idee für die neuen Anwendung. Genau. Oder, ja, Entschuldigung, man macht.
SPEAKER_03: Ja.
SPEAKER_04: Er geht nicht zur Forschung und Entwicklung.
SPEAKER_03: Nein, nein, nein, nein, genau. Er kann das, er kann das, er kann das, grundsätzlich kann er das entweder über diese E-Mail-Adresse melden, die in dem Prozess stand, die DNF, kann jeder dann melden, wo er möchte, oder er spricht mit irgendwen. Und es geht ja nur darum, dass wir das wieder hier sammeln, damit es nicht mehr, damit Raeck als Beispiel, das ist einfach ein gutes Beispiel dafür, nichts wie Raeck, ich mag, dass er so ruhig ist, aber dass er nicht mehr zu Martin gehen kann. Wenn Martin nein sagt, fragt er zwei Tage später mich und ich sage dann vielleicht ja, sondern wenn Martin das gehört hat, ist es spätestens alle zwei Wochen halt in diesem Themenpool mit einer Bewertung.
SPEAKER_02: Was man hier jetzt, was man hier jetzt geschickterweise machen könnte, wäre, man fügt noch eine Schwimmbahn ein, die heißt GPM-Hack.
SPEAKER_04: Genau.
SPEAKER_02: Und dann sagt man, da in der Runde läuft das Thema auf und es kann auch, also das Beispiel, was du gerade gebracht hast, Martin könnte dann sagen, ja, super, Idee-Reit, nämlich wird in die nächste Runde. Genau. So, und dann wird es in der Runde diskutiert und dann wird es entschieden, ob es weiter verfolgt wird oder nicht oder ob wir noch weitere Informationen brauchen und dann geht es eben aus der Runde zurück in irgendeine von diesen anderen Schritten. Stimmenwahn, wer auch immer den Hut aufbricht und der, der den Hut aufbricht, heißt ja auch nur, dass er es verantwortlich macht. Der macht es ja auch nicht alleine, sondern bindet die anderen Abteilungen so, wie es nötig ist, entsprechend mit ein. Und wenn ich es richtig verstanden habe, aus dem heraus, was Lars mit von der ersten Runde erzählt hat, war dann die Idee, dass dann diese Liste der Themen sozusagen ist. Die sollen in der F&E gepflegt werden, weil da am ehesten die Kapazität da ist. Es geht nur um die Liste. Sag mal, was für ein Thema haben wir? Wie ordne ich das möglicherweise zu? Also, was ist das für ein Themenkomplex? Und wer ist verantwortlich? Punkt. Genau, aber es hat keinen Einfluss.
SPEAKER_03: Also, es gibt halt unterschiedliche Wege, wie du das Thema in der NGTN-Kampf reinbringen kannst. Also es geht nur um die Verantwortlichkeit des Sammelns und was halt die F&E auch machen soll. Bis vor einmal schon mal gucken, wie viele Informationen liegen denn vorab. Schon mal einmal diese Korrekturschleifnisse machen mit, komm, gib mir doch mal ein bisschen mehr Input, ein bisschen mehr Futter zu der Idee. Da hatten wir ein, zwei Sachen, wie viel, dass wir dann, wenn wir hier schon zusammengesetzen mit anderen, wenigstens genug Informationen zu der Idee auch vorliegen haben. Wenn die Person das dann in der Lage ist, zusammenzufragen, kommt ja auch noch ein bisschen drauf an.
SPEAKER_01: Ja, aber es geht da ja auch nicht um Inhaltliches, sondern das kann ich ja, da muss ich überhaupt gar nicht wissen, ob das vernünftig ist oder nicht. Aber wenn einer nur mit einem Einsatzidee kommt, dass das nicht ausreicht, das kann einfach jeder entscheiden.
SPEAKER_02: Da würde ich jetzt auch sagen, das müsste jetzt nicht unbedingt irgendjemand aus der F&E machen, der möglicherweise jeder gar nicht dabei ist, sondern im Prinzip könnte das auch eine Standard-Checkliste sein, wo da einfach einen Haken dran machen, was wir wollen. Habe ich die wissen, die Informationen, oder habe ich sie nicht in der Testen, nochmal nachfragen, genau. Bevor ich das hier in der Rumi-Walks-Thema mache, nochmal, nochmal, die Sache, dass F&E da stand, war einfach nur,
SPEAKER_01: dieser jemand ist in unserem Unternehmen recht schwer zu finden. Also im Prinzip habe ich gesagt, okay, zu dem Zeitpunkt hatte ich ja noch zwei Mitarbeiter, die das machen konnten. Die machen das einfach. Die haben eine ganz kurze Checkliste und fragen noch zwei, drei Sachen ab. Und die treffen auch keine Entscheide, sondern die fragen einfach nur, kannst du mir kurz dazu noch sagen, in welchem Markt du dir das vorstellst? Das ist jetzt auch nur ins Blaue gesprochen, in welchem Markt du dir das vorstellst, was auch immer. Und sammeln das zusammen, schreiben das auf einen virtuellen Zettel und dann haben wir ein bisschen mehr als einfach nur eine Einsatzidee. Mehr nicht. Das kann im Prinzip jedes Sekretariat machen. Nur die reißen sich nicht da drum.
SPEAKER_03: Und genau bei dieser Arbeit hatte ich halt gedanklich, wir hatten ja auch ganz am Anfang daran gesprochen, dass der einzige Grund, warum es am Anfang in der Art und die Swimland stand. Und danach geht es ja in diese Runde mit allen zusammen und dann die Zuordnung, je nachdem, wo das Projekt halt hingehört. Das war der ganze Gedanke dahinter. Ich hoffe, dass wir das halt recht einfach weiterhalten können, weil ich meine, da war es noch ein bisschen anfangen.
SPEAKER_02: Ich bin mir nicht ganz sicher, ob ich dich von richtig verstanden habe. Du hast von der Präsentation gesprochen, die du verteilt hast. Die Präsentation beschreibt, meine ich, wenn ich es richtig in Erinnerung habe, im Wesentlichen, wie ihr Projekte in Business Development bearbeitet. Nein. Okay, dann habe ich falsch in Erinnerung.
SPEAKER_04: Die Präsentation beschreibt, wie wir das von Geschäftsverlust bestanden haben, wie die Investition 2030 umsetzen sollen und Vorschlag, wie wir das umsetzen sollen im Unternehmen. Deswegen gibt es diese unterschiedlichen Kategorien, also bestehende Märkte, wo Betriebsverlust ist, neue Märkte, diese Developing, neue Anwendungen, Produktmanagement und dann strategische Innovation ist dann konstrund entwickelt.
SPEAKER_03: Das ist bei uns dann sozusagen der Punkt, wo wir hier in der Runde entscheiden, okay, ob der Plätze sind wir auch, was ist ein Produkt oder ein Projekt, die Idee, die in einer der Runde geht, dann kriegt die Abteilung und das sind sozusagen wieder Kriterienkatalogen und danach dann entscheidest du und wer kriegt die gut auf und gut.
SPEAKER_01: Okay. So von meinem Verständnis. Das Ganze ist übrigens auch Schattenkämpfen, weil die Anzahl an Ideen, die irgendwie aus Out of the Blue kommen, die wird sehr, sehr gering sein. Die allermeisten Projekte werden aus unserer Runde mehr oder minder kommen und dann ist dieser erste Teil sowieso eher irrelevant.
SPEAKER_00: Also ich habe mich jetzt ganz mal ein bisschen zurückgelegt. Ich habe mich jetzt ganz mal ein bisschen auch angeregt und versucht auch so ein bisschen für mich das zu interpretieren, warum gibt es diese Runde überhaupt und was war die ursprüngliche Intention, dass wir hier zusammensitzen und so weiter. Das, was mir komplett in dieser Runde auf sich fehlt, ist im Grunde genommen, was wir immer vergessen, ist die Geschichte, also warum wir überhaupt hier zusammengekommen sind. Sondern der eine Intention war ja, es entstehen hier irgendwo Innovationen, Ideen oder Sonstiges, die dann halt, ich sage mal so, in den einzelnen Abteilungen oder bei den einzelnen Personen weiterverfolgt werden, weiterbearbeitet werden und irgendwann stehe ich das mal bei irgendeinem auf dem Tisch und derjenige muss das weiter bearbeiten. Und das war ja unsere Intention, das so ein bisschen mal zu kanalisieren, mal zu bündeln, wo wir gesagt haben, okay, in dieser Runde treffen wir uns und besprechen dann halt irgendwelche Sachen, die wir links und rechts irgendwie aufgefasst haben oder die wir halt mitbekommen haben oder wo wir involviert sind, schmeißen wir alles mal in den Topf rein, um so ein bisschen Transparenz zu sehen. Das war die erste Sache. Die zweite Sache ist, das was viel läuft in dieser ganzen Diskussion und wenn ich auch nur da drauf gewesen bin, das kennt kein Pro, ich bin ja auch mit dabei, vermisse ist, wir vermissen die Geschäftswertung und vor allen Dingen bei diesen ganzen Sachen, die wir dort überdenken, wir reden hier ja nicht nur über Produktideen oder Innovationen, sondern wir reden hier auch über neue Märkte und so weiter und da fehlt von meiner Seite aus am Anfang überhaupt der gesamte Strategie-Check davor. Weil wir sind hier schon sehr, sehr tief mit drin von wegen, ja, wie kriegen wir die Produktinnovation hier durch, aber wir wissen ja noch gar nicht, ob das in unsere Strategie hineinpasst und wir wissen ja auch alle und das war ja auch mit einer Intention, weswegen wir hier zusammensitzen, dass zum Teil dann, ich sag mal, da oben, von mir aus gesehen, da oben Entscheidungen getroffen werden, die wir dann irgendwann im ersten Halbesjahr später erfahren, weil dann irgendwelche neue Mitarbeiter da stehen oder neue Ideen da sind oder neue Märkte da sind und so weiter. Und dieser Strategie-Check, der muss vorher auch aberfolgen und der muss auch mit der Geschäftswertung erfolgen, weil es macht doch keinen Sinn, dass wir hier tief einsteigen mit Forschung und Entwicklung oder sonst wo, wenn die Geschäftsleitung eine ganz, ganz andere Idee hat oder vielleicht schon irgendwo schon aktiv ist oder halt auch andere Menschen oder Abteilungen davon aktiv sind. Das heißt also, das dürfen wir nicht außen vorlassen. Und ansonsten müssten wir an und für sich diesen Prozess als solches gar nicht so aufsetzen mit, ich verstehe auch nicht, Martin, Entschuldigung, wenn ich etwas falsch sage, aber ich zum Beispiel, die Forschung und Entwicklung ist da irgendwie am Anfang mit drin. Nein, Forschung und Entwicklung ist nur ein Detailbereich. Irgendwo mal da hinten irgendwie sich im Prozess daran teilnimmt. Da sind erstmal ganz, ganz andere Abteilungen oder Wissen halt auch gefragt und das muss in diesen Prozess einfließen. Und ich glaube, das dürfen wir nicht außer Acht lassen. Ich glaube, wir müssen dann vielleicht auch mal in diesem Falle auch mal vielleicht auch unsere Brille absetzen, um halt auch besser zu werden, um halt unsere internen Prozesse, die wir vielleicht über Jahre sich aufgebaut haben, mal abzulösen, um da halt auch mal einen neuen Weg zu gehen, einen professionelleren, strategischen Weg, als das, was momentan halt stattfindet. Und deswegen finde ich so ein bisschen so die Entwicklung, in den letzten Meetings war ich nicht dabei, da war ich krankheitsbedingt raus, beziehungsweise ohneausbedingt raus, aber ich vermisse ich da halt mal ein bisschen dran. Vielleicht sehe ich das auch falsch.
SPEAKER_01: Ja, das siehst du falsch. Zumindest ein Detail davon. F&E hat in diesem Prozess, genau wie du beschreibst, einen Teil, nämlich alles das, was dann wirklich Entwicklung ist. Wir brauchen nur ein Sekretariat, was am Anfang einmal sozusagen die von außen einkommenden Sachen ganz kurz auf formale Vollständigkeit oder zumindest auf minimale Anforderungen prüft und die einfach eingleist. Und das ist nur rein zufällig F&E. Das kann von mir aus auch Martina Nagel machen. Schlechtes Beispiel, weil die geht in Rente. Aber das kann jeder tun. Nur du wirst im Unternehmen keinen finden, der da Bock drauf hat. Deswegen habe ich gesagt, okay, das machen wir vorne weg. Streich da einfach mal das F&E weg. Ist ein Sekretariat. Irgendjemand muss diese fucking E-Mail-Adresse lesen und gucken, was da drin steht.
SPEAKER_00: Mehr ist es nicht. Nee, ich meine das Gleiche. Wir meinen doch das Gleiche. Das ist nur ein Beispiel. Wie gesagt, wir müssen jemanden haben, der sich dort darum kümmert, der das, ich sage mal so, abteilungsübergreifend halt organisiert, ob es ein Projektmanager später ist oder wie halt auch immer. Aber letztendlich, was ich damit halt nur meine, letztendlich, wir sind immer sehr, sehr ganz schnell tief in irgendwelchen Sachen unterwegs. Und ich glaube, wir müssen erstmal, ich sage mal, global den Prozess, und wenn das erstmal nur vier oder fünf Schritte sind, mit anfangen, wie sieht so ein Prozess bei Now aus, oder wie sollte er aussehen? Und da müssen wir halt die Geschäftsleitung auch die Strategie noch mit reinbringen, weil wir sind ganz, ganz schnell sofort in irgendwelchen Details drinnen. Und das ist ja gerade das, was wir an und für sich auch aufheben wollten mit dieser Runde, dass da nicht schon was losgetreten wird und da hinten ist schon jemand unterwegs, das ist schon 20 Kilometer entfernt, sondern wir wollen ja am Anfang dabei sein und darüber entscheiden und das müssen wir ja halt auch, weil halt auch dann später auch alle Abteilungen ja auch dann involviert sind. Später werden es um neue Märkte geht, ist Marketing involviert, ist dann irgendwie auch dann wieder auf FUE dran involviert, ist dann Produktmanagement mit involviert, ist Vertrieb mit involviert, wir müssen erstmal Sachen überhaupt erst mal am ersten Schritt abklären, bevor wir so tief mal einsteigen. Und das war mein Grundgedanke, wo ich einfach sage, jetzt irgendwie, dass es doch den Prozess mal erstmal so gut bleibt, erstmal viele Hauptschritte mal definieren, wie gesagt, dieser strategische Kurzcheck, der sollte meisterachtens am Vorderstelle irgendwo stehen, weil da vielleicht Sachen dann schon irgendwie abgelöst werden, weil es da vielleicht schon ganz, ganz andere Ideen gibt und wie gesagt, das war ja auch so ein bisschen so die Intention halt dieser Runde halt, um halt alles mal zusammenzuschmeißen und dadurch auch mal die Informationen, die wir alle haben oder wo wir was von erfahren, mal so ein bisschen transparent werden. Das müssen wir natürlich halt auch im Prozess dann halt auch abbilden. Ich hoffe,
SPEAKER_02: ihr könnt mich verstehen. Also gehört haben wir dich auf jeden Fall. Verstehen?
SPEAKER_04: Wenn ich jetzt darf, ich sage es mal ganz klar. Ich versuche schon das viertes Mal was zu sagen. Darf ich jetzt bitte?
SPEAKER_00: Ja, ich lasse mir nur vielleicht, Frau Rosen, einen kurzen Satz nur, nur von der Erklärung her. Worum es mir grundsätzlich geht, ist an und für sich, wir sollten uns überlegen, warum wir jetzt zusammengekommen sind, warum es diese Runde überhaupt gibt. Und also, wenn wir ganz ehrlich sind, haben wir selber gemerkt, dass es einmal irgendwie ständig irgendwelche Bestrebungen gab, irgendwelche Aktionen gab, die von hinten aufgeholt werden, wo wir dann diese Runde zusammenwitzen, ich habe gehört, das und das ist jetzt irgendwie unser neuer Plan. Und das war die Intention dieser Runde. Und so sind wir dann später dann halt über F und E und es gibt ja schon einen Prozess und hin und her irgendwie dahin wie hingestolpert. Nur dann sollten wir doch erstmal alles wieder zurückholen und mal wirklich sagen, okay, jetzt endlich, was wollen wir eigentlich erreichen und dann von dort aus an wieder runterbrechen und nicht jetzt schon wieder in Details reingehen, wo wir sagen, ja, okay, wo haben wir hier einen Ideenpool, wer zieht das damit an, welche Abteilung müssen wir damit reinbrechen, weil es bringt uns nicht weiter. Meine Meinung. So, jetzt bin ich fertig. Oh, Mann. Ja, fertig.
SPEAKER_03: So, wir haben keinen Sprechfall.
SPEAKER_04: Alles ganz gut. Also, eine Frage für Sie. Also, wenn Sie reden, wir müssen das tägliche Teil erstmal festlegen. Wir haben Strategiesitzung im Juni gehabt und wir waren alle dabei. Also, haben Sie als Marketingabteilung vielleicht etwas erarbeitet für den täglichen Teil, dass Sie vorschlagen können, weil Geschäftsentwicklung hat das jetzt vor drei, nee, vier Wochen einmal ein erfülltes Vorschlag gemacht und gezeigt, wie wir denken, dass die Strategie, was Geschäftsführung geklärt hat und gesetzt sein sollte. Also, wir haben das gemacht. Keiner hat uns gefragt. Wir haben das, wir haben uns dessen und einfach nochmal definiert, basierend auf dessen, was wir gehört haben, auf Sprechsitzung. Also, ein Dokument existiert. Dann sind die anderen Abteilungen gefragt, bitte, bringt die Kommentaren. Wenn keiner was sagt, heißt es für mich, okay, ich bin einverstanden. Also, das ist Frage Nummer eins. Die andere Kommentare, was ich jetzt habe, ist das jetzt vielleicht, lasse ich nicht dabei. Aber ich muss das so sagen, er hat Hennig mit Prozessmanagement eingebunden, weil ich glaube, er wollte zwei Sachen gleichzeitig regeln. Eine Sache wäre Strategie und andere, wie können wir das im Prozessmanagement erbinden. Wir haben uns tatsächlich jetzt mehr mit Prozessmanagement beschäftigt, obwohl ich denke, wir müssen erstmal strategische Teil festlegen. Das haben wir noch nicht gemacht. Und klar, wir können jetzt alle kritisieren, was alles fehlt, aber die Frage ist, welche Abteilung hast du am Tisch gebracht?
SPEAKER_03: Ich würde gerne noch einmal dazu was ergänzen. So wie ich das verstanden habe und ich lasse jetzt mal bewusst die Abteilung weg, ich gehe jetzt nochmal, auch wenn ich mich jetzt gedanklich an den Prozess dran langhangeln, bitte nagel ich mich jetzt nicht drauf. Aber die Idee und auch mit dem Kernproblem, das Björn jetzt nochmal angesprochen hat, alle sind informiert, wir wollen jemanden den Hut aufsetzen und so weiter. Das ist für mich der Ausgangspunkt, weswegen wir zusammenkommen. Wir sammeln von überall Ideen. Punkt. An ein Wort. Und diese Ideen werden einmal von einer Person überprüft, ob das mehr als einfach nur ein Satz ist, sondern ob noch ein bisschen mehr Information das angemessert werden kann. Damit, wenn sich eine Runde mit hochbezahlten Leuten damit auseinandersetzt, wenigstens eine gewisse Basis an Informationen vorliegt. Dann, und das wird auch so aufgeführt, und ist auch so aufgeführt in den Testen. Gibt es Kriterien? Welche Art von Projekt ist es? Ist es die Management-Projekt? Oder neue Märkte? Wie auch immer, wie rum man die Bezeichnungen jetzt machen wollte. Und da war auch Dinge drin, wie beispielsweise, jetzt nagelt es nicht genau fest, 5.120.000 Jahre oder was hatte Herr Peter gesagt? So ein Pima-Norm oder sowas. Also auch diese Strategien, strategischen Kriterien, die wir mal für Bank haben. Oder die mal da einen anderen Moment wurden. Diese Kriterien, und das war auch so mit abgebildet, oder es ist auch immer noch, stehen da mit drinne. Die überprüfen wir. Und wenn wir dann hier sagen, okay, es geht so und so weiter, dann geht es an der Person weiter. Ich habe nicht das Gefühl, also, oder ich habe hier das Gefühl, besser so gesagt, als sind wir eigentlich, kommt nicht einer Meinung, schafft uns aber gerade nicht, über das
SPEAKER_02: Gleiche wirklich zu reden.
SPEAKER_03: vielleicht ist es einfach nur ein Gefühl von mir. Natürlich kann man manche Sachen detaillierter aufschreiben, aber ich glaube, grundsätzlich wollen wir doch das Gleiche.
SPEAKER_04: Dass wir alle gleich wollen, das ist keine Frage. Klar wollen wir alles gleich. Wir wollen alle ein gutes Unternehmen. Das geht nicht darum, sondern es geht einfach nur, warte bitte, es ist einfach so, du siehst das von Produktmanagementseite, dieser Prozent, das geschrieben ist, ist von Produktmanagementseite und wir haben jetzt gesagt, vielleicht sollten wir einfach nur die Blickweise ändern, dass wir schauen, von drauf, was Markt will, was im Stakeholder war, wo ist jetzt, wo bewegt sich jetzt alle in der ganzen Welt und dann kommen wir langsam, okay, zu Produkt. Früher war das Unternehmen immer so orientiert, Produkt und dann gucke ich mal, jetzt kann ich hier anwenden, hier gehe ich auf den Markt, wo sind die Zwiebeln auch so organisiert und du warst dann mit mir nicht dabei, wenn wir gesagt haben, Prozessmanagement ist an der einen Seite, wir müssen erstmal die ganzen Unternehmen überzeugen, diese Strategie zu befolgen, weil sonst, dass du jetzt erwartest, dass einer sich meldet auf diese Imeradresse und sagt, die Idee, vielleicht werden die Paidee kommen, aber die beste Idee, die werden normalerweise kommen, einer, der mit mir klarkommt, wird mich anrufen und erzählen oder anderer, ruft dem jetzt Cashman an, macht den Tatzel, der ruft dich an, ich weiß nicht, es wird super in die Idee kommen, nicht wirklich über diese Imeradresse, wie wir das organisieren, ich finde, das ist wirklich es mir eigentlich egal, wenn das in Vorstellungen geht. Mir geht es nur darum, dass die Leute erstmal strategische Zeit verstehen und dass wir dann das Prozessmanagement am Ende brauchen und wir sind noch nicht hier bei strategischer Zeit noch nicht fertig, weil, du sagst, du hast das Prozessmanagement erklärt, aber Prozessmanagement liest keiner. Du kannst jetzt einfach komplizieren, nicht sagen, bitte liest mal Prozessmanagement. ich erinnere mich,
SPEAKER_UNASSIGNED: ich erinnere mich,
SPEAKER_04: dass der Prozessmanagement von Marketing damals geschrieben hat, hat das jemand gelesen? Nein, man meldet sich bei Marketing, was muss ich jetzt machen, wer muss ich jetzt mitmelden? Weil keiner hat Prozessmanagement für die Verifizierung und Umsetzung für die Unternehmen. Aber die Strategie liest, und das mussten wir immer klar definieren. Und deswegen ist diese Präsentation, wo das, was Geschäftsführung, weil die Strategiesitzung mitgeteilt hat, wo wir mitdiskutiert haben, jetzt geht es darum, wir mussten das im Unternehmen erst mal präsentieren, aber wir mussten erst mal einschauen, ist das jetzt richtig? Sind wir einmal, seit wir alle einverstanden, ist das jetzt richtig? Oder wollen wir das anpassen?
SPEAKER_03: Das verstehe ich.
SPEAKER_04: Deswegen haben diese Präsentation in die Gruppe gecheckt, gegeben, aber also an dem Tag, glaube ich, du warst nicht dabei, du warst auch nicht dabei, vielleicht nehmt ihr bitte Zeit und gebt die Kommentare. Oder wenn ihr keine Kommentare habt, dann sind wir alle einverstanden.
SPEAKER_03: Ja, wir müssen das dann. Das ist eigentlich, was ich ja schon sagen soll. Nein, nein, nein, nein, nein.
SPEAKER_04: Das ist traditionell bei diesem Vernehmen in die Vergangenheit geführt von Produktmanagementseite. Und das ist prinzipiell nicht falsch. Aber wofür habe ich dann diese Gewährung als Abteilung, wenn das nicht notwendig ist? Wir schauen anders, in Höhe zu, auf den Markt, das gefragt wird, und dann gucke ich mal, kann ich mit meinen Produkten und um dies zu leisten, was das Unternehmen anbieten kann, ein Business-Paket stocken, das ich anbieten kann.
SPEAKER_03: Wahrscheinlich, der Ansatz von der Ideentrennung und der Entwicklung ist zwar auf einer anderen Ebene aber der gleiche Gedankengang. Also sollte ja auch nutzenorientiert und nicht einfach, wir machen jetzt ein Produkt nix absoluter und dann gucken wir mal, wie wir uns verkauft kriegen. Also grundsätzlich sollte vom PM der Ansatz ja auch anders muss sein. Nutzenorientiert, kundenorientiert, etc. Aber ich hatte nicht das Gefühl, dass das jetzt die Aufgabenstellung ist. Ich hatte das Gefühl, die Aufgabenstellung ist, das Wissen zu zentrieren und auch zusammen an verschiedenen Punkten an den gleichen Dingen zu arbeiten. Das einmal als Aufgabenstellung und Projekten einen klaren Projektbeantwortlichen zuzuordnen. Das war vom Schnittstellen-Meeting, so wie es am Anfang hieß, wo Herr Peter hier mit uns saß, die Anforderungen, die an mich oder an uns, ich sage jetzt mich, weil ich es dann auf der Basis ausgearbeitet habe, das war der Gedankengang dahinter. Deswegen haben die Punkte, wie du sagst, hundertprozentig die Ansatzberechtigung und müssen auf jeden Fall auch aufschreiben, wie wir Projekte abarbeiten und so weiter und so fort. Aber das war überhaupt nicht der Gedankengang beim Ausführen in den Prozessen, sondern das war einfach nur, wie und wo sammeln wir die ganze Information, welche Kriterien gehören dahinter, ist es überhaupt Geotechnik und so weiter und so fort und wie gehen wir dann weiter. Das war alles, was da drinnen steckt und das nach diesen Kriterien zu bewerten, dem ist natürlich nicht gerecht. Aber das ist für mich halt auch einfach an einer anderen Stelle. Das ist für mich an der Projekt, also der, der dann den Hut dafür aufhat, wie er das abarbeitet, dass nichts von diesen Aussagen bildet das ab. Aber trotzdem hast du natürlich zu hundertprozent recht mit, dass auch daran gearbeitet werden muss. Aber das habe ich überhaupt nicht umsichtig.
SPEAKER_04: Ich habe das jetzt ein bisschen anders verstanden. Ich habe das richtig verstanden, dass diese Cremie das verdienen sollte, denn wenn du zum Beispiel Geschäftsentwicklung kommt und sagt, wir wollen auf den Markt, so keine Ahnung,
SPEAKER_03: ich weiß nicht, Max,
SPEAKER_04: achten kann.
SPEAKER_03: und dann kommt das jetzt erst mal hier auf den Tisch in irgendeiner Form. so, ob du das jetzt hier reinbringst, ob das per Mail kommt, das ist ja wurscht. Das kommt hier auf den Tisch mit einer gewissen Anzahl an Informationen und dann sehen wir, das ist ein neuer Markt, das ist ein Business Development Projekt. Wer hat ein Business Development den Hut auf? So, dann, was weiß ich, machst du das, der ist eine Person, dann wissen wir hier, dieses Projekt wird gerade bearbeitet, steht in der Liste, du hast den Hut auf und irgendwann wollen wir wieder davon und wenn wir zufällig im Produktmanagement auch das Thema Afrika aus irgendeinem Grund auf den Tisch kommen und jemand zu mir kommt, weiß ich, wir waren aber sogar schon, ich spreche bitte mit dem anderen. Das war für mich die Intention und das soll es abbilden. Aber du hast ja die volle Projektverantwortung dann in dem Sinne, das ist bei euch, das gehört zu euren Kriterien und so weiter, ihr sollt nur laut dem Prozess und das sind die Sachen, die wir diskutieren wollten, irgendwann mal eine Schleife drehen und das hier vielleicht mal wieder vorstellen, was der aktuelle Stand ist und so weiter und so fort. Aber das ist für mich alles, was da abgebildet werden sollte und das ist das, weswegen ich mich auch frage, warum brauchen wir, Entschuldigung, aber sechs, sieben, acht Wochen, das fühlt sich für mich gerade langsam an, einfach.
SPEAKER_04: Ja.
SPEAKER_03: Ja. Ja, aber... Na, da...
SPEAKER_04: Ja, aber die Frage ist jetzt, auch wenn das entschieden wird, okay, wer ist jetzt der Fraordinator? Eigentlich geht es mehr darum, dass ihr Resorten festgelegt werden, weil, wenn ich zum Beispiel sage, ich arbeite jetzt in Afrika, keine Ahnung, in Ghana, alte Komis-Stiftung, dann muss ich, versetze ich es an, auch vom Marketing, also Unterstützung bekommen, ich muss die Unterstützung von WM bekommen, keine Ahnung.
SPEAKER_03: und dann, genau, genau, deswegen sitzen wir hier dann zusammen, kann darüber reden, aber, ja, so hatte ich, das ist einfach mein Verständnis gewesen, letztendlich. So, und ich habe das ja auch extra so gemacht, wir hatten das so ausdiskutiert, so hatte ich es verstanden, deswegen hatte ich es so vorbereitet, in diesem, ich habe einfach den Spendenprozess genutzt, weil es ja eh aufgeschrieben werden muss, und ich habe es ja auch rumgeschickt, bitte in zwei Wochen, bitte Kommentare zurück, und so weiter und so fort, habe keine Kommentare gekriegt und dementsprechend hatte ich angefangen mit der Umsetzung, weil, wie du gerade eben sagtest, kein Kommentar ist Zustimmung, deswegen hatte ich weitergemacht. Und, ja, also ich sehe gerade nicht, dass wir wirklich weiterkommen, das sind die gleichen Punkte, über die wir gerade diskutiert haben, wie vor drei Wochen.
SPEAKER_02: Ich sehe, ehrlich gesagt, das ganze Problem nicht, außer dass man sich jetzt daran aufhängen kann, dass wir hier vermeintlich nur über Prozessmanagement reden, nee, tun wir gar nicht, wir reden einfach nur darüber, wie der Prozess aussehen soll und den bilden wir dann ab, Punkt. So, und wenn das so ist und wenn wir uns da vom Grunde her erstmal einig sind, dann heißt das eigentlich nichts anderes als, da bin ich dann bei Björn, da bin ich auch bei dir, wir haben am Ende des Tages irgendwo, ich sage mal, vier, fünf Schritte zu beschreiben, das können wir in der Präsentation machen, das können wir in der Word-Datei machen, sinnvollerweise machen wir das im Prozessmanagement, okay, bei dir habe ich jetzt rausgehört, Prozessmanagement interessiert sowieso keinen, guckt ja eh keiner rein, sehe ich ein bisschen anders und selbst wenn es so wäre, wäre es trotzdem falsch, das dann eben auch so zu gutieren, sondern da müsste man im zweiten Fall dagegen arbeiten, aber wir sollten es einfach vernünftig beschreiben, Punkt. und ob ich das dann hinterher präsentiere oder ob ich den Trichter präsentiere oder sonstigen, das ist auch noch wieder eine andere Baustelle. Auf jeden Fall habe ich dann, wenn ich das im Prozess einmal beschrieben habe, immer die Möglichkeit, die Kollegen da auch hinzuweisen. Es wäre auch sowieso geschickt, ich sage mal, die Dinge aus dem Prozessmanagement dann für Präsentationen zu nutzen, das sorgt dann vielleicht dafür, dass die Leute eher da reinschauen, als dass sie sich irgendwelche Präsentationen raussuchen, die sie irgendwann mal gekriegt haben. Also es würde Sinn machen, spielt aber alles keine Rolle. Der Punkt ist, ich habe es so verstanden und so habe ich es von Lars verstanden, als wir über diese Runde gesprochen haben, der will mich nicht dabei haben, weil er es im Prozessmanagement geschrieben haben will, sondern der will mich dabei haben, weil ich muss mich um Prüfungen kümmern, ich weiß, welche Regelwerke in den einzelnen Ländern zu tun sind, ich weiß, was ich bei Zulassungen machen muss und all das sind Themen, die wir zwangsweise bei neuen Produkten, neuen Märkten immer mit auf den Tisch stellen. Deswegen wollte er mich dabei haben. Das Prozessmanagement habe ich einfach nur gesagt, es muss vernünftig geschrieben werden. Punkt. Für mich ist es relativ einfach. Es gibt ein Eingangskästchen, das heißt, GTM Hub oder wie auch immer, können wir auch einen anderen Namen nehmen, ist mir völlig egal. Da steht drin, was ist das hier für eine Runde? Was tun wir hier? Was ist unsere Ausgabe? Und dann gibt es ein Kästchen, möglicherweise ein zweites, da ist dann Ideenpool, das können wir irgendwo anhängen, scheißegal, wir können das in der Runde anhängen, wir können das als nächstes Kästchen machen, wir fragen irgendwelche Dinge nach, dann sind wir bei einer Entscheidungsfindung, das ist das, was Björn gesagt hat, was gehört zur Entscheidungsfindung dazu, wir fragen uns strategisch das, wir fragen das, das, das, das sind die Fragen und auf Grundlage der Antworten, die es zu den Fragen gibt, gibt es eine Entscheidung, wem geben wir die Aufgabe.
SPEAKER_03: Genau.
SPEAKER_02: Ich glaube nicht, dass wir an der Stelle schon sagen können, jawohl, fünf Jahre, fünf Millionen, 30 Prozent Lettungsbeitrag, das kommt möglicherweise ein halbes Jahr später, nachdem irgendjemand angefangen hat, eine Basisarbeit zu machen. Unabhängig davon, verteilen wir aber trotzdem eine Aufgabe und sagen, ey, irgendeiner muss sich ja erstmal damit beschäftigen, weil sonst machen wir jede Aufgabe im Ansatz tot, weil das kann realistisch keiner. Wenn ich irgendwo eine Idee habe, ich sage mal ein Beispiel, was Giovanna gerade sagte, keine Ahnung, ob es realistisch ist, in fünf Jahren mit gezielten Aktionen in Ghana fünf Millionen Umsatz mit 30 Prozent Deckungsbeitrag zu machen. Dazu muss ich erstmal ein bisschen Basisarbeit machen. Aber das ist ja das Beispiel. Danach geht es weiter. Und wer die Fachabteilung ist, die sich dann danach kümmert oder den Hut auf hat, das ergibt sich aus den Themen. Klar, wenn es neue Märkte sind, ist es Business Development. Wenn es irgendwelche ganz anderen Themen sind, die bisher mit klassischen Geocoolstocken gar nichts zu tun haben. Ist es möglicherweise auch Business Development oder ist es der PM oder, oder, oder. Das wird man themenabhängig definieren müssen.
SPEAKER_04: Okay, aber wenn du dich erinnerst, was ich hier präsentiert habe, dann haben wir damals sieben Stufen einmal Stages vorgestellt. Dann haben wir gesagt, ich habe das jetzt noch jetzt vom Telefon, kann ich noch mal lesen. Erstmal war diese Opportunity Identification, das wäre zum Beispiel
SPEAKER_01: die Liste.
SPEAKER_04: gelistet, dann kommt die Nischung Screening und die Nischung Screening heißt, ich muss tatsächlich die ersten Daten zusammenfassen, um einfach zu entscheiden. Wir machen weiter, wir pausieren oder wir sagen, nee, geht nicht weiter.
SPEAKER_02: Aber wer ist ich an der Stelle?
SPEAKER_04: Ja, genau. Das ist ja die Frage. Weil in dem Moment muss ich dann entscheiden, in welcher Abteilung geht die Idee.
SPEAKER_02: Und das ist das, was Björn vorhin sagte. Passt das in unsere Strategie? Möglicherweise kann ich das an dem Zeitpunkt noch gar nicht entscheiden, wenn einer die Idee abgegeben hat, sondern erst nach dem zweiten, dritten Step, weil ich erst mal ein bisschen Basic-Arbeit machen muss. Aber all das können wir relativ einfach in Stichpunkten darin beschreiben. Dann weiß jeder, was ist die Aufgabe. Und nochmal, wir können das in FOD reinhängen, wir können das in Produktmanagement reinhängen, wir können das in den Business Development Part reinhängen, wir können das an vielen Stellen können wir das reinhängen. Dann hat das jeder immer vor Augen, kann mit einem Klick da drauf das aufmachen und sieht sofort, ah, was passiert da und wer macht es.
SPEAKER_03: Und das war's. Und wir werden den Ablauf des Projektes auch nicht im Prozessmanagement definieren können. Dafür ist das viel zu unterschiedlich, hat viel zu unterschiedliche Anforderungen, je nachdem, gar nicht.
SPEAKER_02: Nein, aber wir können definieren, wir können definieren bis zu dem Zeitpunkt, wo wir sagen, wir verteilen die Aufgabe und möglicherweise können wir auch definieren, wir haben Prüfpunkte, wo wir ein Feedback in dieser Runde möglicherweise erwarten, das muss man definieren und sagen, jetzt kriegen wir mal was zurückgespielt und jetzt entscheiden wir, geht das Projekt weiter oder geht es nicht weiter oder wir sagen auch, wir sind uns unsicher, müssen wir vielleicht in Geschäftsführung anfangen, wie sie es sieht. Und auch das ist da drinnen schon zu tun. Aber auch um das mal ganz klar zu sagen, ich sitze hier nicht als einer, der sagt, wir müssen den Prozess so oder so oder so leben, beschreiben oder so, beschreiben schon, leben. Das ist nicht meine Aufgabe als Prozessverantwortlicher oder als der, der für das Prozessmanagement verantwortlich ist. Die Aufgabe als Prozessmanagement verantwortlicher ist, zu beschreiben, wie wir arbeiten wollen. Punkt.
SPEAKER_04: Aber wir haben uns noch nicht abgestimmt,
SPEAKER_02: wie wir arbeiten wollen.
SPEAKER_04: Deswegen komme ich einen Schritt vor.
SPEAKER_02: Ja, aber möglicherweise ist auch das einfacher, wenn man mal ein Gruppgerüst stehen hat und sagt, okay, das ist mal der Große und jetzt machen wir hier noch was.
SPEAKER_04: Aber das habe ich präsentiert. Ich habe das präsentiert bei der letzten Termin, wo wir alle zusammen gesammelt haben. Normaler war nicht dabei und der falscher war natürlich dabei.
SPEAKER_02: Das habe ich einmal präsentiert. Habe ich jetzt ehrlich nicht mehr vor Augen, wenn du diesen Funnel präsentiert hast, dann ist das sozusagen wie das Projekt läuft?
SPEAKER_04: Nein, nein, nicht das wie das Projekt. Von der Idee bis die Umsetzung. Welche Schritte müssen wir machen? Weil auch wenn die Idee gut ist, einer muss sich kümmern, wie komme ich mit Ideen auf den Markt? Das habe ich überhaupt nicht definiert. Wie komme ich von der tolle Idee, bis der Betrieb etwas verkaufen kann? Wir haben diesen strategischen Teil und der Prozess definiert noch nicht.
SPEAKER_02: Mag sein.
SPEAKER_04: Nein, haben wir nicht.
SPEAKER_02: Ja. Ich kann es momentan nicht sagen, weil ich habe es jetzt leider nicht vor Augen. Aber die Kunde ist wiederholen.
SPEAKER_04: Ich kann doch mal...
SPEAKER_02: Andersrum, andersrum. Also, vielleicht vom Grunde her sind wir uns einig, was wir wollen, glaube ich. Also, vom Grunde her sind wir uns einig, dass wir eine Runde haben wollen, wo wir sagen, da kommen die Ideen an. Wir reden über die Ideen, sagen, jawohl, an welcher Stufe sind wir womöglich, müssen wir noch Aufgaben verteilen, brauchen wir noch viel Deck oder können wir es komplett verlagern, was auch immer. Und wir entscheiden, jawohl, da und dann geht es weiter. Das ist ja erst mal unsere Kernaufgabe. Da habe ich es zumindest verstanden. Ja, unsere Kernaufgabe ist es möglicherweise, das müssten wir dann definieren, zu sagen, jawohl, diese einzelnen Stufen, die wollen wir sehen oder wir sagen, nee, das liegt in der Verantwortung des Projektverantwortlichen, das können wir definieren. Also ich sage mal, ich habe es bisher nicht so verstanden, dass wir nicht eine Runde sind, wo jetzt einzelne Projektverantwortlichen quasi regelmäßig reporten sollen, wo sie stehen. Nein, das gibt es nicht. Also die Frage ist, wer kümmert sich darum, dass das Projekt dann diese einzelnen Schritte auch beinern wird. Auch das kann man definieren. Wir können im Prozessmanagement, könnten wir so sagen, wenn wir das so wollen, jawohl, das ist ein Projekt, das ist YZ, der kümmert sich darum, hier hast du einen Fahrplan, das sind die einzelnen Steps, die musst du abarbeiten. Irgendwann wollen wir das mal sehen. Vielleicht wollen wir das erst sehen ganz am Ende, aber wir wollen sehen, dass diese einzelnen Schritte eingehalten wurden.
SPEAKER_03: Genau, wir haben halt einfach nur einen einzelnen Loop eingebaut zu den Zeiten, dass einfach, falls es Punkte gibt, die erst, wie du gesagt hast, detaillierter ausgearbeitet werden müssen, beispielsweise, dass es dann halt wieder zurückgeht. Da hat man einen Loop drin, beispielsweise. Aber das ist halt für mich jetzt auch schon Detail, zumindest bei dem Part, weil das ist für mich einfach zwei unterschiedliche Punkte sind. Das eine ist der Ablauf, wie führen wir das jetzt zusammen und durch und das andere ist für mich dann nochmal der detaillierte Projektfahrplan. Ja.
SPEAKER_04: Also ich würde euch alle bitten, also nochmal die Präsentation zu lesen. Ich kann die abgelehnt und die Präsentation schicken und dann, dass wir uns nochmal treffen, das entwürdig zu entscheiden, wer das genau ist.
SPEAKER_02: Die Frage ist, wollen wir in Zukunft jedes Thema, ich sag mal so, hoch hängen? Fragezeichen? Spaß einfach mal. Weil das ist natürlich ein extrem hoher Anspruch oder ist das dann etwa, also ich denke jetzt auch mal an das Thema fünf Millionen Umsatz, wann kann ich das haben, wie viel bringt, ne? Also ich,
SPEAKER_03: wenn man sich überlegen, es wird ja auch Ideen geben,
SPEAKER_UNASSIGNED: wie,
SPEAKER_03: und da gehe ich jetzt mal gerade gedanklich auf das Produkt, drei Prozent Herstellkosten bei Venture Fix durch XY und das hat ja diesen Anspruch dann gar nicht an Detail gemacht und deswegen habe ich es nicht in diesem Detail gerade, in diesen Projekten bei dem Projekt mit aufgeführt, weil man es einfach nicht verallgemeinern kann. Ja, das ist ein gutes Beispiel. Genau. Natürlich bin ich zu 2000 Prozent auf deiner Seite, dass wir anfangen müssen, diesen Detailgrad zu überführen in der Art Projekt mit Dünn kommt fast. Gerade Business Development und auch vermehrt in PR-Anwendungen. Da muss der Gegenstände auch noch...
SPEAKER_04: Und die Gegend sind. Weil es kann sein, dass der Gewerleiter sagt, okay, es ist ein toller Markt, tolle Anwendung und dann, eigentlich die nächste Gegend zum Produktmanager, dann ist das technisch unnötig möglich dann.
SPEAKER_03: Abbruch. Genau, klare Abbruch. Dann geht es nicht da.
SPEAKER_04: Und jetzt ist es so, jeder macht, was er will, oder sie will. Und es ist nicht koordiniert. Und es kann nur koordiniert, wenn alles 100 Prozent funktioniert. Deswegen, ich bitte euch, also nochmal, Präsentation zu lesen. Und ich habe auch verstanden, was jetzt Henning gesagt hat und das wird am Ende von uns dann entscheiden, wie kommen wir in der Freizeit. Weil jetzt diskutieren wir, finde ich, wie jeder das hat verstanden hat. Jeder hat es ein bisschen anders verstanden, das ist auch normal. Aber, letztendlich, wie gesagt, für mich ist klar, wir haben uns vor die Schöpfige-Sitzung getroffen, da war die Schöpfige-Sitzung, da war klar, was zu tun ist. Und jetzt sind wir zwei Momente danach und wir haben uns gesagt, dass wir haben uns nicht bewegt.
SPEAKER_03: Können wir uns, das ist jetzt so ein Vorschlag von meiner Seite, vielleicht darauf einigen, dass wir das gedanklich trennen, dass wir diesen Ablauf, wie wir uns hier zusammensetzen mit den Prozessen und den ersten Kriterien und davon den Gedanken trennen, wie oder welche Kriterien sind für welche Art von Projekten wichtig. Können wir es so nennen, dass wir das von der anderen trennen, dass wir das eine vielleicht endlich zum Abschluss bringen können und dann aber sagen,
SPEAKER_04: in der Präsentation gibt es überhaupt nicht Kriterien. Das sind keine Kriterien genannt.
SPEAKER_03: Wie die Gates, die du gerade angesprochen hast, das ist doch eine Art Kriterien.
SPEAKER_04: Nein, es geht um die Schritte. Vom Anfang, vom Ideen, bis zum Ende, bis zur Umsetzung. Welche Schritte muss ich durchführen?
SPEAKER_03: Ja, das habe ich mir angeguckt, das reicht nicht großartig, also das hat an manchen Punkten einen höheren Detailgrad, aber der Ablauf ist ja nicht großartig anders als das, was wir jetzt auch schon abgebildet haben. Das ist vom Grundgedanken, sehr ähnlich meiner Meinung nach.
SPEAKER_04: Ich glaube nicht, weil ich habe mich mit der Abwägung beschäftigt.
SPEAKER_03: Okay. Ich habe das in Details
SPEAKER_04: viermal gelesen.
SPEAKER_03: Ich habe mir das auch angeguckt, ich habe es so wahrgenommen, ja, dass es vom Grunde her bringt gleich Gedanken.
SPEAKER_02: Also eigentlich ist das relativ einfach und ich sehe auch nicht, also ich bin nicht der Meinung, dass wir uns nicht bewegt haben, im Gegenteil, weil so doof diese Diskussionen manchmal sind, sie sorgen ja dafür, dass man im Gegensatz ein besseres Verständnis hat und das ist ja genau der Fakt, noch vor zwei Monaten haben wir anscheinend alle einander vorbeigeredet und das kommt jetzt mehr und mehr auf den Tisch und jetzt sehen wir langsam, wo die unterschiedlichen Interpretationen dessen sind, was auf den Tisch lag und jetzt müssen wir versuchen, das in eine Richtung zu bringen, damit wir dann wirklich am Ende des Tages auch, ich sage mal, in dieser Runde, aber noch viel näher da draußen, das gleiche kommunizieren denn es bringt ja gar nichts, wenn Giovanna Christberg irgendwas erzählt und zwei Tage später redet er mit dir und du erzählst das Gegenteil. Also das wäre einmal das Allerschlechtste. Dann müssen wir nur die Anforderungen und wenn das dann mal drei Monate dauert, dann dauert es drei Monate. Aber wenn wir dann anschließend mit einer Meinung rausgehen können und die auch gemeinsam mit dem gleichen Wording vertreten nach draußen, dann haben wir richtig was erreicht.
SPEAKER_03: Ja, dann sollten wir meiner Meinung nach wenigstens die Anforderungen nochmal um ein oder zwei Sätze ergänzen, weil für mich kommt das aus dieser ersten Anforderung mit Liste für alle und Projektverantwortlichen ist der Part für mich gedacht, ich habe einfach nie Teil der Institution gewesen. Das ist halt einfach der Punkt. Deswegen, wenn ich ja trotzdem zu 100% bei Giovanna auf der Seite, dass das gemacht werden muss, wirklich wichtig ist, dass wir da auch noch
SPEAKER_02: Nachholbedarf haben. Also für mich wäre es gefühlt gerade relativ einfach, sowohl das, was Giovanna will, abzubilden, als auch das, was möglicherweise an der Stelle too much ist, weil ich glaube, es gibt eine ganze Menge Projekte, wo dieser Anspruchsgedanke möglicherweise zu viel ist oder ich würde damit jedes Thema gleich im Ansatz killen. Auch das kann man glaube ich abbilden, indem wir einfach sagen, in der Runde wird entschieden, wie das A, also A, wer verfolgt wird weiter und B, auf welchem Niveau. Auch das kann man definieren. Und ich sage mal, gerade wenn es um neue Merkel geht, ist es vielleicht relativ gut zu definieren, wenn es um neue Produkte geht auch, wenn es um andere Themen geht, ist es vielleicht nicht mehr so fast zu definieren. Nehmen wir das Beispiel Zulassung, wir diskutieren gerade auch eine Zulassung in einem Land machen, da kann Stand heute keiner realistisch sagen, dass wir damit in 5 Jahren 5 Millionen Umsatz bei 30% Deckungsbeitrag machen. Vielleicht machen wir 5 Millionen Umsatz, aber sicherlich nicht bei 30% Deckungsbeitrag. Lassen wir dann die Zulassung gleich sein? Also ich schicke euch
SPEAKER_04: jetzt die Präsentation weiter und ja, dann
SPEAKER_03: bist du denn aber mit dem restlichen Ablauf einverstanden, wenn wir nochmal darüber gesprochen haben.
SPEAKER_04: auch mit Prozessmanagement, ich habe ja nichts zu sagen, du hast halt so, das soll ich das sagen. Also ich bin, wenn, ist es okay für mich. Auch wenn die Projekte bleiben mit Forschung und Entwicklung, also das ist auch okay. Ich habe nichts dagegen. Für mich ist es wichtig, dass die Wissen-Gewerberkunft integriert ist, weil Wissen-Gewerberkunft irgendwie wo ist und ich habe es okay. Alexandras, was soll ich sagen? Ich stecke mir gar nicht im Detail, dass wir das abgebildet sein muss.
SPEAKER_02: Und es gibt auch kein Muss, es gibt, ja, es gibt schon ein paar Grundanforderungen, aber ich sage mal, wie wir Prozessmanagement abbilden, wie wir Mission 2030 abbilden, wir haben da letztes Jahr mehrfach darüber diskutiert, bis wir zu dem Gedanken gekommen sind, wie wir es jetzt tendenziell machen wollen. Ich kann es auch gerne mal zeigen, wenn ihr es sehen wollt, aber wir sind da wirklich noch am Anfang. Das ist der Gedanke, das ist der Gedanke.
SPEAKER_00: Das ist der Gedanke.
SPEAKER_03: Das ist der Gedanke. Das ist der Gedanke. Das ist der Gedanke. Das ist der Gedanke.
SPEAKER_UNASSIGNED: Das ist der Gedanke. Das ist der Gedanke. Das ist der Gedanke. Das ist der Gedanke.
SPEAKER_02: Danke.
SPEAKER_UNASSIGNED: Danke.
SPEAKER_02: Das ist der Gedanke.
SPEAKER_UNASSIGNED: Das ist der Gedanke. das ist der Gedanke.
SPEAKER_02: Das ist der Gedanke. Das untere ist der Gedanke.
SPEAKER_UNASSIGNED: Das ist der Gedanke. Das ist der Gedanke.
SPEAKER_02: Das untere ist das, wie wir es heute in der Präsentationen im Wesentlichen immer haben, wo man sieht, wer hat mehr oder weniger den Lied und wer sind die beteiligten Abteilungen. Und wir haben das jetzt gedankelich mal so geteilt, dass wir gesagt haben, es gibt eine Phase, bevor der Tender raus ist und es gibt eine Phase, nachdem der Tender raus ist. Und wir sind jetzt angefangen zu gucken, wie wir das im Detail beschreiben können, indem wir gesagt haben, okay, da ist jetzt ein Tender raus. Also das ist hinter den anderen steckt noch nichts dahinter. Das ist ein bestehender Prozess, so wie wir ihn heute haben, nämlich der Prozess, der Auftrag ist da, der Vertrieb hat die Auftrags-Order-Checkliste ausgestellt und jetzt geht es weiter mit Versand und so weiter. An dem Prozess ändert sich erst mal nichts. Insofern haben wir den hier einfach stumpf reinkopiert. So, und dann haben wir angefangen und haben uns den Tender-Prozess angeguckt und das ist jetzt nur ein Entwurf. Den haben wir noch nicht diskutiert. Wir haben das vom Grunde her diskutiert, aber noch nicht vollständig ausdiskutiert und da wird es auch sicherlich noch Änderungen geben, wo wir einfach erst mal gesagt haben, wer ist denn überhaupt in dieser Phase jetzt noch in irgendeiner Form beteiligt. Hauptsachen laufen in Commercial Sales, dann gibt es Technical Pre-Sales, dann gibt es EDD, den PM, den Einkauf, Supply Chain Management, Logistik, Qualitätssicherung und FA war nicht Financial. Ich muss überlegen, was ist der Begriff FA noch mal. Dahinten müsste es dann irgendwann stehen. Finanzbuchhaltung. Also wenn es darum geht, ist jemand kreditwürdig und so weiter. Das heißt, hier detaillieren wir jetzt, beschreiben wir, wie läuft es eigentlich, hier für Kunde schickt ein Tender, was passiert jetzt alles bis zu dem Zeitpunkt, wo wir den Aufrag haben. Und wie gesagt, der basiert jetzt im Wesentlichen an dem alten Prozess, aber eben nicht komplett. Und in einer ähnlichen Form müssen wir das für andere Prozesse auch noch beschreiben. Und hier ist dann zum Beispiel auch ein Unterprozess drin, Kundenzufriedenheit, hier ist ein Unterprozess von Rücknahme von Waren, den wir woanders beschrieben haben, den wir hier jetzt auch noch mal beschreiben. Und genauso können wir es mit dem GTM Hub machen. Wir können hier zum Beispiel einfach reinkopieren, den GTM hat. Und wenn irgendeiner in Commercial Sales eine Idee hat, dann weiß er, ach, so läuft es. An wen wende ich mich jetzt? Ich habe gestern noch mit Malte gesprochen, ich schicke es mal an Malte. Oder ich schicke es an Giovanna oder ich schicke es an Björn oder wen auch immer. Völlig egal, weil von da weiß ich, alles klar, so läuft es weiter. Das ist das, so wie es im Prozessmanagement beschrieben ist. Wir haben hier zum Beispiel Antrag für Sonderartikel, den Prozess gibt es schon. Den habe ich hier einfach reinkopiert, da kannst du hinterher im Prozessmanagement, machst du einen Doppelklick drauf, macht er den auf, Sonderartikel. Und genau so werden wir versuchen, perspektivisch die anderen Prozesse zu beschreiben. Gehe ich jetzt hier nochmal wieder zurück. Business Development, keine Ahnung, da werden wir uns zusammensetzen müssen mit den Leuten, die es betrifft und sagen müssen, wie läuft es. Und zwar müssen wir es so beschreiben, bis wir auf der einen Seite was nachvollziehbar haben und auf der anderen Seite uns aber keine unnötigen Fesseln anlegen. weil es läuft natürlich nicht jedes Mal genau so. Und genauso ist es bei Design Phase until Tender Preparation. Das ist ein neuer Prozess, den haben wir so in der Form. Den werden wir beschreiben, da werden wir mit dem Vertrieb zusammensetzen müssen und sagen müssen, wie soll es denn jetzt eigentlich laufen.
SPEAKER_04: Du hast ja eine Frage, also manchmal ist es tatsächlich so, du hast keinen Geotechnical Engineer, sondern hast nur Specification Engineer. Das ist, was ich meine, du hast da auch keine, das etwas mit Geotechnik zu tun hat. Und der Specification Engineer ist dafür für alles verantwortlich. Achso, ich rede jetzt von diesem Design.
SPEAKER_02: Du meinst das hier.
SPEAKER_04: Ja, genau.
SPEAKER_02: Ja, das ist, also... Das ist, was ich meine,
SPEAKER_04: manchmal fehlt ein Schritt. Oder entfällt ein Schritt.
SPEAKER_02: Ja, aber das ist, das ist an der... Also, ich sag mal, das müssen wir einfach versuchen zu beschreiben. Wir müssen uns, wenn wir sagen, wir haben keinen Geotechnical Engineer, dann müssen wir es halt so beschreiben. Also wir können ja auch hier nicht, wir können ja auch hier schreiben, das ist ein Ingenieurbüro. Neutral. Das hier ist ja, das kommt aus den bisherigen Präsentationen. Das müssen wir jetzt nicht alles eins zu eins übernehmen. Aber darum geht es. Warum das? Weil wir haben nun mal einfach, leider die Situation, dass wir nächstes Jahr wieder ein Rezentifizierungsaudit haben. Und die wissen, die wissen, wir wollen SAP einführen. Die gehen davon, also Stand heute gehen die davon aus, SAP läuft dann schon. Okay, werden sie überrascht sein. Vielleicht auch nicht, dass es dann doch noch nicht läuft. Die haben schon mal was von Salesforce gehört. Das kennen die. Die wissen auch von Alex aus dem letzten Audit, dass wir Salesforce eingeführt haben. Das heißt, die werden solche Sachen hinterfragen. Die haben auch schon bis zum 2030. Also wir müssen es einfach beschreiben. Und hier geht es mir, am Ende geht es mir darum, ich möchte, dass wir die Prozesse nach Möglichkeit so beschreiben, dass wir zu 90% sagen können, jawohl, so ist es, wie wir es haben wollen. Dass es immer mal Ausnahmefälle gibt, wo es anders läuft, aus guten Gründen. Das ist völlig legitim. Nur sollte es nicht genau andersrum sein, dass 10% so laufen, wie es beschrieben ist und 90% laufen anders. Weil dann ist unsere Beschreibung falsch. Wir wollen das beschreiben, wie wir wirklich arbeiten. Und ich muss den Quercheck machen, ob das die Mindestanforderung der ISO 9000 erfüllt. Aber wenn das der Fall ist, haben wir alle Freiheit. Und wenn du sagst, ich, ich, Giovanna, will, dass der Business-Ziel Welt auf meinem Prozess so läuft und genau so und nicht anders, dann beschreiben wir das so und dann musst du halt deinen Leuten das einfach nur entsprechend vermitteln, dass du es genau so haben willst. Sofern es denn die Basics passiert. So läuft es bei allen anderen Prozessen auch. Das ist an der Stelle jetzt auch wirklich Standard.
SPEAKER_04: Deswegen habe ich gesagt, ich kann nichts dafür sagen, weil ich war hier nicht involviert.
SPEAKER_02: Nein, nicht.
SPEAKER_04: Das ist für mich somit okay.
SPEAKER_02: Nein, du bist hier, momentan bist du hier ganz bewusst noch nicht involviert.
SPEAKER_04: Nein.
SPEAKER_02: Weil wir das Thema Business Development ja noch gar nicht bearbeiten. Also es macht auch wenig Sinn, ich sage mal, als Beispiel, wenn wir jetzt Tender to Order besprechen, dass wir da fünf Leute von Pre-Sales dabei haben und so weiter und so viel.
SPEAKER_04: Das ist jetzt auch die Frage, weil das ist die Construction Tender. Aber du hast vor Construction Tender, du hast auch Tender für die, also Final Design, Tender für die Pre-Feasibility Studio, Tender für die Feasibility Studio.
SPEAKER_02: Ja, das ist aber dann mehr Business Development.
SPEAKER_04: Ja, ich wollte nur sagen, so viele Details geben wir hier nicht.
SPEAKER_03: Wissen wir vielleicht auch nicht. Macht man ja auch bewusst ganz häufig nicht, man versucht ja wirklich nur die, den Haupt, oder das häufigste.
SPEAKER_04: Ja, aber deswegen, ja, aber deswegen ist hier, würde ich vereinfachen, und nicht für Technologien Spezification Engineer, sondern Engineer schreiben. Aber manchmal hast du kein Technologien.
SPEAKER_02: Johanna, überhaupt kein Problem. Der untere Part, der ist für mich entscheidend. Den habe ich ganz bewusst dort reinkopiert, der ist auch nicht verlinkt, gar nichts, weil das ist das, was unsere Kollegen aus den verschiedenen Präsentationen kennen. Und ich will einfach nur sicherstellen, dass die sich wiederfinden.
SPEAKER_04: Ja.
SPEAKER_02: dass die wissen, ah, alles klar, ich bin da oben, weil das ist das, was ich kenne, ah, ich muss da reingucken, ich muss da reingucken, darum geht es. Irgendwann werden wir das untere Netz schmeißen, weil das hat kein, außer dem Wiedererkennungswert, hat es keinen Mehrwert. Da drin, ob wir einen Prozess mit drei Schwimmbahnen haben oder mit 20 Schwimmbahnen, ob der gigantisch ist oder klein ist, da haben wir keine Vorgaben. Wir müssen es so beschreiben, wie wir es für richtig halten. Und dabei sollte man immer vor Augen haben, im Idealfall ist das ja auch etwas, was den Kollegen durchaus mal hier und da eine Hilfestemmung sagen kann. Also schon einen vernünftigen Detaillierungsrat, ohne mir auch unnötige Fester anzunehmen.
SPEAKER_03: Ich habe jetzt einmal aufgeschrieben. Wir gucken nochmal über den Business Development Vorschlag. Ich würde auch nochmal mit reiten schreiben, dass das auch nochmal eine andere Zielsetzung ist. Also dass wir diese drei Punkte jetzt, also der Projektablauf, Ideen zusammenführen und das gut aufsetzen, dass das drei Anforderungen sind, damit wir uns da alle vom Grunde her einig sind, was wir damit erreichen wollen und dass wir den Part von Vibana, wie die Projekte abläufen, halt nochmal jetzt neu bewerten, sozusagen, und dass wir den ersten Teil erst mal, wie ist grob der Ablauf, dass wir den so lassen, dass wir uns da eben einig sind, dass wir aber nochmal das beim Projekt selbst, wie ist der Projektablauf, welche Kriterien möglich sind, dass wir da jetzt nochmal detailliert da reinschauen. Seid ihr damit einversprochen?
SPEAKER_04: Ja gut, aber Projekt, je nachdem, wie du das Projekt definierst, weil ein Projekt könnte, also, ja, das Projekt kann alle sein,
SPEAKER_03: und das ist ja die Problematik beim Dachstellen tatsächlich. Das ist ja auch das,
SPEAKER_02: was Henning meinte mit Wir werden nicht... Vielleicht können wir, vielleicht können wir dem, was du ausgearbeitet hast, einen prägnanten Namen geben und einfach sagen, es gibt bestimmte Themen, die arbeiten wir immer so ab und dann ist das genau das und es gibt andere Themen, die könnten, da mag es Sinn machen, die so abzuarbeiten oder es mag auch Gründe geben, warum es keinen Sinn macht, die so abzuarbeiten, dann werden sie eben anders... Also sprich, dass wir immer sagen, wenn wir das so abgearbeitet haben wollen, wie dann der auch immer der Name ist, dann müssen wir das eben auch klar definieren. Aber, dass wir im Umkehrschluss, weil, das Spielchen Ken Martin ja, von F&E, wenn dann der Auditor da sitzt und wir sagen, jedes Projekt wird so abgearbeitet, dann ist das das, was wir selber sagen, dann geht der Auditor da hin und sagt, können Sie mir mal drei Beispiele zeigen? Und wenn du dann zwei Beispiele hast, wie super sind und ein Beispiel, was scheiße ist, dann haben wir eine Abweichung.
SPEAKER_04: Wo wir sehen, heißt es jetzt nicht Projekt, sondern heißt Opportunity und Opportunity muss qualifiziert sein, Projekt zu werden. Aber vielleicht, wenn ich die Präsentation gelesen habe, dann, also einmal vielleicht lesen Okay, danke.
SPEAKER_01: Ganz ehrlich, ich habe das gleiche eben auch
@@ -75,7 +75,7 @@ mentioned_people:
aliases: [] aliases: []
role: null role: null
department: null department: null
attendance_status: "not_present" attendance_status: "mentioned_only"
notes: null notes: null
organization: organization:
@@ -127,4 +127,4 @@ context_rules:
do_not_infer_departments: true do_not_infer_departments: true
do_not_infer_responsibilities: true do_not_infer_responsibilities: true
do_not_infer_attendance: true do_not_infer_attendance: true
mentioned_people_are_not_participants: true mentioned_people_are_not_participants: true
@@ -64,7 +64,7 @@ mentioned_people:
- "Giovana" - "Giovana"
role: "Leiterin Business Development" role: "Leiterin Business Development"
department_id: "bd" department_id: "bd"
attendance_status: "not_present" attendance_status: "mentioned_only"
notes: null notes: null
organization: organization:
@@ -4,6 +4,8 @@
schema_version: "1" schema_version: "1"
meeting: meeting:
# Stable identifier used for provenance across corrections and later runs.
meeting_id: ""
# Human-readable title for the meeting. # Human-readable title for the meeting.
title: "" title: ""
# Dominant meeting language, for example "de" or "en". # Dominant meeting language, for example "de" or "en".
@@ -28,6 +30,12 @@ participants:
attendance_status: "present" attendance_status: "present"
notes: null notes: null
# Optional authoritative mapping from diarization labels to actual participants.
# Add entries only after a human or trusted external process confirms identity.
# Never infer mappings from conversational context. Unmapped labels stay anonymous.
speaker_mappings: {}
# SPEAKER_00: "participant-id"
mentioned_people: mentioned_people:
# People discussed or referenced but not present in the meeting. # People discussed or referenced but not present in the meeting.
# Mentioned people are not speakers and must not become responsible persons # Mentioned people are not speakers and must not become responsible persons
@@ -37,7 +45,7 @@ mentioned_people:
aliases: [] aliases: []
role: null role: null
department: null department: null
attendance_status: "not_present" attendance_status: "mentioned_only"
notes: null notes: null
organization: organization:
+142
View File
@@ -0,0 +1,142 @@
#!/usr/bin/env python3
"""Run the one-call direct protocol MVP from compact Whisper JSON."""
from __future__ import annotations
import argparse
import json
import re
import shutil
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Callable
REPO_ROOT = Path(__file__).resolve().parents[1]
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from src.meeting_lab.models.meeting_context import load_meeting_context # noqa: E402
from src.meeting_lab.protocol.history import persist_protocol_generation # noqa: E402
from src.meeting_lab.llm.ollama import DEFAULT_ENDPOINT # noqa: E402
from src.meeting_lab.protocol.generate_direct_protocol import ( # noqa: E402
DEFAULT_MODEL,
DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
generate_direct_protocol,
)
DEFAULT_OUTPUT_ROOT = Path("meeting_data/runs")
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Generate one direct protocol from compact Whisper JSON.")
parser.add_argument("transcript", type=Path)
parser.add_argument("--context", type=Path)
parser.add_argument("--output-root", type=Path, default=DEFAULT_OUTPUT_ROOT)
parser.add_argument("--model", default=DEFAULT_MODEL)
parser.add_argument("--ollama-endpoint", default=DEFAULT_ENDPOINT)
parser.add_argument(
"--safe-input-token-budget",
type=int,
default=DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
)
return parser.parse_args(argv)
def create_unique_run_dir(
output_root: Path,
transcript_stem: str,
now: Callable[[], datetime] = datetime.now,
) -> Path:
safe_stem = re.sub(r"[^A-Za-z0-9_.-]+", "_", transcript_stem).strip("._-") or "meeting"
base = output_root / f"{safe_stem}_{now().strftime('%Y%m%d_%H%M%S')}"
candidate = base
suffix = 1
while candidate.exists():
candidate = output_root / f"{base.name}_{suffix:02d}"
suffix += 1
candidate.mkdir(parents=True)
return candidate
def write_json(path: Path, data: Any) -> None:
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def run(args: argparse.Namespace) -> tuple[int, Path, Path | None]:
run_dir = create_unique_run_dir(args.output_root, args.transcript.stem)
timestamp = datetime.now().astimezone().isoformat(timespec="seconds")
started = time.perf_counter()
protocol_path: Path | None = None
metadata: dict[str, Any] = {
"run_id": run_dir.name,
"timestamp": timestamp,
"transcript_path": str(args.transcript.resolve()),
"context_path": str(args.context.resolve()) if args.context else None,
"model": args.model,
"ollama_endpoint": args.ollama_endpoint,
"status": "running",
"total_runtime_seconds": None,
"final_protocol_path": None,
}
try:
transcript_dir = run_dir / "transcript"
transcript_dir.mkdir()
if not args.transcript.is_file():
raise FileNotFoundError(f"Transcript file does not exist: {args.transcript}")
preserved_transcript = transcript_dir / "transcript.json"
shutil.copy2(args.transcript, preserved_transcript)
preserved_context: Path | None = None
if args.context is not None:
if not args.context.is_file():
raise FileNotFoundError(f"Meeting Context file does not exist: {args.context}")
context_dir = run_dir / "context"
context_dir.mkdir()
preserved_context = context_dir / "meeting_context.yaml"
shutil.copy2(args.context, preserved_context)
write_json(
run_dir / "input_manifest.json",
{
"transcript_source": str(args.transcript.resolve()),
"transcript_copy": str(preserved_transcript.resolve()),
"context_source": str(args.context.resolve()) if args.context else None,
"context_copy": str(preserved_context.resolve()) if preserved_context else None,
},
)
result = generate_direct_protocol(
preserved_transcript,
preserved_context,
model=args.model,
endpoint=args.ollama_endpoint,
safe_input_token_budget=args.safe_input_token_budget,
)
protocol_path = persist_protocol_generation(
run_dir,
result,
context=load_meeting_context(preserved_context) if preserved_context else None,
)
metadata["status"] = "completed"
metadata["final_protocol_path"] = str(protocol_path.resolve())
except Exception as exc:
metadata["status"] = "failed"
metadata["failure"] = f"{type(exc).__name__}: {exc}"
print(f"Error: {metadata['failure']}", file=sys.stderr)
finally:
metadata["total_runtime_seconds"] = round(time.perf_counter() - started, 3)
write_json(run_dir / "run_metadata.json", metadata)
return (0 if metadata["status"] == "completed" else 2), run_dir, protocol_path
def main(argv: list[str] | None = None) -> int:
code, _run_dir, protocol_path = run(parse_args(argv))
if protocol_path is not None:
print(protocol_path)
return code
if __name__ == "__main__":
raise SystemExit(main())
+151
View File
@@ -0,0 +1,151 @@
#!/usr/bin/env python3
"""CLI adapter for the reusable Meeting Lab MVP orchestration API."""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Any
REPO_ROOT = Path(__file__).resolve().parents[1]
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from src.meeting_lab.llm.ollama import DEFAULT_ENDPOINT # noqa: E402
from src.meeting_lab.models.meeting_context import MeetingContext # noqa: E402
from src.meeting_lab.orchestration.mvp import ( # noqa: E402
DEFAULT_DIARIZATION_MODEL,
DEFAULT_MODEL,
DEFAULT_OUTPUT_ROOT,
DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
MvpMeetingConfig,
create_unique_run_dir,
run_mvp_meeting,
)
from src.meeting_lab.progress import ProgressSink # noqa: E402
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Transcribe one meeting and generate one direct protocol."
)
parser.add_argument("audio_file", type=Path)
parser.add_argument("--whisper-model", type=Path, required=True)
parser.add_argument("--whisper-executable", default="whisper-cli")
parser.add_argument("--ffmpeg-executable", default="ffmpeg")
parser.add_argument(
"--audio-normalization",
action=argparse.BooleanOptionalAction,
default=True,
help=(
"Enable FFmpeg loudness normalization during canonical audio preparation "
"(default: enabled)."
),
)
parser.add_argument("--context", type=Path)
parser.add_argument("--output-root", type=Path, default=DEFAULT_OUTPUT_ROOT)
parser.add_argument("--language", default="de")
parser.add_argument(
"--threads",
default="auto",
help="Thread count or 'auto' for physical CPU cores (default: auto).",
)
parser.add_argument("--model", default=DEFAULT_MODEL)
parser.add_argument("--ollama-endpoint", default=DEFAULT_ENDPOINT)
parser.add_argument(
"--protocol-safe-input-token-budget",
type=int,
default=DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
help="Conservative estimated prompt-token limit before any Ollama request.",
)
parser.add_argument(
"--diarization",
choices=("auto", "gpu", "cpu", "off"),
default="off",
help="Optional Community-1 diarization device mode (default: off).",
)
parser.add_argument(
"--diarization-runtime",
choices=("native", "container"),
default="native",
help="Run pyannote in this Python environment or an explicit container.",
)
parser.add_argument(
"--diarization-container-image",
help="Container image required with --diarization-runtime container.",
)
parser.add_argument(
"--diarization-container-arg",
action="append",
default=[],
help="Additional docker argument; repeat and use = for values beginning with --.",
)
return parser.parse_args(argv)
def config_from_args(args: argparse.Namespace) -> MvpMeetingConfig:
return MvpMeetingConfig(
audio_file=args.audio_file,
whisper_model=args.whisper_model,
whisper_executable=args.whisper_executable,
ffmpeg_executable=args.ffmpeg_executable,
audio_normalization=args.audio_normalization,
context_file=args.context,
output_root=args.output_root,
language=args.language,
threads=args.threads,
model=args.model,
ollama_endpoint=args.ollama_endpoint,
protocol_safe_input_token_budget=args.protocol_safe_input_token_budget,
diarization=args.diarization,
diarization_runtime=args.diarization_runtime,
diarization_container_image=args.diarization_container_image,
diarization_container_args=tuple(args.diarization_container_arg),
)
def run(
args: argparse.Namespace,
*,
context_override: MeetingContext | dict[str, Any] | None = None,
progress_sink: ProgressSink | None = None,
) -> tuple[int, Path | None, Path | None]:
"""Compatibility wrapper for existing Python callers of the CLI module."""
result = run_mvp_meeting(
config_from_args(args),
meeting_context=context_override,
progress_sink=progress_sink,
)
return result.exit_code, result.run_dir, result.protocol_path
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
if args.diarization == "off":
print("Diarization: disabled")
else:
print(
f"Diarization: enabled; backend=pyannote.audio; "
f"model={DEFAULT_DIARIZATION_MODEL}; requested_device={args.diarization}; "
f"runtime={args.diarization_runtime}"
)
code, run_dir, protocol_path = run(args)
if run_dir is not None and args.diarization != "off":
metadata_path = run_dir / "diarization" / "metadata.json"
if metadata_path.is_file():
details = json.loads(metadata_path.read_text(encoding="utf-8"))
print(
f"Diarization result: device={details.get('actual_device')}; "
f"device_name={details.get('device_name') or 'n/a'}; "
f"runtime={details.get('runtime_seconds'):.3f}s; "
f"speakers={details.get('speaker_count')}; artifacts={metadata_path.parent}"
)
if protocol_path is not None:
print(protocol_path)
return code
if __name__ == "__main__":
raise SystemExit(main())
+51
View File
@@ -0,0 +1,51 @@
#!/usr/bin/env python3
"""Transcribe one audio file with whisper.cpp; do not generate a protocol."""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from src.meeting_lab.transcription.whisper import ( # noqa: E402
TranscriptionError,
transcribe_audio,
)
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Create a compact Meeting Lab transcript with whisper.cpp.")
parser.add_argument("audio_file", type=Path)
parser.add_argument("--model", type=Path, required=True, help="Path to a whisper.cpp GGML model.")
parser.add_argument("--output-dir", type=Path, required=True)
parser.add_argument("--language", default="auto", help="Language code or 'auto' (default: auto).")
parser.add_argument("--threads", default="auto", help="Thread count or 'auto' for physical CPU cores (default: auto).")
parser.add_argument("--whisper-executable", default="whisper-cli", help="whisper.cpp CLI executable (default: whisper-cli).")
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
try:
result = transcribe_audio(
args.audio_file,
args.model,
args.output_dir,
args.language,
executable=args.whisper_executable,
threads=args.threads,
)
except TranscriptionError as exc:
print(f"Error: {exc}")
return 1
print(f"Transcript: {result.transcript_json}")
print(f"Runtime: {result.runtime_seconds:.3f} seconds")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+5
View File
@@ -0,0 +1,5 @@
"""Canonical audio preparation boundary."""
from .preparation import AudioPreparationError, PreparedAudio, prepare_audio
__all__ = ["AudioPreparationError", "PreparedAudio", "prepare_audio"]
+190
View File
@@ -0,0 +1,190 @@
"""Prepare supported recordings for deterministic downstream processing."""
from __future__ import annotations
import os
import shutil
import subprocess
import wave
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from pathlib import Path
SUPPORTED_EXTENSIONS = {".wav", ".flac", ".m4a"}
CANONICAL_SAMPLE_RATE = 16_000
CANONICAL_CHANNELS = 1
CANONICAL_SAMPLE_WIDTH_BYTES = 2
CANONICAL_CODEC = "pcm_s16le"
DEFAULT_NORMALIZATION_FILTER = "loudnorm=I=-16:LRA=11:TP=-1.5"
DEFAULT_NORMALIZATION_METHOD = "ffmpeg_loudnorm"
class AudioPreparationError(RuntimeError):
"""Raised when source audio cannot be prepared as canonical WAV."""
@dataclass(frozen=True)
class PreparedAudio:
source_path: Path
source_format: str
prepared_path: Path
method: str
ffmpeg_executable: str
normalization_enabled: bool = True
normalization_method: str | None = DEFAULT_NORMALIZATION_METHOD
normalization_filter: str | None = DEFAULT_NORMALIZATION_FILTER
def metadata(self) -> dict[str, object]:
return {
"original_source_path": str(self.source_path.resolve()),
"original_source_name": self.source_path.name,
"original_format": self.source_format,
"prepared_audio_path": str(self.prepared_path.resolve()),
"preparation_method": self.method,
"ffmpeg_executable": self.ffmpeg_executable,
"normalization_enabled": self.normalization_enabled,
"normalization_method": self.normalization_method,
"normalization_filter": self.normalization_filter,
"canonical_output": {
"container": "wav",
"codec": CANONICAL_CODEC,
"channels": CANONICAL_CHANNELS,
"sample_rate_hz": CANONICAL_SAMPLE_RATE,
"bits_per_sample": CANONICAL_SAMPLE_WIDTH_BYTES * 8,
},
}
Runner = Callable[..., subprocess.CompletedProcess[str]]
def prepare_audio(
source_path: Path,
prepared_path: Path,
*,
ffmpeg_executable: str = "ffmpeg",
normalization_enabled: bool = True,
runner: Runner = subprocess.run,
) -> PreparedAudio:
"""Create and validate a canonical mono 16 kHz signed PCM16 WAV artifact."""
source_path = Path(source_path)
prepared_path = Path(prepared_path)
source_format = source_path.suffix.lower()
if not source_path.is_file():
raise AudioPreparationError(f"Source audio does not exist: {source_path}")
if source_format not in SUPPORTED_EXTENSIONS:
supported = ", ".join(sorted(SUPPORTED_EXTENSIONS))
raise AudioPreparationError(
f"Unsupported audio format {source_format or '<none>'!r}; supported: {supported}."
)
resolved_executable = _resolve_executable(ffmpeg_executable)
prepared_path.parent.mkdir(parents=True, exist_ok=True)
temporary_path = prepared_path.with_name(f".{prepared_path.name}.tmp.wav")
command_parts = [
resolved_executable,
"-nostdin",
"-hide_banner",
"-loglevel",
"error",
"-y",
"-i",
str(source_path),
"-map_metadata",
"-1",
"-vn",
]
if normalization_enabled:
command_parts.extend(("-af", DEFAULT_NORMALIZATION_FILTER))
command_parts.extend(
(
"-ac",
str(CANONICAL_CHANNELS),
"-ar",
str(CANONICAL_SAMPLE_RATE),
"-c:a",
CANONICAL_CODEC,
"-fflags",
"+bitexact",
str(temporary_path),
)
)
command: Sequence[str] = tuple(command_parts)
try:
completed = runner(command, capture_output=True, text=True, check=False)
except OSError as exc:
raise AudioPreparationError(f"Could not run FFmpeg: {exc}") from exc
if completed.returncode != 0:
detail = (
completed.stderr or completed.stdout or "no diagnostic output"
).strip()
raise AudioPreparationError(
f"FFmpeg failed to prepare {source_path.name} (exit {completed.returncode}): "
f"{detail}"
)
try:
_validate_canonical_wav(temporary_path)
os.replace(temporary_path, prepared_path)
except Exception:
temporary_path.unlink(missing_ok=True)
raise
return PreparedAudio(
source_path=source_path,
source_format=source_format.removeprefix("."),
prepared_path=prepared_path,
method="ffmpeg",
ffmpeg_executable=resolved_executable,
normalization_enabled=normalization_enabled,
normalization_method=(
DEFAULT_NORMALIZATION_METHOD if normalization_enabled else None
),
normalization_filter=(
DEFAULT_NORMALIZATION_FILTER if normalization_enabled else None
),
)
def _resolve_executable(executable: str) -> str:
value = executable.strip()
if not value:
raise AudioPreparationError("FFmpeg executable must not be empty.")
if Path(value).parent != Path("."):
path = Path(value)
if path.is_file() and os.access(path, os.X_OK):
return str(path)
raise AudioPreparationError(f"FFmpeg executable is not available: {value}")
resolved = shutil.which(value)
if resolved is None:
raise AudioPreparationError(
f"FFmpeg executable {value!r} was not found on PATH. Install FFmpeg or "
"configure its executable path."
)
return resolved
def _validate_canonical_wav(path: Path) -> None:
try:
with wave.open(str(path), "rb") as recording:
properties = (
recording.getnchannels(),
recording.getframerate(),
recording.getsampwidth(),
recording.getcomptype(),
)
except (OSError, EOFError, wave.Error) as exc:
raise AudioPreparationError(
f"FFmpeg did not produce a readable WAV file: {path}: {exc}"
) from exc
expected = (
CANONICAL_CHANNELS,
CANONICAL_SAMPLE_RATE,
CANONICAL_SAMPLE_WIDTH_BYTES,
"NONE",
)
if properties != expected:
raise AudioPreparationError(
"Prepared audio is not canonical mono 16 kHz PCM16 WAV: "
f"channels={properties[0]}, sample_rate={properties[1]}, "
f"sample_width={properties[2]}, compression={properties[3]}."
)
+20
View File
@@ -0,0 +1,20 @@
"""Optional speaker diarization and transcript alignment."""
from src.meeting_lab.diarization.alignment import align_transcript, write_diarized_transcript
from src.meeting_lab.diarization.backend import (
DEFAULT_MODEL,
DiarizationError,
DiarizationResult,
diarize_audio,
select_device,
)
__all__ = [
"DEFAULT_MODEL",
"DiarizationError",
"DiarizationResult",
"align_transcript",
"diarize_audio",
"select_device",
"write_diarized_transcript",
]
+129
View File
@@ -0,0 +1,129 @@
"""Deterministic Whisper-segment alignment to anonymous diarization turns."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
class AlignmentError(ValueError):
"""Raised when transcript or diarization inputs are malformed."""
def _number(value: Any, description: str) -> float:
if not isinstance(value, (int, float)) or isinstance(value, bool):
raise AlignmentError(f"{description} must be a number.")
return float(value)
def _validated_turns(turns: list[dict[str, Any]]) -> list[dict[str, Any]]:
validated = []
for index, turn in enumerate(turns):
if not isinstance(turn, dict):
raise AlignmentError(f"Diarization turn {index} must be an object.")
start = _number(turn.get("start"), f"Diarization turn {index} start")
end = _number(turn.get("end"), f"Diarization turn {index} end")
speaker = turn.get("speaker_id", turn.get("speaker"))
if end < start:
raise AlignmentError(f"Diarization turn {index} ends before it starts.")
if not isinstance(speaker, str) or not speaker.startswith("SPEAKER_"):
raise AlignmentError(
f"Diarization turn {index} must have an anonymous SPEAKER_ label."
)
validated.append({"start": start, "end": end, "speaker_id": speaker})
return validated
def align_transcript(
transcript: dict[str, Any], exclusive_turns: list[dict[str, Any]]
) -> dict[str, Any]:
"""Return a derived transcript using maximum exclusive-turn overlap per segment."""
if not isinstance(transcript, dict) or not isinstance(transcript.get("segments"), list):
raise AlignmentError("Whisper transcript must contain a 'segments' list.")
turns = _validated_turns(exclusive_turns)
aligned_segments: list[dict[str, Any]] = []
for index, source in enumerate(transcript["segments"]):
if not isinstance(source, dict):
raise AlignmentError(f"Transcript segment {index} must be an object.")
start = _number(source.get("start"), f"Transcript segment {index} start")
end = _number(source.get("end"), f"Transcript segment {index} end")
if end < start:
raise AlignmentError(f"Transcript segment {index} ends before it starts.")
overlap_by_speaker: dict[str, float] = {}
for turn in turns:
overlap = max(0.0, min(end, turn["end"]) - max(start, turn["start"]))
if overlap:
speaker = turn["speaker_id"]
overlap_by_speaker[speaker] = overlap_by_speaker.get(speaker, 0.0) + overlap
speaker_id = None
overlap_seconds = 0.0
if overlap_by_speaker:
speaker_id, overlap_seconds = min(
overlap_by_speaker.items(), key=lambda item: (-item[1], item[0])
)
duration = end - start
aligned = dict(source)
aligned.update(
{
"speaker_id": speaker_id,
"speaker_overlap_seconds": round(overlap_seconds, 6),
"speaker_overlap_ratio": round(
overlap_seconds / duration if duration > 0 else 0.0, 6
),
}
)
aligned_segments.append(aligned)
return {
"text": diarized_transcript_text(aligned_segments, include_end=True),
"segments": aligned_segments,
"speaker_labels_anonymous": True,
"alignment_source": "exclusive_diarization",
}
def _timestamp(seconds: float) -> str:
milliseconds = int(round(seconds * 1000))
hours, remainder = divmod(milliseconds, 3_600_000)
minutes, remainder = divmod(remainder, 60_000)
secs, millis = divmod(remainder, 1000)
return f"{hours:02d}:{minutes:02d}:{secs:02d}.{millis:03d}"
def diarized_transcript_text(
segments: list[dict[str, Any]], *, include_end: bool = True
) -> str:
lines = []
for segment in segments:
start = _timestamp(float(segment["start"]))
end = _timestamp(float(segment["end"]))
speaker = segment.get("speaker_id") or "SPEAKER_UNASSIGNED"
timestamp = f"[{start} - {end}]" if include_end else f"[{start}]"
lines.append(f"{timestamp} {speaker}: {str(segment.get('text', '')).strip()}")
return "\n".join(lines) + ("\n" if lines else "")
def write_diarized_transcript(
transcript_path: Path,
exclusive_turns_path: Path,
output_dir: Path,
) -> tuple[Path, Path]:
"""Read source artifacts and write a separate speaker-aware transcript pair."""
transcript = json.loads(Path(transcript_path).read_text(encoding="utf-8-sig"))
turns = json.loads(Path(exclusive_turns_path).read_text(encoding="utf-8"))
if not isinstance(turns, list):
raise AlignmentError("Exclusive diarization turns must contain a JSON list.")
derived = align_transcript(transcript, turns)
output_dir = Path(output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
json_path = output_dir / "transcript_diarized.json"
text_path = output_dir / "transcript_diarized.txt"
json_path.write_text(
json.dumps(derived, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
text_path.write_text(
diarized_transcript_text(derived["segments"], include_end=True), encoding="utf-8"
)
return json_path, text_path
+314
View File
@@ -0,0 +1,314 @@
"""pyannote Community-1 backend with native and isolated-container runtimes."""
from __future__ import annotations
import importlib.metadata
import json
import os
import subprocess
import time
import wave
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Literal, Sequence
DEFAULT_MODEL = "pyannote/speaker-diarization-community-1"
PYANNOTE_VERSION = "4.0.7"
DeviceMode = Literal["auto", "gpu", "cpu"]
RuntimeMode = Literal["native", "container"]
class DiarizationError(RuntimeError):
"""Raised when diarization configuration or execution fails."""
@dataclass(frozen=True)
class DiarizationResult:
output_dir: Path
metadata_path: Path
ordinary_rttm: Path
exclusive_rttm: Path
turns_json: Path
exclusive_turns_json: Path
metadata: dict[str, Any]
def _host_uid() -> int:
getter = getattr(os, "getuid", None)
if getter is None:
raise DiarizationError("Container diarization requires host UID discovery.")
return int(getter())
def _host_gid() -> int:
getter = getattr(os, "getgid", None)
if getter is None:
raise DiarizationError("Container diarization requires host GID discovery.")
return int(getter())
def select_device(mode: DeviceMode, torch_module: Any) -> tuple[Any, str | None]:
"""Resolve CPU/GPU without depending on the GPU vendor."""
if mode == "cpu":
return torch_module.device("cpu"), None
if mode not in ("auto", "gpu"):
raise DiarizationError(f"Unsupported diarization device mode: {mode}")
try:
available = bool(torch_module.cuda.is_available())
if available:
name = str(torch_module.cuda.get_device_name(0))
probe = torch_module.zeros(1, device="cuda")
del probe
return torch_module.device("cuda"), name
except Exception as exc:
if mode == "gpu":
raise DiarizationError(f"Requested PyTorch GPU is not usable: {exc}") from exc
if mode == "gpu":
raise DiarizationError("Requested PyTorch GPU is unavailable.")
return torch_module.device("cpu"), None
def _load_pcm_wave(audio_path: Path, torch_module: Any) -> tuple[Any, int, float, dict[str, Any]]:
try:
with wave.open(str(audio_path), "rb") as source:
channels = source.getnchannels()
sample_rate = source.getframerate()
sample_width = source.getsampwidth()
frame_count = source.getnframes()
pcm = bytearray(source.readframes(frame_count))
except (OSError, wave.Error) as exc:
raise DiarizationError(f"Cannot read PCM WAV input {audio_path}: {exc}") from exc
if channels != 1 or sample_rate != 16000 or sample_width != 2:
raise DiarizationError(
"Diarization currently requires mono 16 kHz signed 16-bit PCM WAV; "
f"got channels={channels}, sample_rate={sample_rate}, sample_width={sample_width}."
)
waveform = torch_module.frombuffer(pcm, dtype=torch_module.int16).to(
torch_module.float32
)
waveform = (waveform / 32768.0).reshape(channels, frame_count)
duration = frame_count / sample_rate
validation = {
"waveform_dtype": str(waveform.dtype),
"waveform_shape": list(waveform.shape),
"sample_rate": sample_rate,
"sample_count": frame_count,
"duration_seconds": duration,
"min_sample_value": waveform.min().item(),
"max_sample_value": waveform.max().item(),
"audio_loading": "python_wave_pcm16",
}
return waveform, sample_rate, duration, validation
def _turns(annotation: Any) -> list[dict[str, Any]]:
return [
{
"start": segment.start,
"end": segment.end,
"speaker_id": speaker,
}
for segment, _track, speaker in annotation.itertracks(yield_label=True)
]
def _write_json(path: Path, value: Any) -> None:
path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def _result_from_output(output_dir: Path) -> DiarizationResult:
metadata_path = output_dir / "metadata.json"
try:
metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise DiarizationError(f"Cannot read diarization metadata: {exc}") from exc
return DiarizationResult(
output_dir=output_dir,
metadata_path=metadata_path,
ordinary_rttm=output_dir / "diarization.rttm",
exclusive_rttm=output_dir / "exclusive_diarization.rttm",
turns_json=output_dir / "turns.json",
exclusive_turns_json=output_dir / "exclusive_turns.json",
metadata=metadata,
)
def _require_writable_output(output_dir: Path) -> None:
unwritable = [
path
for path in (output_dir, *output_dir.rglob("*"))
if not os.access(path, os.W_OK)
]
if unwritable:
rendered = ", ".join(str(path) for path in unwritable[:3])
if len(unwritable) > 3:
rendered += f", and {len(unwritable) - 3} more"
raise DiarizationError(
f"Container diarization artifacts are not writable by the host user: {rendered}"
)
def run_native_pyannote(
audio_path: Path,
output_dir: Path,
device_mode: DeviceMode,
*,
model: str = DEFAULT_MODEL,
) -> DiarizationResult:
"""Run one local pyannote inference using an in-memory waveform mapping."""
try:
import torch
from pyannote.audio import Pipeline
except ImportError as exc:
raise DiarizationError(
f"Native diarization requires pyannote.audio=={PYANNOTE_VERSION} and PyTorch."
) from exc
token = os.environ.get("HF_TOKEN")
if not token:
raise DiarizationError("HF_TOKEN is required for the pyannote model.")
output_dir.mkdir(parents=True, exist_ok=True)
waveform, sample_rate, duration, audio_metadata = _load_pcm_wave(audio_path, torch)
device, device_name = select_device(device_mode, torch)
try:
pipeline = Pipeline.from_pretrained(model, token=token)
pipeline.to(device)
started = time.perf_counter()
output = pipeline(
{
"waveform": waveform,
"sample_rate": sample_rate,
"uri": audio_path.stem,
}
)
runtime = time.perf_counter() - started
except Exception as exc:
raise DiarizationError(f"pyannote diarization failed: {type(exc).__name__}: {exc}") from exc
ordinary = getattr(output, "speaker_diarization", output)
exclusive = getattr(output, "exclusive_speaker_diarization", None)
if exclusive is None:
raise DiarizationError("Community-1 did not return exclusive diarization.")
ordinary_turns = _turns(ordinary)
exclusive_turns = _turns(exclusive)
with (output_dir / "diarization.rttm").open("w", encoding="utf-8") as handle:
ordinary.write_rttm(handle)
with (output_dir / "exclusive_diarization.rttm").open(
"w", encoding="utf-8"
) as handle:
exclusive.write_rttm(handle)
_write_json(output_dir / "turns.json", ordinary_turns)
_write_json(output_dir / "exclusive_turns.json", exclusive_turns)
speakers = sorted({turn["speaker_id"] for turn in ordinary_turns})
actual_device = str(device)
metadata = {
"backend": "pyannote.audio",
"model": model,
"pyannote_version": importlib.metadata.version("pyannote.audio"),
"torch_version": torch.__version__,
"hip_version": getattr(torch.version, "hip", None),
"cuda_version": getattr(torch.version, "cuda", None),
"runtime_adapter": "native",
"requested_device_mode": device_mode,
"actual_device": actual_device,
"device_name": device_name if actual_device == "cuda" else None,
"audio_duration_seconds": duration,
"runtime_seconds": runtime,
"rtf": runtime / duration,
"speaker_count": len(speakers),
"speaker_labels": speakers,
"turn_count": len(ordinary_turns),
"exclusive_turn_count": len(exclusive_turns),
"audio": audio_metadata,
"credentials_persisted": False,
"output_files": {
"ordinary_rttm": "diarization.rttm",
"exclusive_rttm": "exclusive_diarization.rttm",
"turns": "turns.json",
"exclusive_turns": "exclusive_turns.json",
},
}
_write_json(output_dir / "metadata.json", metadata)
return _result_from_output(output_dir)
def run_container_pyannote(
audio_path: Path,
output_dir: Path,
device_mode: DeviceMode,
*,
image: str,
container_args: Sequence[str] = (),
runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
uid_getter: Callable[[], int] = _host_uid,
gid_getter: Callable[[], int] = _host_gid,
) -> DiarizationResult:
"""Run the same backend in an explicitly configured disposable container."""
if not image.strip():
raise DiarizationError("A diarization container image is required.")
output_dir.mkdir(parents=True, exist_ok=True)
host_uid = uid_getter()
host_gid = gid_getter()
if host_uid < 0 or host_gid < 0:
raise DiarizationError("Host UID and GID must be non-negative integers.")
command = [
"docker", "run", "--rm", "--ipc=host", "--shm-size=8g", "-e", "HF_TOKEN",
*container_args,
"-v", f"{Path(audio_path).resolve()}:/input/audio.wav:ro",
"-v", f"{output_dir.resolve()}:/output:rw",
"-v", f"{Path(__file__).resolve().parents[3]}:/work/meeting-lab:ro",
"-w", "/work/meeting-lab",
image,
"/bin/bash", "-lc",
(
"inference_status=0; "
f"python -m pip install --disable-pip-version-check pyannote.audio=={PYANNOTE_VERSION} "
"> /output/pip-install.log 2>&1 && "
"python -m src.meeting_lab.diarization.container_entry "
f"/input/audio.wav /output --device {device_mode} || inference_status=$?; "
f"chown -R {host_uid}:{host_gid} /output || exit $?; "
"chmod -R u+rwX /output || exit $?; "
'exit "$inference_status"'
),
]
try:
completed = runner(command, check=False, capture_output=True, text=True)
except OSError as exc:
raise DiarizationError(f"Could not start diarization container: {exc}") from exc
(output_dir / "container_stdout.log").write_text(completed.stdout, encoding="utf-8")
(output_dir / "container_stderr.log").write_text(completed.stderr, encoding="utf-8")
if completed.returncode != 0:
detail = completed.stderr.strip() or completed.stdout.strip() or "no diagnostic output"
raise DiarizationError(
f"Diarization container failed with exit code {completed.returncode}: {detail}"
)
_require_writable_output(output_dir)
result = _result_from_output(output_dir)
metadata = dict(result.metadata)
metadata["runtime_adapter"] = "container"
_write_json(result.metadata_path, metadata)
return _result_from_output(output_dir)
def diarize_audio(
audio_path: Path,
output_dir: Path,
device_mode: DeviceMode,
*,
runtime: RuntimeMode = "native",
container_image: str | None = None,
container_args: Sequence[str] = (),
) -> DiarizationResult:
if runtime == "native":
return run_native_pyannote(audio_path, output_dir, device_mode)
if runtime == "container":
return run_container_pyannote(
audio_path,
output_dir,
device_mode,
image=container_image or "",
container_args=container_args,
)
raise DiarizationError(f"Unsupported diarization runtime: {runtime}")
@@ -0,0 +1,22 @@
"""Internal entry point for the isolated pyannote container adapter."""
from __future__ import annotations
import argparse
from pathlib import Path
from src.meeting_lab.diarization.backend import run_native_pyannote
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("audio", type=Path)
parser.add_argument("output", type=Path)
parser.add_argument("--device", choices=("auto", "gpu", "cpu"), required=True)
args = parser.parse_args()
run_native_pyannote(args.audio, args.output, args.device)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+100
View File
@@ -0,0 +1,100 @@
"""Minimal Ollama client behavior used by the direct protocol MVP."""
from __future__ import annotations
import time
from dataclasses import dataclass
from typing import Any
import requests
DEFAULT_ENDPOINT = "http://127.0.0.1:11434"
class OllamaError(RuntimeError):
"""Raised when Ollama cannot safely complete the requested operation."""
@dataclass(frozen=True)
class OllamaGeneration:
raw_response: dict[str, Any]
text: str
client_wall_time_seconds: float
def ollama_base_url(endpoint: str) -> str:
endpoint = endpoint.rstrip("/")
return endpoint.rsplit("/api/", 1)[0] if "/api/" in endpoint else endpoint
def generate_url(endpoint: str) -> str:
return f"{ollama_base_url(endpoint)}/api/generate"
def require_model(endpoint: str, model: str, timeout: int = 10) -> dict[str, Any]:
base_url = ollama_base_url(endpoint)
try:
response = requests.get(f"{base_url}/api/tags", timeout=timeout)
response.raise_for_status()
data = response.json()
except (requests.RequestException, ValueError) as exc:
raise OllamaError(f"Ollama endpoint is not reachable at {base_url}: {exc}") from exc
models = data.get("models") if isinstance(data, dict) else None
if not isinstance(models, list):
raise OllamaError("Ollama /api/tags returned a malformed response.")
installed = {
item.get("name")
for item in models
if isinstance(item, dict) and isinstance(item.get("name"), str)
}
if model not in installed:
raise OllamaError(f"Requested model is not installed in Ollama: {model}")
return {"base_url": base_url, "model": model, "installed": True}
def generate_once(
endpoint: str,
model: str,
prompt: str,
*,
timeout: int,
num_ctx: int,
num_predict: int,
num_thread: int | None = None,
) -> OllamaGeneration:
payload = {
"model": model,
"prompt": prompt,
"think": False,
"stream": False,
"options": {
"temperature": 0.0,
"num_ctx": num_ctx,
"num_predict": num_predict,
},
}
if num_thread is not None:
if type(num_thread) is not int or num_thread <= 0:
raise ValueError("Protocol thread count must be a positive integer.")
payload["options"]["num_thread"] = num_thread
started = time.perf_counter()
try:
response = requests.post(generate_url(endpoint), json=payload, timeout=timeout)
response.raise_for_status()
data = response.json()
except requests.RequestException as exc:
raise OllamaError(f"Ollama generation request failed: {exc}") from exc
except ValueError as exc:
raise OllamaError("Ollama generation response is not valid JSON.") from exc
wall_time = time.perf_counter() - started
if not isinstance(data, dict):
raise OllamaError("Ollama generation response must be a JSON object.")
text = data.get("response")
if not isinstance(text, str):
raise OllamaError("Ollama generation response has no string 'response' field.")
if not text.strip():
raise OllamaError("Ollama returned an empty protocol.")
return OllamaGeneration(data, text, wall_time)
+119 -11
View File
@@ -3,13 +3,15 @@
from __future__ import annotations from __future__ import annotations
import ast import ast
import copy
import re
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
SUPPORTED_SCHEMA_VERSIONS = {"1"} SUPPORTED_SCHEMA_VERSIONS = {"1"}
VALID_ATTENDANCE_STATUSES = {"present", "not_present", "absent"} VALID_ATTENDANCE_STATUSES = {"present", "mentioned_only"}
class MeetingContextValidationError(ValueError): class MeetingContextValidationError(ValueError):
@@ -29,6 +31,21 @@ class MeetingContext:
def meeting_id(self) -> str: def meeting_id(self) -> str:
return str(self.data["meeting"]["meeting_id"]) return str(self.data["meeting"]["meeting_id"])
@property
def speaker_mappings(self) -> dict[str, str]:
mappings = self.data.get("speaker_mappings")
return dict(mappings) if isinstance(mappings, dict) else {}
def participant_for_speaker(self, speaker_label: str) -> dict[str, Any] | None:
"""Resolve only an explicit authoritative mapping; never infer identity."""
participant_id = self.speaker_mappings.get(speaker_label)
if participant_id is None:
return None
for participant in self.data.get("participants", []):
if participant.get("participant_id") == participant_id:
return participant
return None
def provenance(self) -> dict[str, str]: def provenance(self) -> dict[str, str]:
return { return {
"meeting_id": self.meeting_id, "meeting_id": self.meeting_id,
@@ -42,8 +59,43 @@ def load_meeting_context(path: Path) -> MeetingContext:
if not isinstance(loaded, dict): if not isinstance(loaded, dict):
raise MeetingContextValidationError("Meeting Context must be a YAML object.") raise MeetingContextValidationError("Meeting Context must be a YAML object.")
validate_meeting_context(loaded) normalized = _with_attendance_defaults(loaded)
return MeetingContext(data=loaded, source_file=path) validate_meeting_context(normalized)
return MeetingContext(data=normalized, source_file=path)
def create_meeting_context(
data: dict[str, Any], *, source_file: Path = Path("<generated>")
) -> MeetingContext:
"""Validate structured data and return an immutable context boundary."""
validated = _with_attendance_defaults(data)
validate_meeting_context(validated)
return MeetingContext(data=validated, source_file=source_file)
def serialize_meeting_context_yaml(context: MeetingContext) -> str:
"""Serialize validated Meeting Context data deterministically as YAML."""
validate_meeting_context(context.data)
try:
import yaml # type: ignore[import-not-found]
except ModuleNotFoundError as exc:
raise MeetingContextValidationError(
"PyYAML is required to write Meeting Context YAML."
) from exc
return yaml.safe_dump(
context.data,
allow_unicode=True,
sort_keys=False,
default_flow_style=False,
)
def write_meeting_context(context: MeetingContext, path: Path) -> Path:
"""Persist a validated context without changing its schema or semantics."""
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(serialize_meeting_context_yaml(context), encoding="utf-8")
return path
def validate_meeting_context(data: dict[str, Any]) -> None: def validate_meeting_context(data: dict[str, Any]) -> None:
@@ -56,7 +108,9 @@ def validate_meeting_context(data: dict[str, Any]) -> None:
meeting = _require_mapping(data, "meeting") meeting = _require_mapping(data, "meeting")
_require_non_empty_string(meeting, "meeting.meeting_id") _require_non_empty_string(meeting, "meeting.meeting_id")
_require_non_empty_string(meeting, "meeting.title") _require_non_empty_string(meeting, "meeting.title")
_require_non_empty_string(meeting, "meeting.language") # Legacy contexts may omit language; protocol generation defaults to German.
if "language" in meeting:
_require_non_empty_string(meeting, "meeting.language")
organization = _optional_mapping(data.get("organization"), "organization") organization = _optional_mapping(data.get("organization"), "organization")
departments = _optional_list(organization.get("departments"), "organization.departments") departments = _optional_list(organization.get("departments"), "organization.departments")
@@ -74,10 +128,26 @@ def validate_meeting_context(data: dict[str, Any]) -> None:
+ ", ".join(collisions) + ", ".join(collisions)
) )
speaker_mappings = _optional_mapping(
data.get("speaker_mappings"), "speaker_mappings"
)
for speaker_label, participant_id in speaker_mappings.items():
if not isinstance(speaker_label, str) or re.fullmatch(
r"SPEAKER_\d+", speaker_label
) is None:
raise MeetingContextValidationError(
f"Invalid diarization speaker label: {speaker_label!r}."
)
if not isinstance(participant_id, str) or participant_id not in participant_ids:
raise MeetingContextValidationError(
f"speaker_mappings.{speaker_label} references unknown participant: "
f"{participant_id!r}."
)
for index, participant in enumerate(participants): for index, participant in enumerate(participants):
item_path = f"participants[{index}]" item_path = f"participants[{index}]"
_validate_attendance(participant, item_path) status = _validate_attendance(participant, item_path, default="present")
if participant.get("attendance_status") != "present": if status != "present":
raise MeetingContextValidationError( raise MeetingContextValidationError(
f"{item_path}.attendance_status must be 'present'." f"{item_path}.attendance_status must be 'present'."
) )
@@ -85,10 +155,10 @@ def validate_meeting_context(data: dict[str, Any]) -> None:
for index, person in enumerate(mentioned_people): for index, person in enumerate(mentioned_people):
item_path = f"mentioned_people[{index}]" item_path = f"mentioned_people[{index}]"
_validate_attendance(person, item_path) status = _validate_attendance(person, item_path, default="mentioned_only")
if person.get("attendance_status") == "present": if status != "mentioned_only":
raise MeetingContextValidationError( raise MeetingContextValidationError(
f"{item_path}.attendance_status must not be 'present'." f"{item_path}.attendance_status must be 'mentioned_only'."
) )
_validate_department_reference(person, item_path, department_ids) _validate_department_reference(person, item_path, department_ids)
@@ -131,6 +201,28 @@ def render_meeting_context_for_prompt(context: MeetingContext) -> str:
for participant in participants: for participant in participants:
lines.append(_render_person_line(participant, "participant_id", departments_by_id)) lines.append(_render_person_line(participant, "participant_id", departments_by_id))
speaker_mappings = context.speaker_mappings
if speaker_mappings:
participants_by_id = {
participant["participant_id"]: participant
for participant in participants
if isinstance(participant, dict) and participant.get("participant_id")
}
lines.extend(
[
"",
"Confirmed diarization speaker mappings (authoritative):",
"- Use only these explicit mappings. Never infer identities for other speaker labels.",
"- Unmapped SPEAKER_XX labels must remain anonymous.",
]
)
for speaker_label, participant_id in sorted(speaker_mappings.items()):
participant = participants_by_id[participant_id]
lines.append(
f"- {speaker_label}: {_text(participant.get('display_name'))} "
f"(participant_id: {participant_id})"
)
mentioned_people = _optional_list(data.get("mentioned_people"), "mentioned_people") mentioned_people = _optional_list(data.get("mentioned_people"), "mentioned_people")
if mentioned_people: if mentioned_people:
lines.extend(["", "Mentioned but absent people:"]) lines.extend(["", "Mentioned but absent people:"])
@@ -245,12 +337,28 @@ def _collect_unique_ids(items: list[Any], key: str, path: str) -> set[str]:
return ids return ids
def _validate_attendance(item: dict[str, Any], path: str) -> None: def _validate_attendance(item: dict[str, Any], path: str, *, default: str) -> str:
status = item.get("attendance_status") status = item.get("attendance_status", default)
if status not in VALID_ATTENDANCE_STATUSES: if status not in VALID_ATTENDANCE_STATUSES:
raise MeetingContextValidationError( raise MeetingContextValidationError(
f"{path}.attendance_status has invalid value: {status!r}." f"{path}.attendance_status has invalid value: {status!r}."
) )
return status
def _with_attendance_defaults(data: dict[str, Any]) -> dict[str, Any]:
normalized = copy.deepcopy(data)
participants = normalized.get("participants")
if isinstance(participants, list):
for participant in participants:
if isinstance(participant, dict):
participant.setdefault("attendance_status", "present")
mentioned_people = normalized.get("mentioned_people")
if isinstance(mentioned_people, list):
for person in mentioned_people:
if isinstance(person, dict):
person.setdefault("attendance_status", "mentioned_only")
return normalized
def _validate_department_reference( def _validate_department_reference(
+17
View File
@@ -0,0 +1,17 @@
"""Reusable orchestration APIs for Meeting Lab applications and CLIs."""
from src.meeting_lab.orchestration.mvp import (
DEFAULT_OUTPUT_ROOT,
MvpMeetingConfig,
MvpRunResult,
create_unique_run_dir,
run_mvp_meeting,
)
__all__ = [
"DEFAULT_OUTPUT_ROOT",
"MvpMeetingConfig",
"MvpRunResult",
"create_unique_run_dir",
"run_mvp_meeting",
]
+514
View File
@@ -0,0 +1,514 @@
"""Reusable audio-to-direct-protocol MVP orchestration."""
from __future__ import annotations
import json
import re
import shutil
import sys
import tempfile
import time
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any
from src.meeting_lab.audio import prepare_audio
from src.meeting_lab.diarization import (
DEFAULT_MODEL as DEFAULT_DIARIZATION_MODEL,
diarize_audio,
write_diarized_transcript,
)
from src.meeting_lab.llm.ollama import DEFAULT_ENDPOINT
from src.meeting_lab.models.meeting_context import (
MeetingContext,
create_meeting_context,
load_meeting_context,
validate_meeting_context,
write_meeting_context,
)
from src.meeting_lab.progress import ProgressEvent, ProgressSink, ProgressStatus
from src.meeting_lab.protocol.history import persist_protocol_generation
from src.meeting_lab.protocol.generate_direct_protocol import (
DEFAULT_MODEL,
DEFAULT_NUM_CTX,
DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
generate_direct_protocol,
load_compact_transcript,
)
from src.meeting_lab.transcription.whisper import transcribe_audio
DEFAULT_OUTPUT_ROOT = Path("meeting_data/runs")
ContextInput = MeetingContext | Mapping[str, Any]
@dataclass(frozen=True)
class MvpMeetingConfig:
audio_file: Path
whisper_model: Path
whisper_executable: str = "whisper-cli"
ffmpeg_executable: str = "ffmpeg"
audio_normalization: bool = True
context_file: Path | None = None
output_root: Path = DEFAULT_OUTPUT_ROOT
language: str = "de"
threads: str | int = "auto"
model: str = DEFAULT_MODEL
ollama_endpoint: str = DEFAULT_ENDPOINT
glossary_aliases: Mapping[str, str] | None = None
protocol_num_thread: int | None = None
protocol_num_ctx: int = DEFAULT_NUM_CTX
protocol_safe_input_token_budget: int = DEFAULT_SAFE_INPUT_TOKEN_BUDGET
diarization: str = "off"
diarization_runtime: str = "native"
diarization_container_image: str | None = None
diarization_container_args: Sequence[str] = ()
stop_after_diarization: bool = False
@dataclass(frozen=True)
class MvpRunResult:
exit_code: int
run_dir: Path | None
protocol_path: Path | None
def regenerate_mvp_protocol(
run_dir: Path,
*,
meeting_context: ContextInput,
model: str = DEFAULT_MODEL,
ollama_endpoint: str = DEFAULT_ENDPOINT,
glossary_aliases: Mapping[str, str] | None = None,
protocol_num_thread: int | None = None,
protocol_num_ctx: int = DEFAULT_NUM_CTX,
protocol_safe_input_token_budget: int = DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
progress_sink: ProgressSink | None = None,
) -> MvpRunResult:
"""Regenerate only protocol artifacts from an existing completed run."""
started = time.perf_counter()
run_dir = Path(run_dir)
context = _effective_context(meeting_context)
if context is None:
raise ValueError("Meeting Context is required for protocol regeneration.")
if protocol_num_thread is not None and (
type(protocol_num_thread) is not int or protocol_num_thread <= 0
):
raise ValueError("Protocol Ollama thread count must be a positive integer.")
if protocol_num_ctx <= 0:
raise ValueError("Protocol Ollama context size must be positive.")
if protocol_safe_input_token_budget <= 0:
raise ValueError("Protocol safe input token budget must be positive.")
diarized_transcript = run_dir / "diarization" / "transcript_diarized.json"
plain_transcript = run_dir / "transcript" / "transcript.json"
transcript_path = (
diarized_transcript if diarized_transcript.is_file() else plain_transcript
)
if not transcript_path.is_file():
raise FileNotFoundError(
f"Existing run has no protocol transcript artifact: {run_dir}"
)
_emit(progress_sink, "protocol_generation", "started", started)
try:
with tempfile.TemporaryDirectory(
prefix=".protocol-context-", dir=run_dir, ignore_cleanup_errors=True
) as temporary:
context_path = Path(temporary) / "meeting_context.yaml"
write_meeting_context(context, context_path)
result = generate_direct_protocol(
transcript_path,
context_path,
model=model,
endpoint=ollama_endpoint,
num_ctx=protocol_num_ctx,
num_thread=protocol_num_thread,
glossary_aliases=glossary_aliases,
safe_input_token_budget=protocol_safe_input_token_budget,
)
protocol_path = persist_protocol_generation(run_dir, result, context=context)
except Exception as exc:
_record_protocol_generation_attempt(
run_dir, status="failed", runtime_seconds=time.perf_counter() - started,
failure={"type": type(exc).__name__, "message": str(exc)},
)
_emit(
progress_sink,
"failed",
"failed",
started,
message=f"protocol_generation: {type(exc).__name__}: {exc}",
)
raise
_record_protocol_generation_attempt(
run_dir, status="completed", runtime_seconds=time.perf_counter() - started
)
_emit(progress_sink, "protocol_generation", "completed", started)
_emit(progress_sink, "completed", "completed", started)
return MvpRunResult(0, run_dir, protocol_path)
def _record_protocol_generation_attempt(
run_dir: Path,
*,
status: str,
runtime_seconds: float,
failure: dict[str, str] | None = None,
) -> None:
"""Record downstream attempts without changing completed upstream artifacts."""
metadata_path = run_dir / "run_metadata.json"
if not metadata_path.is_file():
return
metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
attempts = metadata.setdefault("protocol_generation_attempts", [])
if not isinstance(attempts, list):
attempts = metadata["protocol_generation_attempts"] = []
attempt: dict[str, Any] = {
"status": status,
"runtime_seconds": round(runtime_seconds, 3),
}
if failure is not None:
attempt["failure"] = failure
attempts.append(attempt)
metadata["status"] = "completed" if status == "completed" else "awaiting_speaker_review"
_write_json(metadata_path, metadata)
def create_unique_run_dir(
output_root: Path,
meeting_name: str,
now: Callable[[], datetime] = datetime.now,
) -> Path:
safe_name = re.sub(r"[^A-Za-z0-9_.-]+", "_", meeting_name).strip("._-") or "meeting"
base = output_root / f"{safe_name}_{now().strftime('%Y%m%d_%H%M%S')}"
candidate = base
suffix = 1
while candidate.exists():
candidate = output_root / f"{base.name}_{suffix:02d}"
suffix += 1
candidate.mkdir(parents=True)
return candidate
def _write_json(path: Path, value: Any) -> None:
path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def _effective_context(value: ContextInput | None) -> MeetingContext | None:
if value is None:
return None
if isinstance(value, MeetingContext):
validate_meeting_context(value.data)
return value
if isinstance(value, Mapping):
return create_meeting_context(dict(value), source_file=Path("<programmatic>"))
raise TypeError("meeting_context must be MeetingContext, mapping, or None.")
def _validate_inputs(
config: MvpMeetingConfig, meeting_context: MeetingContext | None
) -> None:
if not config.audio_file.is_file():
raise FileNotFoundError(f"Audio file does not exist: {config.audio_file}")
if not config.whisper_model.is_file():
raise FileNotFoundError(f"Whisper model does not exist: {config.whisper_model}")
if meeting_context is not None and config.context_file is not None:
raise ValueError("Use either a context file or a programmatic Meeting Context, not both.")
if meeting_context is not None:
validate_meeting_context(meeting_context.data)
elif config.context_file is not None:
if not config.context_file.is_file():
raise FileNotFoundError(
f"Meeting Context file does not exist: {config.context_file}"
)
load_meeting_context(config.context_file)
if config.diarization not in ("off", "auto", "gpu", "cpu"):
raise ValueError(f"Unsupported diarization mode: {config.diarization}")
if config.diarization_runtime not in ("native", "container"):
raise ValueError(
f"Unsupported diarization runtime: {config.diarization_runtime}"
)
if (
config.diarization != "off"
and config.diarization_runtime == "container"
and not config.diarization_container_image
):
raise ValueError("A diarization container image is required.")
if config.protocol_safe_input_token_budget <= 0:
raise ValueError("Protocol safe input token budget must be positive.")
if config.protocol_num_thread is not None and (
type(config.protocol_num_thread) is not int or config.protocol_num_thread <= 0
):
raise ValueError("Protocol Ollama thread count must be a positive integer.")
if config.protocol_num_ctx <= 0:
raise ValueError("Protocol Ollama context size must be positive.")
def _emit(
sink: ProgressSink | None,
stage: str,
status: ProgressStatus,
overall_started: float,
*,
message: str | None = None,
) -> None:
if sink is not None:
sink(
ProgressEvent(
stage=stage,
status=status,
elapsed_seconds=time.perf_counter() - overall_started,
message=message,
)
)
def run_mvp_meeting(
config: MvpMeetingConfig,
*,
meeting_context: ContextInput | None = None,
progress_sink: ProgressSink | None = None,
) -> MvpRunResult:
"""Run the existing MVP directly, without subprocess or GUI dependencies."""
overall_started = time.perf_counter()
validation_started = time.perf_counter()
_emit(progress_sink, "preparing", "started", overall_started)
try:
effective_context = _effective_context(meeting_context)
_validate_inputs(config, effective_context)
except Exception as exc:
_emit(
progress_sink,
"failed",
"failed",
overall_started,
message=f"preparing: {type(exc).__name__}: {exc}",
)
print(f"Error: {type(exc).__name__}: {exc}", file=sys.stderr)
return MvpRunResult(2, None, None)
validation_runtime = time.perf_counter() - validation_started
run_dir = create_unique_run_dir(config.output_root, config.audio_file.stem)
timestamp = datetime.now().astimezone().isoformat(timespec="seconds")
transcript_path = run_dir / "transcript" / "transcript.json"
protocol_path = run_dir / "protocol.md"
stage_runtimes: dict[str, float | None] = {
"validation": round(validation_runtime, 3),
"setup": None,
"audio_preparation": None,
"whisper": None,
"transcript_validation": None,
"protocol": None,
}
if config.diarization != "off":
stage_runtimes["diarization"] = None
stage_runtimes["diarization_alignment"] = None
metadata: dict[str, Any] = {
"run_id": run_dir.name,
"timestamp": timestamp,
"input_audio": str(config.audio_file.resolve()),
"audio_preparation": None,
"transcript_output": str(transcript_path.resolve()),
"protocol_output": str(protocol_path.resolve()),
"whisper_model": str(config.whisper_model.resolve()),
"model": config.model,
"ollama_endpoint": config.ollama_endpoint,
"status": "running",
"stage_runtimes_seconds": stage_runtimes,
"total_runtime_seconds": None,
"failure": None,
"diarization": {
"enabled": config.diarization != "off",
"backend": "pyannote.audio" if config.diarization != "off" else None,
"model": DEFAULT_DIARIZATION_MODEL if config.diarization != "off" else None,
"requested_device_mode": config.diarization,
"runtime": config.diarization_runtime if config.diarization != "off" else None,
"metadata_path": None,
"transcript_diarized": None,
},
}
current_stage = "preparing"
stage_started = time.perf_counter()
try:
audio_dir = run_dir / "audio"
transcript_dir = run_dir / "transcript"
context_dir = run_dir / "context"
protocol_dir = run_dir / "protocol"
audio_dir.mkdir()
transcript_dir.mkdir()
context_dir.mkdir()
protocol_dir.mkdir()
_write_json(
audio_dir / "input_manifest.json",
{
"source_file": str(config.audio_file.resolve()),
"filename": config.audio_file.name,
"size_bytes": config.audio_file.stat().st_size,
},
)
preparation_started = time.perf_counter()
current_stage = "audio_preparation"
stage_started = preparation_started
prepared_audio = prepare_audio(
config.audio_file,
audio_dir / "prepared.wav",
ffmpeg_executable=config.ffmpeg_executable,
normalization_enabled=config.audio_normalization,
)
stage_runtimes["audio_preparation"] = round(
time.perf_counter() - preparation_started, 3
)
current_stage = "preparing"
preparation_metadata = prepared_audio.metadata()
metadata["audio_preparation"] = preparation_metadata
_write_json(audio_dir / "preparation_metadata.json", preparation_metadata)
_write_json(
audio_dir / "input_manifest.json",
{
"source_file": str(config.audio_file.resolve()),
"filename": config.audio_file.name,
"size_bytes": config.audio_file.stat().st_size,
"format": config.audio_file.suffix.lower().removeprefix("."),
"prepared_audio": preparation_metadata,
},
)
preserved_context: Path | None = None
if effective_context is not None:
preserved_context = context_dir / "meeting_context.yaml"
write_meeting_context(effective_context, preserved_context)
elif config.context_file is not None:
preserved_context = context_dir / "meeting_context.yaml"
shutil.copy2(config.context_file, preserved_context)
stage_runtimes["setup"] = round(time.perf_counter() - stage_started, 3)
_emit(progress_sink, "preparing", "completed", overall_started)
current_stage = "transcription"
stage_started = time.perf_counter()
_emit(progress_sink, "transcription", "started", overall_started)
transcription = transcribe_audio(
prepared_audio.prepared_path,
config.whisper_model,
transcript_dir,
config.language,
executable=config.whisper_executable,
threads=config.threads,
)
stage_runtimes["whisper"] = round(time.perf_counter() - stage_started, 3)
_emit(progress_sink, "transcription", "completed", overall_started)
stage_started = time.perf_counter()
load_compact_transcript(transcription.transcript_json)
stage_runtimes["transcript_validation"] = round(
time.perf_counter() - stage_started, 3
)
protocol_transcript = transcription.transcript_json
if config.diarization != "off":
current_stage = "diarization"
stage_started = time.perf_counter()
_emit(progress_sink, "diarization", "started", overall_started)
diarization_dir = run_dir / "diarization"
diarization = diarize_audio(
prepared_audio.prepared_path,
diarization_dir,
config.diarization,
runtime=config.diarization_runtime,
container_image=config.diarization_container_image,
container_args=config.diarization_container_args,
)
stage_runtimes["diarization"] = round(
time.perf_counter() - stage_started, 3
)
metadata["diarization"].update(
{
"actual_device": diarization.metadata.get("actual_device"),
"device_name": diarization.metadata.get("device_name"),
"runtime_seconds": diarization.metadata.get("runtime_seconds"),
"speaker_count": diarization.metadata.get("speaker_count"),
"metadata_path": str(diarization.metadata_path.resolve()),
}
)
stage_started = time.perf_counter()
protocol_transcript, diarized_text = write_diarized_transcript(
transcription.transcript_json,
diarization.exclusive_turns_json,
diarization_dir,
)
load_compact_transcript(protocol_transcript)
stage_runtimes["diarization_alignment"] = round(
time.perf_counter() - stage_started, 3
)
metadata["diarization"].update(
{
"transcript_diarized": str(protocol_transcript.resolve()),
"transcript_diarized_text": str(diarized_text.resolve()),
}
)
_emit(progress_sink, "diarization", "completed", overall_started)
if config.stop_after_diarization:
metadata["status"] = "awaiting_speaker_review"
metadata["speaker_review"] = {
"status": "awaiting_optional_mapping",
"speaker_count": metadata["diarization"].get("speaker_count"),
}
_emit(progress_sink, "completed", "completed", overall_started)
return MvpRunResult(0, run_dir, None)
current_stage = "protocol_generation"
stage_started = time.perf_counter()
_emit(progress_sink, "protocol_generation", "started", overall_started)
result = generate_direct_protocol(
protocol_transcript,
preserved_context,
model=config.model,
endpoint=config.ollama_endpoint,
num_ctx=config.protocol_num_ctx,
num_thread=config.protocol_num_thread,
glossary_aliases=config.glossary_aliases,
safe_input_token_budget=config.protocol_safe_input_token_budget,
)
stage_runtimes["protocol"] = round(time.perf_counter() - stage_started, 3)
protocol_path = persist_protocol_generation(run_dir, result, context=effective_context)
_emit(progress_sink, "protocol_generation", "completed", overall_started)
metadata["status"] = "completed"
_emit(progress_sink, "completed", "completed", overall_started)
except Exception as exc:
metadata_stage = {
"preparing": "setup",
"audio_preparation": "audio_preparation",
"transcription": "whisper",
"diarization": "diarization",
"protocol_generation": "protocol",
}.get(current_stage, current_stage)
runtime_key = metadata_stage
if runtime_key in stage_runtimes and stage_runtimes[runtime_key] is None:
stage_runtimes[runtime_key] = round(time.perf_counter() - stage_started, 3)
metadata["status"] = "failed"
metadata["failure"] = {
"stage": metadata_stage,
"type": type(exc).__name__,
"message": str(exc),
}
protocol_path = None
_emit(
progress_sink,
"failed",
"failed",
overall_started,
message=f"{current_stage}: {type(exc).__name__}: {exc}",
)
print(f"Error: {type(exc).__name__}: {exc}", file=sys.stderr)
finally:
metadata["total_runtime_seconds"] = round(time.perf_counter() - overall_started, 3)
_write_json(run_dir / "run_metadata.json", metadata)
exit_code = 0 if metadata["status"] == "completed" else 2
return MvpRunResult(exit_code, run_dir, protocol_path)
+21
View File
@@ -0,0 +1,21 @@
"""Small observer boundary for long-running Meeting Lab operations."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Callable, Literal
ProgressStatus = Literal["started", "completed", "failed"]
@dataclass(frozen=True)
class ProgressEvent:
stage: str
status: ProgressStatus
elapsed_seconds: float
progress: float | None = None
message: str | None = None
ProgressSink = Callable[[ProgressEvent], None]
@@ -0,0 +1,36 @@
"""Prompt construction for the direct transcript-to-protocol MVP."""
from __future__ import annotations
DIRECT_PROTOCOL_INSTRUCTION = """Erstelle aus dem vollständigen Transkript und dem Meeting-Kontext ein vollständiges, strukturiertes und professionelles internes Besprechungsprotokoll.
Das Protokoll muss themenorientiert sein, nicht chronologisch und nicht nach technischen Kategorien gegliedert. Beginne mit # Meeting Protocol. Verwende für jedes kohärente Thema eine Überschrift ## <Thema> und darunter eine strukturierte Synthese der Diskussion. Bewahre relevante Diskussionsverläufe, unterschiedliche Positionen, offene Punkte und Entscheidungsgrundlagen. Dokumentiere die wesentlichen Inhalte nachvollziehbar und fasse Themenblöcke so zusammen, dass auch Personen, die nicht am Meeting teilgenommen haben, den Kontext und die Entwicklung der Diskussion verstehen können. Nenne Entscheidungen oder abgestimmte Positionen nur, wenn sie tatsächlich belegt sind. Führe Maßnahmen nur auf, wenn eine konkrete zukünftige Handlung gestützt ist; nenne verantwortliche Personen und Fristen ausschließlich bei expliziter Zuweisung, Annahme oder Bestätigung im Transkript. Vorschläge, Einwände, Möglichkeiten und vorläufige Ideen sind keine Entscheidungen oder Verpflichtungen. Bewahre relevante Einschränkungen und ungelöste Meinungsverschiedenheiten. Nenne offene Punkte nur, wenn sie wirklich offen bleiben. Nicht jedes Thema benötigt Entscheidungen, Maßnahmen oder offene Punkte.
Erzeuge keine reine Wiedergabe des Transkripts und verlängere das Protokoll nicht unnötig durch Wiederholungen. Synthetisiere zusammengehörige Aussagen, entferne Füllwörter und Gesprächsrauschen und erfinde keine Fakten, Entscheidungen, Zustimmungen, Verantwortlichen oder Fristen. Gib kein JSON, keine internen Labels und keine Analyse oder Denkprotokolle aus. Das Ergebnis soll als Markdown-Protokoll nach geringfügiger menschlicher Redaktion intern versendbar sein. Eine kompakte themenübergreifende Maßnahmenliste am Ende ist optional, wenn sie nützlich und vollständig belegt ist."""
COMPACT_DIARIZED_PROTOCOL_INSTRUCTION = """Erstelle aus dem vollständigen Transkript und Meeting-Kontext ein vollständiges, professionelles internes Besprechungsprotokoll. Das Transkript ist in aufeinanderfolgende anonyme Sprecherblöcke gegliedert.
Beginne mit # Meeting Protocol. Gliedere themenorientiert mit ## <Thema> und synthetisiere je Thema den relevanten Diskussionsverlauf, Kontext, unterschiedliche Positionen, Entscheidungsgrundlagen, Einschränkungen und ungelöste Meinungsverschiedenheiten so, dass Dritte ihn nachvollziehen können. Nenne Entscheidungen nur bei Beleg. Nenne Maßnahmen, Verantwortliche und Fristen nur bei expliziter Zuweisung, Annahme oder Bestätigung; Vorschläge sind keine Verpflichtungen.
Entferne nur Wiederholungen, Füllwörter und Gesprächsrauschen. Erfinde keine Fakten oder Identitäten. Gib kein JSON, keine Sprecherlabels und kein Denkprotokoll aus. Eine belegte themenübergreifende Maßnahmenliste am Ende ist optional."""
MAPPED_SPEAKER_ATTRIBUTION_INSTRUCTION = """Nutze die autoritativen SPEAKER_XX-zu-Teilnehmer-Zuordnungen im Meeting-Kontext, um ausdrücklich belegte Aussagen, Positionen, Entscheidungen, Zuweisungen und angenommene persönliche Verpflichtungen namentlich zuzuordnen. Eine ausdrückliche Ich-Zusage eines zugeordneten Sprechers belegt persönliche Verantwortung. Unterscheide stets den Sprecher einer Aussage von darin nur erwähnten Personen. Leite für nicht zugeordnete Sprecher keine Identität ab und erfinde keine persönliche Verantwortung. Gib die technischen SPEAKER_XX-Bezeichnungen nicht im nutzerseitigen Protokoll aus."""
def build_direct_protocol_prompt(
transcript: str,
meeting_context: str | None = None,
*,
instruction: str = DIRECT_PROTOCOL_INSTRUCTION,
meeting_language: str = "de",
) -> str:
context = meeting_context.strip() if meeting_context else "Kein Meeting-Kontext bereitgestellt."
language = {"de": "German", "en": "English"}.get(meeting_language, meeting_language)
return (
f"{instruction}\n\n"
f"Write the meeting protocol in {language}. "
"Preserve speaker and person names exactly as supplied.\n\n"
f"MEETING-KONTEXT:\n{context}\n\n"
f"VOLLSTAENDIGES TRANSKRIPT:\n{transcript.strip()}\n"
)
@@ -0,0 +1,253 @@
"""One-call direct protocol generation from a compact Whisper transcript."""
from __future__ import annotations
import json
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable
from src.meeting_lab.llm.ollama import (
DEFAULT_ENDPOINT,
OllamaGeneration,
generate_once,
require_model,
)
from src.meeting_lab.models.meeting_context import (
MeetingContext,
load_meeting_context,
render_meeting_context_for_prompt,
)
from src.meeting_lab.protocol.direct_protocol_prompt import (
COMPACT_DIARIZED_PROTOCOL_INSTRUCTION,
MAPPED_SPEAKER_ATTRIBUTION_INSTRUCTION,
build_direct_protocol_prompt,
)
from src.meeting_lab.protocol.transcript_input import (
TranscriptInputError,
compact_diarized_transcript,
plain_segment_transcript,
)
DEFAULT_MODEL = "qwen3.6:35B-A3B"
DEFAULT_NUM_CTX = 32768
DEFAULT_NUM_PREDICT = 8192
DEFAULT_TIMEOUT = 1800
DEFAULT_SAFE_INPUT_TOKEN_BUDGET = 29_000
ESTIMATED_UTF8_BYTES_PER_TOKEN = 4.4
class DirectProtocolError(ValueError):
"""Raised for invalid direct-protocol inputs or model output."""
@dataclass(frozen=True)
class DirectProtocolResult:
protocol_text: str
exact_prompt: str
model_metadata: dict[str, Any]
runtime_metadata: dict[str, Any]
raw_response: dict[str, Any]
transcript_input: str | None = None
@dataclass(frozen=True)
class SelectedTranscriptInput:
text: str
prompt: str
representation: str
estimated_input_tokens: int
safe_input_token_budget: int
fallback_used: bool
diarization_enabled: bool
def load_compact_transcript(path: Path) -> str:
data = _load_transcript_document(path)
text = data.get("text")
if not isinstance(text, str) or not text.strip():
raise DirectProtocolError("Transcript top-level 'text' must be a non-empty string.")
return text
def _load_transcript_document(path: Path) -> dict[str, Any]:
if not path.is_file():
raise DirectProtocolError(f"Transcript file does not exist: {path}")
try:
data = json.loads(path.read_text(encoding="utf-8-sig"))
except json.JSONDecodeError as exc:
raise DirectProtocolError(f"Transcript is not valid JSON: {path}: {exc}") from exc
if not isinstance(data, dict):
raise DirectProtocolError("Transcript JSON must contain a top-level object.")
if "text" not in data:
raise DirectProtocolError("Transcript JSON must contain top-level 'text'.")
return data
def estimate_input_tokens(prompt: str) -> int:
"""Estimate tokens without adding a model-specific tokenizer dependency."""
byte_count = len(prompt.encode("utf-8"))
return max(1, int(byte_count / ESTIMATED_UTF8_BYTES_PER_TOKEN + 0.999999))
def select_transcript_input(
transcript: dict[str, Any],
rendered_context: str | None,
*,
safe_input_token_budget: int = DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
meeting_language: str = "de",
) -> SelectedTranscriptInput:
"""Select complete prompt input without allowing silent tail truncation."""
if safe_input_token_budget <= 0:
raise DirectProtocolError("Safe protocol input token budget must be positive.")
diarization_enabled = transcript.get("speaker_labels_anonymous") is True
if diarization_enabled:
try:
compact = compact_diarized_transcript(transcript.get("segments"))
plain_text = plain_segment_transcript(transcript.get("segments"))
except TranscriptInputError as exc:
raise DirectProtocolError(str(exc)) from exc
instruction = COMPACT_DIARIZED_PROTOCOL_INSTRUCTION
if (
rendered_context
and "Confirmed diarization speaker mappings" in rendered_context
):
instruction = f"{instruction}\n\n{MAPPED_SPEAKER_ATTRIBUTION_INSTRUCTION}"
compact_prompt = build_direct_protocol_prompt(
compact.text,
rendered_context,
instruction=instruction,
meeting_language=meeting_language,
)
compact_estimate = estimate_input_tokens(compact_prompt)
if compact_estimate <= safe_input_token_budget:
return SelectedTranscriptInput(
text=compact.text,
prompt=compact_prompt,
representation="diarized_compact",
estimated_input_tokens=compact_estimate,
safe_input_token_budget=safe_input_token_budget,
fallback_used=False,
diarization_enabled=True,
)
representation = "plain_transcript_fallback"
fallback_used = True
else:
plain_text = transcript.get("text")
if not isinstance(plain_text, str) or not plain_text.strip():
raise DirectProtocolError("Transcript top-level 'text' must be a non-empty string.")
representation = "plain_transcript"
fallback_used = False
plain_prompt = build_direct_protocol_prompt(
plain_text, rendered_context, meeting_language=meeting_language
)
plain_estimate = estimate_input_tokens(plain_prompt)
if plain_estimate > safe_input_token_budget:
raise DirectProtocolError(
"Protocol prompt/input is too large for the configured safe input budget "
f"({plain_estimate} estimated tokens > {safe_input_token_budget}). "
"No LLM request was made; silent truncation is not allowed."
)
return SelectedTranscriptInput(
text=plain_text,
prompt=plain_prompt,
representation=representation,
estimated_input_tokens=plain_estimate,
safe_input_token_budget=safe_input_token_budget,
fallback_used=fallback_used,
diarization_enabled=diarization_enabled,
)
def generate_direct_protocol(
transcript_path: Path,
context_path: Path | None = None,
*,
model: str = DEFAULT_MODEL,
endpoint: str = DEFAULT_ENDPOINT,
timeout: int = DEFAULT_TIMEOUT,
num_ctx: int = DEFAULT_NUM_CTX,
num_predict: int = DEFAULT_NUM_PREDICT,
num_thread: int | None = None,
safe_input_token_budget: int = DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
model_check: Callable[[str, str, int], dict[str, Any]] = require_model,
generation_call: Callable[..., OllamaGeneration] = generate_once,
glossary_aliases: Mapping[str, str] | None = None,
) -> DirectProtocolResult:
transcript = _load_transcript_document(transcript_path)
context: MeetingContext | None = (
load_meeting_context(context_path) if context_path is not None else None
)
rendered_context = render_meeting_context_for_prompt(context) if context else None
# Older runs without a meeting language retain the historical German output.
meeting_language = (
str(context.data["meeting"].get("language") or "de") if context else "de"
)
selected = select_transcript_input(
transcript,
rendered_context,
safe_input_token_budget=safe_input_token_budget,
meeting_language=meeting_language,
)
model_metadata = model_check(endpoint, model, 10)
generation = generation_call(
endpoint,
model,
selected.prompt,
timeout=timeout,
num_ctx=num_ctx,
num_predict=num_predict,
num_thread=num_thread,
)
data = generation.raw_response
runtime_metadata = {
"output_language": meeting_language,
"model": model,
"prompt_token_count": data.get("prompt_eval_count"),
"output_token_count": data.get("eval_count"),
"prompt_evaluation_duration_ns": data.get("prompt_eval_duration"),
"generation_duration_ns": data.get("eval_duration"),
"total_ollama_duration_ns": data.get("total_duration"),
"client_wall_time_seconds": generation.client_wall_time_seconds,
"completion_reason": data.get("done_reason"),
"done": data.get("done"),
"request_count": 1,
"temperature": 0.0,
"think": False,
"num_ctx": num_ctx,
"num_predict": num_predict,
"num_thread": num_thread,
"selected_transcript_representation": selected.representation,
"estimated_input_tokens": selected.estimated_input_tokens,
"safe_input_token_budget": selected.safe_input_token_budget,
"input_token_estimation_method": "utf8_bytes_divided_by_4.4",
"fallback_used": selected.fallback_used,
"diarization_enabled": selected.diarization_enabled,
"speaker_attribution_available": (
True
if selected.representation == "diarized_compact"
else False
if selected.representation == "plain_transcript_fallback"
else None
),
"speaker_attribution_loss_reason": (
"plain_transcript_fallback"
if selected.representation == "plain_transcript_fallback"
else None
),
"speaker_mapping_count": len(context.speaker_mappings) if context else 0,
"glossary_aliases_configured": dict(sorted((glossary_aliases or {}).items())),
"glossary_replacements": [],
}
return DirectProtocolResult(
protocol_text=generation.text,
exact_prompt=selected.prompt,
model_metadata=model_metadata,
runtime_metadata=runtime_metadata,
raw_response=data,
transcript_input=selected.text,
)
+188
View File
@@ -0,0 +1,188 @@
"""Immutable protocol records with one atomic latest-generation publication."""
from __future__ import annotations
import fcntl
import json
import os
import shutil
import subprocess
import tempfile
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from uuid import uuid4
from src.meeting_lab.models.meeting_context import MeetingContext, write_meeting_context
from src.meeting_lab.protocol.generate_direct_protocol import DirectProtocolResult
_DIAGNOSTICS = (
"exact_prompt.txt",
"transcript_input.txt",
"raw_response.json",
"runtime_metadata.json",
)
def _git_provenance(repository: Path) -> dict[str, Any]:
provenance: dict[str, Any] = {"repository": str(repository), "sha": None, "dirty": None}
try:
provenance["sha"] = subprocess.check_output(
["git", "rev-parse", "HEAD"], cwd=repository, text=True, stderr=subprocess.DEVNULL
).strip()
provenance["dirty"] = bool(
subprocess.check_output(
["git", "status", "--porcelain"],
cwd=repository,
text=True,
stderr=subprocess.DEVNULL,
)
)
except (OSError, subprocess.SubprocessError):
pass
return provenance
def _write_json(path: Path, value: Any) -> None:
path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def _replace_link(path: Path, target: str) -> None:
"""Replace a link/file atomically; never write through an existing symlink."""
temporary = path.with_name(f".{path.name}-{uuid4().hex}")
try:
temporary.symlink_to(target)
os.replace(temporary, path)
finally:
temporary.unlink(missing_ok=True)
def _sync_record(directory: Path) -> None:
for path in directory.iterdir():
with path.open("rb") as stream:
os.fsync(stream.fileno())
descriptor = os.open(directory, os.O_RDONLY)
try:
os.fsync(descriptor)
finally:
os.close(descriptor)
def _legacy_generation(run_dir: Path, protocol_dir: Path, generations: Path) -> None:
"""Bootstrap old regular files before switching any compatibility paths."""
if (protocol_dir / "current").is_symlink() or not (run_dir / "protocol.md").is_file():
return
# Preserve the latest regular-file state even for runs created by an older
# history implementation. Never reuse or overwrite an existing record.
index = max(
(int(p.name) for p in generations.iterdir() if p.is_dir() and p.name.isdigit()),
default=0,
) + 1
legacy = generations / f"{index:03d}"
pending = Path(tempfile.mkdtemp(prefix=".legacy-", dir=generations))
try:
shutil.copyfile(run_dir / "protocol.md", pending / "protocol.md")
for name in _DIAGNOSTICS:
source = protocol_dir / name
if source.is_file():
shutil.copyfile(source, pending / name)
context = run_dir / "context/meeting_context.yaml"
if context.is_file():
shutil.copyfile(context, pending / "meeting_context.yaml")
_sync_record(pending)
pending.rename(legacy)
finally:
shutil.rmtree(pending, ignore_errors=True)
_replace_link(protocol_dir / "current", f"generations/{legacy.name}")
def _compatibility_links(run_dir: Path, protocol_dir: Path, pending: Path) -> None:
"""Install indirections while they still resolve to the previous generation."""
links = [(run_dir / "protocol.md", "protocol/current/protocol.md")]
links.extend((protocol_dir / name, f"current/{name}") for name in _DIAGNOSTICS)
if (pending / "meeting_context.yaml").is_file():
context_dir = run_dir / "context"
context_dir.mkdir(exist_ok=True)
links.append(
(context_dir / "meeting_context.yaml", "../protocol/current/meeting_context.yaml")
)
for path, target in links:
if not path.is_symlink() or os.readlink(path) != target:
_replace_link(path, target)
def persist_protocol_generation(
run_dir: Path,
result: DirectProtocolResult,
*,
context: MeetingContext | None,
) -> Path:
"""Publish a complete record; current is the sole successful-generation pointer.
Readers needing a multi-file snapshot should resolve current once. Relative
symlinks keep complete run directories movable. Writes are serialized locally.
"""
protocol_dir = run_dir / "protocol"
protocol_dir.mkdir(exist_ok=True)
generations = protocol_dir / "generations"
generations.mkdir(exist_ok=True)
with (protocol_dir / ".publication.lock").open("a") as lock:
fcntl.flock(lock, fcntl.LOCK_EX)
_legacy_generation(run_dir, protocol_dir, generations)
index = (
max(
(int(p.name) for p in generations.iterdir() if p.is_dir() and p.name.isdigit()),
default=0,
)
+ 1
)
destination = generations / f"{index:03d}"
pending = Path(tempfile.mkdtemp(prefix=".pending-", dir=generations))
published = False
try:
metadata = dict(result.runtime_metadata)
metadata.update(
{
"generation_index": index,
"generation_timestamp": datetime.now(UTC).isoformat(timespec="seconds"),
"source_run_id": run_dir.name,
"speaker_mapping": dict(sorted(context.speaker_mappings.items()))
if context
else {},
"speaker_mapping_names": {
label: context.participant_for_speaker(label).get("display_name", "")
for label in sorted(context.speaker_mappings)
if context.participant_for_speaker(label) is not None
}
if context
else {},
"git": {"meeting_lab": _git_provenance(Path(__file__).resolve().parents[3])},
}
)
(pending / "protocol.md").write_text(result.protocol_text, encoding="utf-8")
(pending / "exact_prompt.txt").write_text(result.exact_prompt, encoding="utf-8")
if result.transcript_input is not None:
(pending / "transcript_input.txt").write_text(
result.transcript_input, encoding="utf-8"
)
_write_json(pending / "raw_response.json", result.raw_response)
_write_json(pending / "runtime_metadata.json", metadata)
_write_json(pending / "model_metadata.json", result.model_metadata)
if context is not None:
write_meeting_context(context, pending / "meeting_context.yaml")
_sync_record(pending)
_compatibility_links(run_dir, protocol_dir, pending)
pending.rename(destination)
_replace_link(protocol_dir / "current", f"generations/{destination.name}")
published = True
finally:
shutil.rmtree(pending, ignore_errors=True)
# An interrupted process may leave an unreferenced complete record;
# never delete the record selected by current, even after interruption.
if (
not published
and destination.exists()
and (protocol_dir / "current").resolve() != destination
):
shutil.rmtree(destination, ignore_errors=True)
return run_dir / "protocol.md"
@@ -0,0 +1,97 @@
"""Deterministic transcript representations for one-call protocol prompts."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
class TranscriptInputError(ValueError):
"""Raised when a transcript cannot be represented without content loss."""
@dataclass(frozen=True)
class SpeakerBlock:
"""One contiguous run of transcript segments assigned to one speaker."""
speaker_id: str
segment_texts: tuple[str, ...]
@dataclass(frozen=True)
class CompactDiarizedTranscript:
"""Compact prompt text plus structural evidence of segment preservation."""
text: str
blocks: tuple[SpeakerBlock, ...]
source_segment_count: int
@property
def represented_segment_count(self) -> int:
return sum(len(block.segment_texts) for block in self.blocks)
@property
def segment_texts(self) -> tuple[str, ...]:
return tuple(text for block in self.blocks for text in block.segment_texts)
def normalize_segment_text(value: Any, index: int) -> str:
"""Normalize formatting whitespace while retaining all semantic text."""
if not isinstance(value, str):
raise TranscriptInputError(f"Transcript segment {index} text must be a string.")
return " ".join(value.split())
def compact_diarized_transcript(segments: Any) -> CompactDiarizedTranscript:
"""Group only adjacent same-speaker segments and omit repeated timestamps."""
if not isinstance(segments, list) or not segments:
raise TranscriptInputError(
"Diarized transcript must contain a non-empty 'segments' list."
)
mutable_blocks: list[tuple[str, list[str]]] = []
source_texts: list[str] = []
for index, segment in enumerate(segments):
if not isinstance(segment, dict):
raise TranscriptInputError(f"Transcript segment {index} must be an object.")
speaker = segment.get("speaker_id") or "SPEAKER_UNASSIGNED"
if not isinstance(speaker, str) or not speaker.startswith("SPEAKER_"):
raise TranscriptInputError(
f"Transcript segment {index} must use an anonymous SPEAKER_ label."
)
text = normalize_segment_text(segment.get("text"), index)
source_texts.append(text)
if mutable_blocks and mutable_blocks[-1][0] == speaker:
mutable_blocks[-1][1].append(text)
else:
mutable_blocks.append((speaker, [text]))
blocks = tuple(
SpeakerBlock(speaker_id=speaker, segment_texts=tuple(texts))
for speaker, texts in mutable_blocks
)
rendered = "\n".join(
f"{block.speaker_id}: {' '.join(block.segment_texts)}" for block in blocks
)
result = CompactDiarizedTranscript(
text=rendered + "\n",
blocks=blocks,
source_segment_count=len(segments),
)
if result.represented_segment_count != len(segments):
raise TranscriptInputError("Compact diarized transcript lost source segments.")
if result.segment_texts != tuple(source_texts):
raise TranscriptInputError("Compact diarized transcript changed segment order or text.")
return result
def plain_segment_transcript(segments: Any) -> str:
"""Reconstruct plain transcript text from every segment in source order."""
if not isinstance(segments, list) or not segments:
raise TranscriptInputError("Transcript must contain a non-empty 'segments' list.")
texts = []
for index, segment in enumerate(segments):
if not isinstance(segment, dict):
raise TranscriptInputError(f"Transcript segment {index} must be an object.")
texts.append(normalize_segment_text(segment.get("text"), index))
return " ".join(texts)
@@ -0,0 +1,5 @@
"""Audio transcription support for the direct-protocol MVP."""
from .whisper import TranscriptionError, TranscriptionResult, transcribe_audio
__all__ = ["TranscriptionError", "TranscriptionResult", "transcribe_audio"]
+290
View File
@@ -0,0 +1,290 @@
"""Isolated whisper.cpp wrapper producing Meeting Lab compact transcripts."""
from __future__ import annotations
import json
import os
import platform
import re
import shutil
import subprocess
import tempfile
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Sequence
BACKEND = "whisper.cpp"
RAW_FILENAME = "whisper_raw.json"
TRANSCRIPT_FILENAME = "transcript.json"
TEXT_FILENAME = "transcript.txt"
METADATA_FILENAME = "runtime_metadata.json"
DEFAULT_THREADS = "auto"
class TranscriptionError(RuntimeError):
"""Raised when parameters, Whisper execution, or output are invalid."""
@dataclass(frozen=True)
class TranscriptionResult:
output_dir: Path
raw_output: Path
transcript_json: Path
transcript_text: Path
runtime_metadata: Path
runtime_seconds: float
def _logical_cpu_count() -> int:
"""Return the available logical CPU count as a last-resort fallback."""
if hasattr(os, "sched_getaffinity"):
try:
count = len(os.sched_getaffinity(0))
if count > 0:
return count
except OSError:
pass
return os.cpu_count() or 1
def _linux_physical_core_count() -> int | None:
affinity = None
if hasattr(os, "sched_getaffinity"):
try:
affinity = os.sched_getaffinity(0)
except OSError:
pass
cores: set[tuple[str, str]] = set()
for cpu_dir in Path("/sys/devices/system/cpu").glob("cpu[0-9]*"):
try:
cpu_number = int(cpu_dir.name[3:])
if affinity is not None and cpu_number not in affinity:
continue
topology = cpu_dir / "topology"
package = (topology / "physical_package_id").read_text().strip()
core = (topology / "core_id").read_text().strip()
cores.add((package, core))
except (OSError, ValueError):
continue
return len(cores) or None
def _darwin_physical_core_count() -> int | None:
try:
completed = subprocess.run(
("sysctl", "-n", "hw.physicalcpu"),
check=False,
capture_output=True,
text=True,
)
count = int(completed.stdout.strip())
return count if completed.returncode == 0 and count > 0 else None
except (OSError, ValueError):
return None
def physical_core_count() -> int:
"""Detect physical cores where supported, falling back to available threads."""
system = platform.system()
detected = _linux_physical_core_count() if system == "Linux" else None
if system == "Darwin":
detected = _darwin_physical_core_count()
return detected or _logical_cpu_count()
def resolve_threads(
threads: str | int,
detector: Callable[[], int] = physical_core_count,
) -> int:
if isinstance(threads, bool):
raise TranscriptionError("Threads must be 'auto' or a positive integer.")
if threads == "auto":
count = detector()
else:
try:
count = int(threads)
except (TypeError, ValueError) as exc:
raise TranscriptionError("Threads must be 'auto' or a positive integer.") from exc
if count <= 0:
raise TranscriptionError("Threads must be 'auto' or a positive integer.")
return count
def _vulkan_support(raw: dict[str, Any]) -> bool | None:
system_info = raw.get("systeminfo")
if not isinstance(system_info, str) or "VULKAN" not in system_info.upper():
return None
return re.search(r"VULKAN\s*=\s*1", system_info, re.IGNORECASE) is not None
def _json_object(path: Path) -> dict[str, Any]:
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
raise TranscriptionError(f"Cannot read Whisper JSON output {path}: {exc}") from exc
if not isinstance(data, dict):
raise TranscriptionError("Whisper JSON output must contain a top-level object.")
return data
def compact_transcript(raw: dict[str, Any]) -> dict[str, Any]:
"""Convert whisper.cpp JSON without linguistic cleanup or reordering."""
entries = raw.get("transcription")
if not isinstance(entries, list):
raise TranscriptionError("Whisper JSON output must contain a 'transcription' list.")
segments: list[dict[str, Any]] = []
for index, entry in enumerate(entries):
if not isinstance(entry, dict):
raise TranscriptionError(f"transcription[{index}] must be an object.")
offsets = entry.get("offsets")
if not isinstance(offsets, dict):
raise TranscriptionError(f"transcription[{index}].offsets must be an object.")
start_ms = offsets.get("from")
end_ms = offsets.get("to")
if not isinstance(start_ms, (int, float)) or isinstance(start_ms, bool):
raise TranscriptionError(f"transcription[{index}].offsets.from must be a number.")
if not isinstance(end_ms, (int, float)) or isinstance(end_ms, bool):
raise TranscriptionError(f"transcription[{index}].offsets.to must be a number.")
if end_ms < start_ms:
raise TranscriptionError(
f"transcription[{index}].offsets.to must be greater than or equal to offsets.from."
)
text_value = entry.get("text", "")
if not isinstance(text_value, str):
raise TranscriptionError(f"transcription[{index}].text must be a string.")
text = text_value.strip()
if text:
segments.append(
{
"id": len(segments),
"start": float(start_ms) / 1000.0,
"end": float(end_ms) / 1000.0,
"text": text,
}
)
return {"text": " ".join(item["text"] for item in segments), "segments": segments}
def _timestamp(seconds: float) -> str:
milliseconds = int(round(seconds * 1000))
hours, remainder = divmod(milliseconds, 3_600_000)
minutes, remainder = divmod(remainder, 60_000)
secs, millis = divmod(remainder, 1000)
return f"{hours:02d}:{minutes:02d}:{secs:02d}.{millis:03d}"
def transcript_text(transcript: dict[str, Any]) -> str:
lines = [
f"[{_timestamp(item['start'])} - {_timestamp(item['end'])}] {item['text']}"
for item in transcript["segments"]
]
return "\n".join(lines) + ("\n" if lines else "")
def _write_json(path: Path, value: dict[str, Any]) -> None:
path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def transcribe_audio(
audio_path: Path,
model_path: Path,
output_dir: Path,
language: str = "auto",
*,
executable: str = "whisper-cli",
threads: str | int = DEFAULT_THREADS,
thread_detector: Callable[[], int] = physical_core_count,
runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
monotonic: Callable[[], float] = time.monotonic,
now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
) -> TranscriptionResult:
"""Run one whisper.cpp call and write raw, compact, text, and metadata outputs."""
audio_path = Path(audio_path)
model_path = Path(model_path)
output_dir = Path(output_dir)
if not audio_path.is_file():
raise TranscriptionError(f"Audio file does not exist: {audio_path}")
if not model_path.is_file():
raise TranscriptionError(f"Whisper model does not exist: {model_path}")
if not isinstance(language, str) or not language.strip():
raise TranscriptionError("Language must be a non-empty string.")
if not executable.strip():
raise TranscriptionError("Whisper executable must be a non-empty string.")
if output_dir.exists() and not output_dir.is_dir():
raise TranscriptionError(f"Output directory path is not a directory: {output_dir}")
thread_count = resolve_threads(threads, thread_detector)
output_dir.mkdir(parents=True, exist_ok=True)
raw_output = output_dir / RAW_FILENAME
started_at = now().astimezone(timezone.utc)
started = monotonic()
with tempfile.TemporaryDirectory(prefix=".whisper-", dir=output_dir) as temp_name:
temporary_prefix = Path(temp_name) / "whisper_raw"
command: Sequence[str] = (
executable,
"-m", str(model_path),
"-f", str(audio_path),
"-l", language.strip(),
"-t", str(thread_count),
"-fa",
"-oj",
"-of", str(temporary_prefix),
)
try:
completed = runner(command, check=False, capture_output=True, text=True)
except OSError as exc:
raise TranscriptionError(f"Could not start {BACKEND}: {exc}") from exc
runtime_seconds = monotonic() - started
temporary_raw = temporary_prefix.with_suffix(".json")
if temporary_raw.is_file():
shutil.copyfile(temporary_raw, raw_output)
if completed.returncode != 0:
detail = completed.stderr.strip() or completed.stdout.strip() or "no diagnostic output"
raise TranscriptionError(
f"{BACKEND} failed with exit code {completed.returncode}: {detail}"
)
if not raw_output.is_file():
raise TranscriptionError(f"{BACKEND} completed without producing JSON output.")
raw_data = _json_object(raw_output)
transcript = compact_transcript(raw_data)
transcript_json_path = output_dir / TRANSCRIPT_FILENAME
transcript_text_path = output_dir / TEXT_FILENAME
metadata_path = output_dir / METADATA_FILENAME
_write_json(transcript_json_path, transcript)
transcript_text_path.write_text(transcript_text(transcript), encoding="utf-8")
duration = max((item["end"] for item in transcript["segments"]), default=None)
metadata = {
"input_file": str(audio_path.resolve()),
"model": str(model_path.resolve()),
"backend": BACKEND,
"whisper_executable": executable,
"language": language.strip(),
"threads": thread_count,
"threads_option": str(threads),
"flash_attention": True,
"vulkan_support_detected": _vulkan_support(raw_data),
"duration_seconds": duration,
"runtime_seconds": runtime_seconds,
"timestamp": started_at.isoformat(),
"output_files": {
"whisper_raw": RAW_FILENAME,
"transcript_json": TRANSCRIPT_FILENAME,
"transcript_text": TEXT_FILENAME,
"runtime_metadata": METADATA_FILENAME,
},
}
_write_json(metadata_path, metadata)
return TranscriptionResult(
output_dir=output_dir,
raw_output=raw_output,
transcript_json=transcript_json_path,
transcript_text=transcript_text_path,
runtime_metadata=metadata_path,
runtime_seconds=runtime_seconds,
)
+230
View File
@@ -0,0 +1,230 @@
import subprocess
import tempfile
import unittest
import wave
from pathlib import Path
from unittest.mock import patch
from src.meeting_lab.audio.preparation import (
DEFAULT_NORMALIZATION_FILTER,
DEFAULT_NORMALIZATION_METHOD,
AudioPreparationError,
prepare_audio,
)
def _write_wav(
path: Path, *, channels: int = 1, sample_rate: int = 16_000, sample_width: int = 2
) -> None:
with wave.open(str(path), "wb") as recording:
recording.setnchannels(channels)
recording.setsampwidth(sample_width)
recording.setframerate(sample_rate)
recording.writeframes(b"\x00" * channels * sample_width * 32)
def _successful_runner(commands: list[list[str]]):
def run(command, **kwargs):
commands.append(list(command))
_write_wav(Path(command[-1]))
return subprocess.CompletedProcess(command, 0, "", "")
return run
class AudioPreparationTests(unittest.TestCase):
def test_supported_inputs_are_prepared_with_normalization_on_and_off(self) -> None:
for suffix in (".wav", ".flac", ".m4a"):
for normalization_enabled in (True, False):
with (
self.subTest(
suffix=suffix, normalization_enabled=normalization_enabled
),
tempfile.TemporaryDirectory() as directory,
):
root = Path(directory)
source = root / f"meeting{suffix}"
if suffix == ".wav":
_write_wav(source)
else:
source.write_bytes(b"original encoded audio")
original = source.read_bytes()
destination = root / "run" / "audio" / "prepared.wav"
commands: list[list[str]] = []
with patch(
"src.meeting_lab.audio.preparation.shutil.which",
return_value="/usr/bin/ffmpeg",
):
result = prepare_audio(
source,
destination,
normalization_enabled=normalization_enabled,
runner=_successful_runner(commands),
)
self.assertEqual(source.read_bytes(), original)
self.assertEqual(result.prepared_path, destination)
with wave.open(str(destination), "rb") as recording:
self.assertEqual(recording.getnchannels(), 1)
self.assertEqual(recording.getframerate(), 16_000)
self.assertEqual(recording.getsampwidth(), 2)
self.assertEqual(recording.getcomptype(), "NONE")
self.assertEqual(commands[0][commands[0].index("-ac") + 1], "1")
self.assertEqual(commands[0][commands[0].index("-ar") + 1], "16000")
self.assertEqual(
commands[0][commands[0].index("-c:a") + 1], "pcm_s16le"
)
self.assertEqual("-af" in commands[0], normalization_enabled)
if normalization_enabled:
self.assertEqual(
commands[0][commands[0].index("-af") + 1],
DEFAULT_NORMALIZATION_FILTER,
)
self.assertEqual(
result.normalization_enabled, normalization_enabled
)
def test_normalization_defaults_to_on_and_explicit_on_matches(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / "meeting.wav"
_write_wav(source)
commands: list[list[str]] = []
with patch(
"src.meeting_lab.audio.preparation.shutil.which",
return_value="/usr/bin/ffmpeg",
):
default = prepare_audio(
source, root / "default.wav", runner=_successful_runner(commands)
)
explicit = prepare_audio(
source,
root / "explicit.wav",
normalization_enabled=True,
runner=_successful_runner(commands),
)
self.assertTrue(default.normalization_enabled)
self.assertTrue(explicit.normalization_enabled)
self.assertEqual(
commands[0][commands[0].index("-af") + 1],
commands[1][commands[1].index("-af") + 1],
)
def test_noncanonical_wav_is_normalized(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / "stereo-48k.wav"
_write_wav(source, channels=2, sample_rate=48_000)
destination = root / "prepared.wav"
with patch(
"src.meeting_lab.audio.preparation.shutil.which",
return_value="/usr/bin/ffmpeg",
):
prepare_audio(source, destination, runner=_successful_runner([]))
with wave.open(str(destination), "rb") as recording:
self.assertEqual(
(recording.getnchannels(), recording.getframerate()), (1, 16_000)
)
def test_ffmpeg_missing_has_actionable_error(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / "meeting.flac"
source.write_bytes(b"audio")
with (
patch(
"src.meeting_lab.audio.preparation.shutil.which", return_value=None
),
self.assertRaisesRegex(AudioPreparationError, "not found on PATH"),
):
prepare_audio(source, root / "prepared.wav")
def test_ffmpeg_failure_includes_diagnostic_and_preserves_source(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / "meeting.m4a"
source.write_bytes(b"original")
def fail(command, **kwargs):
return subprocess.CompletedProcess(command, 1, "", "decoder exploded")
with (
patch(
"src.meeting_lab.audio.preparation.shutil.which",
return_value="/usr/bin/ffmpeg",
),
self.assertRaisesRegex(AudioPreparationError, "decoder exploded"),
):
prepare_audio(source, root / "prepared.wav", runner=fail)
self.assertEqual(source.read_bytes(), b"original")
self.assertFalse((root / "prepared.wav").exists())
def test_prepared_audio_metadata_is_traceable(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / "unknown_meeting.flac"
source.write_bytes(b"source")
destination = root / "audio" / "prepared.wav"
with patch(
"src.meeting_lab.audio.preparation.shutil.which",
return_value="/usr/bin/ffmpeg",
):
result = prepare_audio(
source, destination, runner=_successful_runner([])
)
metadata = result.metadata()
self.assertEqual(metadata["original_source_name"], "unknown_meeting.flac")
self.assertEqual(metadata["original_format"], "flac")
self.assertEqual(
metadata["prepared_audio_path"], str(destination.resolve())
)
self.assertEqual(metadata["preparation_method"], "ffmpeg")
self.assertTrue(metadata["normalization_enabled"])
self.assertEqual(
metadata["normalization_method"], DEFAULT_NORMALIZATION_METHOD
)
self.assertEqual(
metadata["normalization_filter"], DEFAULT_NORMALIZATION_FILTER
)
self.assertEqual(
metadata["canonical_output"],
{
"container": "wav",
"codec": "pcm_s16le",
"channels": 1,
"sample_rate_hz": 16_000,
"bits_per_sample": 16,
},
)
def test_disabled_normalization_metadata_has_no_method_or_filter(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / "meeting.m4a"
source.write_bytes(b"source")
with patch(
"src.meeting_lab.audio.preparation.shutil.which",
return_value="/usr/bin/ffmpeg",
):
result = prepare_audio(
source,
root / "prepared.wav",
normalization_enabled=False,
runner=_successful_runner([]),
)
metadata = result.metadata()
self.assertFalse(metadata["normalization_enabled"])
self.assertIsNone(metadata["normalization_method"])
self.assertIsNone(metadata["normalization_filter"])
if __name__ == "__main__":
unittest.main()
+222
View File
@@ -0,0 +1,222 @@
import json
import os
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
from src.meeting_lab.diarization.alignment import align_transcript, write_diarized_transcript
from src.meeting_lab.diarization.backend import (
DiarizationError,
diarize_audio,
run_container_pyannote,
select_device,
)
class FakeCuda:
def __init__(self, available: bool, *, name: str = "Test GPU", failure=None):
self.available = available
self.name = name
self.failure = failure
def is_available(self):
return self.available
def get_device_name(self, index):
if self.failure:
raise self.failure
return self.name
class FakeTorch:
def __init__(self, available: bool, *, failure=None):
self.cuda = FakeCuda(available, failure=failure)
self.probes = []
def device(self, name):
return name
def zeros(self, size, *, device):
self.probes.append(device)
if self.cuda.failure:
raise self.cuda.failure
return [0]
class DeviceSelectionTests(unittest.TestCase):
def test_auto_selects_usable_gpu(self):
torch = FakeTorch(True)
self.assertEqual(select_device("auto", torch), ("cuda", "Test GPU"))
self.assertEqual(torch.probes, ["cuda"])
def test_auto_falls_back_to_cpu(self):
self.assertEqual(select_device("auto", FakeTorch(False)), ("cpu", None))
self.assertEqual(
select_device("auto", FakeTorch(True, failure=RuntimeError("probe"))),
("cpu", None),
)
def test_explicit_cpu_does_not_probe_gpu(self):
torch = FakeTorch(True)
self.assertEqual(select_device("cpu", torch), ("cpu", None))
self.assertEqual(torch.probes, [])
def test_explicit_gpu_fails_when_unavailable(self):
with self.assertRaisesRegex(DiarizationError, "GPU is unavailable"):
select_device("gpu", FakeTorch(False))
class AlignmentTests(unittest.TestCase):
def test_exclusive_overlap_assigns_anonymous_speakers(self):
transcript = {
"text": "Original unchanged text.",
"segments": [
{"id": 0, "start": 0.0, "end": 4.0, "text": "Hallo"},
{"id": 1, "start": 4.0, "end": 6.0, "text": "Antwort"},
],
}
original = json.loads(json.dumps(transcript))
turns = [
{"start": 0.0, "end": 3.0, "speaker_id": "SPEAKER_00"},
{"start": 3.0, "end": 6.0, "speaker_id": "SPEAKER_01"},
]
derived = align_transcript(transcript, turns)
self.assertEqual(transcript, original)
self.assertEqual(derived["segments"][0]["speaker_id"], "SPEAKER_00")
self.assertEqual(derived["segments"][0]["speaker_overlap_seconds"], 3.0)
self.assertEqual(derived["segments"][1]["speaker_id"], "SPEAKER_01")
self.assertIn("SPEAKER_00: Hallo", derived["text"])
self.assertTrue(derived["speaker_labels_anonymous"])
self.assertEqual(derived["alignment_source"], "exclusive_diarization")
def test_speaker_aware_transcript_is_a_separate_artifact(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
source = root / "transcript.json"
turns = root / "exclusive_turns.json"
source_text = json.dumps(
{"text": "Original", "segments": [{"start": 0, "end": 1, "text": "Hi"}]}
)
source.write_text(source_text, encoding="utf-8")
turns.write_text(
json.dumps([{"start": 0, "end": 1, "speaker_id": "SPEAKER_07"}]),
encoding="utf-8",
)
json_path, text_path = write_diarized_transcript(source, turns, root / "derived")
self.assertEqual(source.read_text(encoding="utf-8"), source_text)
self.assertNotEqual(json_path, source)
self.assertIn("SPEAKER_07", json_path.read_text(encoding="utf-8"))
self.assertIn("SPEAKER_07", text_path.read_text(encoding="utf-8"))
class ContainerAdapterTests(unittest.TestCase):
def test_container_configuration_and_metadata_do_not_persist_token(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
audio = root / "audio.wav"
output = root / "output"
audio.write_bytes(b"audio")
observed = {}
def fake_runner(command, **kwargs):
observed["command"] = command
output.mkdir(exist_ok=True)
metadata = {
"backend": "pyannote.audio",
"model": "pyannote/speaker-diarization-community-1",
"requested_device_mode": "gpu",
"actual_device": "cuda",
"credentials_persisted": False,
}
(output / "metadata.json").write_text(json.dumps(metadata))
return SimpleNamespace(returncode=0, stdout="ok", stderr="")
with patch.dict("os.environ", {"HF_TOKEN": "secret-token"}):
result = run_container_pyannote(
audio,
output,
"gpu",
image="test/image",
container_args=("--device=/dev/test",),
runner=fake_runner,
uid_getter=lambda: 2345,
gid_getter=lambda: 3456,
)
command = observed["command"]
shell_command = command[-1]
persisted = "".join(
path.read_text(encoding="utf-8")
for path in output.iterdir()
if path.is_file()
)
self.assertNotIn("secret-token", persisted)
self.assertNotIn("secret-token", command)
self.assertIn("HF_TOKEN", command)
self.assertIn("chown -R 2345:3456 /output", shell_command)
self.assertIn("chmod -R u+rwX /output", shell_command)
self.assertNotIn("1000:1000", shell_command)
device_index = command.index("--device=/dev/test")
self.assertLess(device_index, command.index("test/image"))
self.assertFalse(result.metadata["credentials_persisted"])
self.assertEqual(result.metadata["runtime_adapter"], "container")
self.assertTrue(
all(os.access(path, os.W_OK) for path in (output, *output.rglob("*")))
)
def test_unwritable_container_artifact_is_rejected_before_metadata_update(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
audio = root / "audio.wav"
output = root / "output"
audio.write_bytes(b"audio")
def fake_runner(command, **kwargs):
output.mkdir(exist_ok=True)
metadata = output / "metadata.json"
metadata.write_text("{}", encoding="utf-8")
metadata.chmod(0o444)
return SimpleNamespace(returncode=0, stdout="", stderr="")
with patch(
"src.meeting_lab.diarization.backend.os.access",
side_effect=lambda path, mode: Path(path).name != "metadata.json",
):
with self.assertRaisesRegex(DiarizationError, "not writable"):
run_container_pyannote(
audio,
output,
"cpu",
image="test/image",
runner=fake_runner,
)
def test_orchestrator_dispatches_runtime(self):
with patch(
"src.meeting_lab.diarization.backend.run_container_pyannote"
) as container:
diarize_audio(
Path("audio.wav"),
Path("out"),
"cpu",
runtime="container",
container_image="image",
container_args=("--arg",),
)
container.assert_called_once_with(
Path("audio.wav"),
Path("out"),
"cpu",
image="image",
container_args=("--arg",),
)
if __name__ == "__main__":
unittest.main()
+341
View File
@@ -0,0 +1,341 @@
import json
import tempfile
import unittest
from datetime import datetime
from pathlib import Path
from unittest.mock import Mock, patch
import requests
from scripts import run_direct_protocol
from src.meeting_lab.llm import ollama
from src.meeting_lab.llm.ollama import OllamaError, OllamaGeneration
from src.meeting_lab.protocol.generate_direct_protocol import (
DirectProtocolError,
generate_direct_protocol,
load_compact_transcript,
)
VALID_CONTEXT = """schema_version: "1"
meeting:
meeting_id: "test-meeting"
title: "Test Meeting"
language: "de"
participants: []
mentioned_people: []
organization:
departments: []
known_entities: {}
"""
def write_transcript(path: Path, text: str = "Wir besprechen den Projektstatus.") -> None:
path.write_text(json.dumps({"text": text, "segments": []}), encoding="utf-8")
def generation(text: str = "# Meeting Protocol\n\n## Status\nUnveraendert.") -> OllamaGeneration:
return OllamaGeneration(
raw_response={
"response": text,
"done": True,
"done_reason": "stop",
"prompt_eval_count": 123,
"eval_count": 17,
"prompt_eval_duration": 1000,
"eval_duration": 2000,
"total_duration": 4000,
},
text=text,
client_wall_time_seconds=0.25,
)
class TranscriptLoadingTests(unittest.TestCase):
def test_valid_transcript_is_accepted(self) -> None:
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "transcript.json"
write_transcript(path)
self.assertEqual(load_compact_transcript(path), "Wir besprechen den Projektstatus.")
def test_missing_top_level_text_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "transcript.json"
path.write_text('{"segments": []}', encoding="utf-8")
with self.assertRaisesRegex(DirectProtocolError, "top-level 'text'"):
load_compact_transcript(path)
def test_empty_transcript_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "transcript.json"
write_transcript(path, " \n")
with self.assertRaisesRegex(DirectProtocolError, "non-empty string"):
load_compact_transcript(path)
def test_malformed_json_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "transcript.json"
path.write_text("{", encoding="utf-8")
with self.assertRaisesRegex(DirectProtocolError, "not valid JSON"):
load_compact_transcript(path)
class GeneratorTests(unittest.TestCase):
def test_prompt_requires_contextual_discussion_density_without_transcript_replay(self) -> None:
with tempfile.TemporaryDirectory() as directory:
transcript = Path(directory) / "transcript.json"
write_transcript(transcript)
result = generate_direct_protocol(
transcript,
model_check=Mock(return_value={}),
generation_call=Mock(return_value=generation()),
)
self.assertIn("vollständiges, strukturiertes", result.exact_prompt)
self.assertIn("relevante Diskussionsverläufe", result.exact_prompt)
self.assertIn("unterschiedliche Positionen", result.exact_prompt)
self.assertIn("Entscheidungsgrundlagen", result.exact_prompt)
self.assertIn("nicht am Meeting teilgenommen haben", result.exact_prompt)
self.assertIn("keine reine Wiedergabe des Transkripts", result.exact_prompt)
self.assertIn("nicht unnötig durch Wiederholungen", result.exact_prompt)
def test_optional_context_absent_and_generation_called_once(self) -> None:
with tempfile.TemporaryDirectory() as directory:
transcript = Path(directory) / "transcript.json"
write_transcript(transcript)
check = Mock(return_value={"model": "qwen3.6:35B-A3B"})
call = Mock(return_value=generation())
result = generate_direct_protocol(
transcript,
model_check=check,
generation_call=call,
)
self.assertIn("Kein Meeting-Kontext", result.exact_prompt)
self.assertEqual(check.call_count, 1)
self.assertEqual(call.call_count, 1)
self.assertEqual(call.call_args.args[1], "qwen3.6:35B-A3B")
self.assertEqual(call.call_args.kwargs["num_ctx"], 32768)
self.assertEqual(result.runtime_metadata["request_count"], 1)
self.assertEqual(result.runtime_metadata["prompt_token_count"], 123)
self.assertFalse(result.runtime_metadata["think"])
self.assertEqual(result.runtime_metadata["temperature"], 0.0)
def test_valid_context_is_loaded_and_rendered(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
transcript = root / "transcript.json"
context = root / "context.yaml"
write_transcript(transcript)
context.write_text(VALID_CONTEXT, encoding="utf-8")
result = generate_direct_protocol(
transcript,
context,
model_check=Mock(return_value={}),
generation_call=Mock(return_value=generation()),
)
self.assertIn("MEETING CONTEXT V1", result.exact_prompt)
self.assertIn("Test Meeting", result.exact_prompt)
def test_invalid_context_is_rejected_before_network_calls(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
transcript = root / "transcript.json"
context = root / "context.yaml"
write_transcript(transcript)
context.write_text("schema_version: wrong", encoding="utf-8")
check = Mock()
call = Mock()
with self.assertRaisesRegex(ValueError, "schema_version"):
generate_direct_protocol(
transcript,
context,
model_check=check,
generation_call=call,
)
check.assert_not_called()
call.assert_not_called()
class OllamaTests(unittest.TestCase):
def test_unavailable_endpoint_failure(self) -> None:
with patch.object(ollama.requests, "get", side_effect=requests.ConnectionError("down")):
with self.assertRaisesRegex(OllamaError, "not reachable"):
ollama.require_model("http://127.0.0.1:11434", "model")
def test_missing_model_failure(self) -> None:
response = Mock()
response.raise_for_status.return_value = None
response.json.return_value = {"models": [{"name": "other:model"}]}
with patch.object(ollama.requests, "get", return_value=response):
with self.assertRaisesRegex(OllamaError, "not installed"):
ollama.require_model("http://127.0.0.1:11434", "model")
def test_request_settings_and_raw_response(self) -> None:
raw = {"response": "# Meeting Protocol", "done": True}
response = Mock()
response.raise_for_status.return_value = None
response.json.return_value = raw
with patch.object(ollama.requests, "post", return_value=response) as post:
result = ollama.generate_once(
"http://localhost:11434",
"qwen3.8:27b",
"prompt",
timeout=30,
num_ctx=32768,
num_predict=8192,
)
self.assertEqual(post.call_count, 1)
payload = post.call_args.kwargs["json"]
self.assertEqual(payload["model"], "qwen3.8:27b")
self.assertEqual(payload["options"]["temperature"], 0.0)
self.assertEqual(payload["options"]["num_ctx"], 32768)
self.assertFalse(payload["think"])
self.assertFalse(payload["stream"])
self.assertEqual(result.raw_response, raw)
def test_request_omits_num_thread_when_backend_selection_is_requested(self) -> None:
response = Mock()
response.raise_for_status.return_value = None
response.json.return_value = {"response": "# Meeting Protocol"}
with patch.object(ollama.requests, "post", return_value=response) as post:
ollama.generate_once(
"http://localhost:11434",
"qwen3.8:27b",
"prompt",
timeout=30,
num_ctx=32768,
num_predict=8192,
)
payload = post.call_args.kwargs["json"]
self.assertNotIn("num_thread", payload["options"])
def test_malformed_response_failure_without_retry(self) -> None:
response = Mock()
response.raise_for_status.return_value = None
response.json.return_value = {"message": "missing response"}
with patch.object(ollama.requests, "post", return_value=response) as post:
with self.assertRaisesRegex(OllamaError, "no string 'response'"):
ollama.generate_once("url", "model", "prompt", timeout=1, num_ctx=1, num_predict=1)
self.assertEqual(post.call_count, 1)
def test_empty_response_failure(self) -> None:
response = Mock()
response.raise_for_status.return_value = None
response.json.return_value = {"response": " "}
with patch.object(ollama.requests, "post", return_value=response):
with self.assertRaisesRegex(OllamaError, "empty protocol"):
ollama.generate_once("url", "model", "prompt", timeout=1, num_ctx=1, num_predict=1)
class DirectProtocolCliTests(unittest.TestCase):
def test_artifacts_are_preserved_and_protocol_is_untouched(self) -> None:
protocol_text = "# Meeting Protocol\n\nExact output. \n"
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
transcript = root / "source.json"
context = root / "source.yaml"
write_transcript(transcript)
context.write_text(VALID_CONTEXT, encoding="utf-8")
args = run_direct_protocol.parse_args(
[str(transcript), "--context", str(context), "--output-root", str(root / "runs")]
)
with patch.object(
run_direct_protocol,
"generate_direct_protocol",
return_value=type("Result", (), {
"protocol_text": protocol_text,
"exact_prompt": "exact prompt\n",
"transcript_input": "selected transcript\n",
"raw_response": {"response": protocol_text},
"runtime_metadata": {"request_count": 1},
"model_metadata": {},
})(),
) as generator:
code, run_dir, protocol_path = run_direct_protocol.run(args)
self.assertEqual(code, 0)
self.assertEqual(generator.call_count, 1)
self.assertEqual(protocol_path.read_text(encoding="utf-8"), protocol_text)
self.assertEqual(
(run_dir / "protocol/exact_prompt.txt").read_text(encoding="utf-8"),
"exact prompt\n",
)
self.assertEqual(
(run_dir / "protocol/transcript_input.txt").read_text(encoding="utf-8"),
"selected transcript\n",
)
self.assertEqual(
json.loads((run_dir / "protocol/raw_response.json").read_text())["response"],
protocol_text,
)
self.assertEqual(
json.loads((run_dir / "protocol/runtime_metadata.json").read_text())["request_count"],
1,
)
self.assertTrue((run_dir / "transcript/transcript.json").is_file())
self.assertTrue((run_dir / "context/meeting_context.yaml").is_file())
self.assertTrue((run_dir / "input_manifest.json").is_file())
self.assertEqual(json.loads((run_dir / "run_metadata.json").read_text())["status"], "completed")
def test_unique_run_directories_do_not_overwrite(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
fixed = datetime(2026, 8, 20, 12, 0, 0)
first = run_direct_protocol.create_unique_run_dir(root, "meeting", lambda: fixed)
marker = first / "keep.txt"
marker.write_text("keep", encoding="utf-8")
second = run_direct_protocol.create_unique_run_dir(root, "meeting", lambda: fixed)
self.assertEqual(first.name, "meeting_20260820_120000")
self.assertEqual(second.name, "meeting_20260820_120000_01")
self.assertEqual(marker.read_text(encoding="utf-8"), "keep")
def test_failure_after_directory_creation_preserves_metadata(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
args = run_direct_protocol.parse_args(
[str(root / "missing.json"), "--output-root", str(root / "runs")]
)
code, run_dir, protocol_path = run_direct_protocol.run(args)
metadata = json.loads((run_dir / "run_metadata.json").read_text())
self.assertEqual(code, 2)
self.assertIsNone(protocol_path)
self.assertEqual(metadata["status"], "failed")
self.assertIn("does not exist", metadata["failure"])
def test_semantic_pipeline_functions_are_never_invoked(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
transcript = root / "source.json"
write_transcript(transcript)
args = run_direct_protocol.parse_args(
[str(transcript), "--output-root", str(root / "runs")]
)
fake_result = type("Result", (), {
"protocol_text": "# Meeting Protocol",
"exact_prompt": "prompt",
"raw_response": {"response": "# Meeting Protocol"},
"runtime_metadata": {},
"model_metadata": {},
"transcript_input": None,
})()
with (
patch("src.meeting_lab.extraction.extract_chunks.extract_input") as extraction,
patch("src.meeting_lab.consolidation.consolidate_facts.call_ollama") as consolidation,
patch.object(run_direct_protocol, "generate_direct_protocol", return_value=fake_result),
):
code, _run_dir, _protocol_path = run_direct_protocol.run(args)
self.assertEqual(code, 0)
extraction.assert_not_called()
consolidation.assert_not_called()
if __name__ == "__main__":
unittest.main()
+79
View File
@@ -11,8 +11,10 @@ from src.meeting_lab.extraction.extract_chunks import (
) )
from src.meeting_lab.models.meeting_context import ( from src.meeting_lab.models.meeting_context import (
MeetingContextValidationError, MeetingContextValidationError,
create_meeting_context,
load_meeting_context, load_meeting_context,
render_meeting_context_for_prompt, render_meeting_context_for_prompt,
serialize_meeting_context_yaml,
validate_meeting_context, validate_meeting_context,
) )
@@ -70,6 +72,38 @@ class MeetingContextTests(unittest.TestCase):
with self.assertRaisesRegex(MeetingContextValidationError, "invalid value"): with self.assertRaisesRegex(MeetingContextValidationError, "invalid value"):
validate_meeting_context(data) validate_meeting_context(data)
def test_missing_participant_attendance_defaults_to_present(self) -> None:
data = copy.deepcopy(self.context.data)
del data["participants"][0]["attendance_status"]
context = create_meeting_context(data)
self.assertEqual(
context.data["participants"][0]["attendance_status"], "present"
)
def test_explicit_mentioned_only_is_preserved(self) -> None:
data = copy.deepcopy(self.context.data)
data["mentioned_people"][0]["attendance_status"] = "mentioned_only"
context = create_meeting_context(data)
self.assertEqual(
context.data["mentioned_people"][0]["attendance_status"],
"mentioned_only",
)
self.assertIn(
"Mentioned but absent people:", render_meeting_context_for_prompt(context)
)
def test_mentioned_only_person_cannot_be_a_diarized_speaker(self) -> None:
data = copy.deepcopy(self.context.data)
mentioned_id = data["mentioned_people"][0]["person_id"]
data["speaker_mappings"] = {"SPEAKER_00": mentioned_id}
with self.assertRaisesRegex(MeetingContextValidationError, "unknown participant"):
validate_meeting_context(data)
def test_prompt_representation_is_deterministic(self) -> None: def test_prompt_representation_is_deterministic(self) -> None:
first = render_meeting_context_for_prompt(self.context) first = render_meeting_context_for_prompt(self.context)
second = render_meeting_context_for_prompt(self.context) second = render_meeting_context_for_prompt(self.context)
@@ -192,6 +226,51 @@ class MeetingContextTests(unittest.TestCase):
self.assertNotIn("responsible: Björn", prompt_context) self.assertNotIn("responsible: Björn", prompt_context)
self.assertNotIn("responsible: Jovana", prompt_context) self.assertNotIn("responsible: Jovana", prompt_context)
def test_existing_context_without_speaker_mappings_remains_valid(self) -> None:
self.assertEqual(self.context.speaker_mappings, {})
self.assertIsNone(self.context.participant_for_speaker("SPEAKER_00"))
def test_explicit_speaker_mapping_is_authoritative(self) -> None:
data = copy.deepcopy(self.context.data)
participant = data["participants"][0]
data["speaker_mappings"] = {"SPEAKER_03": participant["participant_id"]}
context = create_meeting_context(data)
rendered = render_meeting_context_for_prompt(context)
self.assertEqual(
context.participant_for_speaker("SPEAKER_03")["participant_id"],
participant["participant_id"],
)
self.assertIsNone(context.participant_for_speaker("SPEAKER_04"))
self.assertIn("Confirmed diarization speaker mappings (authoritative)", rendered)
self.assertIn("Unmapped SPEAKER_XX labels must remain anonymous", rendered)
def test_speaker_mapping_must_reference_existing_participant(self) -> None:
data = copy.deepcopy(self.context.data)
data["speaker_mappings"] = {"SPEAKER_00": "unknown-person"}
with self.assertRaisesRegex(MeetingContextValidationError, "unknown participant"):
validate_meeting_context(data)
def test_speaker_mapping_label_must_use_pyannote_shape(self) -> None:
data = copy.deepcopy(self.context.data)
data["speaker_mappings"] = {
"Martin": data["participants"][0]["participant_id"]
}
with self.assertRaisesRegex(MeetingContextValidationError, "speaker label"):
validate_meeting_context(data)
def test_generated_context_yaml_is_deterministic_and_round_trips(self) -> None:
first = serialize_meeting_context_yaml(self.context)
second = serialize_meeting_context_yaml(self.context)
self.assertEqual(first, second)
SCRATCH_DIR.mkdir(exist_ok=True)
path = SCRATCH_DIR / "generated_context.yaml"
path.write_text(first, encoding="utf-8")
loaded = load_meeting_context(path)
self.assertEqual(loaded.data, self.context.data)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+164
View File
@@ -0,0 +1,164 @@
"""Meeting-language regression tests without media or model execution."""
import json
import tempfile
import unittest
from functools import partial
from pathlib import Path
from unittest.mock import Mock, patch
from src.meeting_lab.llm.ollama import OllamaGeneration
from src.meeting_lab.models.meeting_context import load_meeting_context
from src.meeting_lab.orchestration.mvp import regenerate_mvp_protocol
from src.meeting_lab.protocol.direct_protocol_prompt import build_direct_protocol_prompt
from src.meeting_lab.protocol.generate_direct_protocol import (
estimate_input_tokens,
generate_direct_protocol,
select_transcript_input,
)
class MeetingLanguageTests(unittest.TestCase):
def test_diarized_plain_fallback_retains_english_instruction(self) -> None:
segments = [
{
"start": index,
"end": index + 1,
"text": "Word.",
"speaker_id": f"SPEAKER_{index % 2:02d}",
}
for index in range(200)
]
plain = " ".join(segment["text"] for segment in segments)
budget = estimate_input_tokens(
build_direct_protocol_prompt(plain, meeting_language="en")
)
selected = select_transcript_input(
{"text": plain, "segments": segments, "speaker_labels_anonymous": True},
None,
meeting_language="en",
safe_input_token_budget=budget,
)
self.assertEqual(selected.representation, "plain_transcript_fallback")
self.assertIn("Write the meeting protocol in English.", selected.prompt)
self.assertEqual(selected.text, plain)
def test_generation_and_regeneration_preserve_language_and_inputs(self) -> None:
for language, expected in (
("de", "German"),
("en", "English"),
(None, "German"),
):
for diarized in (False, True):
with (
self.subTest(language=language, diarized=diarized),
tempfile.TemporaryDirectory() as directory,
):
run = Path(directory)
context_path = run / "context" / "meeting_context.yaml"
context_path.parent.mkdir()
context_text = (
'schema_version: "1"\nmeeting:\n'
" meeting_id: test\n title: Mixed terminology\n"
+ (f" language: {language}\n" if language else "")
+ ' notes: "Freigabe für Product X"\n'
"participants:\n - participant_id: person\n"
' display_name: "Jörg Müller"\n'
"speaker_mappings:\n SPEAKER_00: person\n"
)
context_path.write_text(context_text, encoding="utf-8")
transcript = run / ("diarization" if diarized else "transcript")
transcript.mkdir()
transcript /= (
"transcript_diarized.json" if diarized else "transcript.json"
)
transcript.write_text(
json.dumps(
{
"text": "I will check Product X.",
"speaker_labels_anonymous": diarized,
"segments": [
{
"id": 0,
"start": 0,
"end": 1,
"speaker_id": "SPEAKER_00",
"text": "I will check Product X.",
}
],
}
),
encoding="utf-8",
)
original = transcript.read_bytes()
call = Mock(
return_value=OllamaGeneration(
text="# Meeting Protocol",
raw_response={},
client_wall_time_seconds=0.1,
)
)
generate = partial(
generate_direct_protocol,
model_check=Mock(return_value={}),
generation_call=call,
)
result = generate(transcript, context_path)
self.assertIn(
f"Write the meeting protocol in {expected}.",
result.exact_prompt,
)
self.assertNotIn("in deutscher Sprache", result.exact_prompt)
self.assertNotIn("auf Deutsch", result.exact_prompt)
self.assertIn("Jörg Müller", result.exact_prompt)
self.assertIn("Mixed terminology", result.exact_prompt)
self.assertIn("I will check Product X.", result.transcript_input)
self.assertEqual(
result.runtime_metadata["output_language"], language or "de"
)
self.assertEqual(
context_path.read_text(encoding="utf-8"), context_text
)
context = load_meeting_context(context_path)
with (
patch(
"src.meeting_lab.orchestration.mvp.generate_direct_protocol",
side_effect=generate,
),
patch(
"src.meeting_lab.orchestration.mvp.transcribe_audio",
side_effect=AssertionError("Retranscription is forbidden"),
),
):
regenerate_mvp_protocol(run, meeting_context=context)
metadata = json.loads(
(run / "protocol" / "runtime_metadata.json").read_text()
)
self.assertEqual(metadata["output_language"], language or "de")
self.assertIn(
f"Write the meeting protocol in {expected}.",
call.call_args.args[2],
)
self.assertEqual(transcript.read_bytes(), original)
self.assertEqual(
load_meeting_context(context_path).data, context.data
)
self.assertEqual(context.speaker_mappings, {"SPEAKER_00": "person"})
def test_no_context_defaults_to_german(self) -> None:
with tempfile.TemporaryDirectory() as directory:
transcript = Path(directory) / "transcript.json"
transcript.write_text('{"text": "English source text."}')
result = generate_direct_protocol(
transcript,
model_check=Mock(return_value={}),
generation_call=Mock(
return_value=OllamaGeneration(
text="Protocol",
raw_response={},
client_wall_time_seconds=0.1,
)
),
)
self.assertIn("Write the meeting protocol in German.", result.exact_prompt)
self.assertEqual(result.runtime_metadata["output_language"], "de")
+434
View File
@@ -0,0 +1,434 @@
import json
import shutil
import subprocess
import tempfile
import unittest
from dataclasses import replace
from pathlib import Path
from unittest.mock import patch
from scripts import run_mvp_meeting as cli
from src.meeting_lab.audio import PreparedAudio
from src.meeting_lab.diarization.backend import DiarizationResult
from src.meeting_lab.llm.ollama import OllamaGeneration
from src.meeting_lab.protocol.generate_direct_protocol import generate_direct_protocol
from src.meeting_lab.models.meeting_context import load_meeting_context
from src.meeting_lab.orchestration import mvp as mvp_api
from src.meeting_lab.orchestration.mvp import MvpMeetingConfig, MvpRunResult
from src.meeting_lab.protocol.generate_direct_protocol import DirectProtocolResult
from src.meeting_lab.transcription.whisper import TranscriptionError, TranscriptionResult
def context_data():
return {
"schema_version": "1",
"meeting": {
"meeting_id": "programmatic-test",
"title": "Programmatic Test",
"language": "de",
"date": None,
"objective": "API prüfen",
"notes": "",
},
"participants": [
{
"participant_id": "person-1",
"display_name": "Test Person",
"aliases": [],
"role": "Projektleitung",
"department": None,
"attendance_status": "present",
"notes": None,
}
],
"speaker_mappings": {"SPEAKER_00": "person-1"},
"mentioned_people": [],
"organization": {"name": "Example", "departments": []},
"known_entities": {},
"context_rules": {"do_not_infer_responsibilities": True},
}
def fake_transcribe(audio, model, output, language, **kwargs):
output.mkdir(parents=True, exist_ok=True)
raw = output / "whisper_raw.json"
transcript = output / "transcript.json"
text = output / "transcript.txt"
metadata = output / "runtime_metadata.json"
raw.write_text('{"transcription": []}\n', encoding="utf-8")
transcript.write_text(
json.dumps(
{
"text": "Ein kurzer Besprechungstext.",
"segments": [
{
"id": 0,
"start": 0.0,
"end": 1.0,
"text": "Ein kurzer Besprechungstext.",
}
],
}
),
encoding="utf-8",
)
text.write_text("Ein kurzer Besprechungstext.\n", encoding="utf-8")
metadata.write_text('{"runtime_seconds": 0.1}\n', encoding="utf-8")
return TranscriptionResult(output, raw, transcript, text, metadata, 0.1)
def fake_prepare(source, destination, **kwargs):
destination.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(source, destination)
return PreparedAudio(source, source.suffix.removeprefix("."), destination, "ffmpeg", "ffmpeg")
def fake_diarize(audio, output, mode, **kwargs):
output.mkdir(parents=True, exist_ok=True)
metadata = output / "metadata.json"
ordinary = output / "diarization.rttm"
exclusive = output / "exclusive_diarization.rttm"
turns = output / "turns.json"
exclusive_turns = output / "exclusive_turns.json"
details = {"speaker_count": 2, "runtime_seconds": 0.1}
metadata.write_text(json.dumps(details), encoding="utf-8")
ordinary.write_text("", encoding="utf-8")
exclusive.write_text("", encoding="utf-8")
turns.write_text("[]", encoding="utf-8")
exclusive_turns.write_text(
json.dumps([
{"start": 0, "end": 0.5, "speaker_id": "SPEAKER_00"},
{"start": 0.5, "end": 1, "speaker_id": "SPEAKER_01"},
]), encoding="utf-8"
)
return DiarizationResult(output, metadata, ordinary, exclusive, turns, exclusive_turns, details)
def fake_protocol(transcript, context, **kwargs):
rendered_context = load_meeting_context(context)
assert rendered_context.meeting_id == "programmatic-test"
return DirectProtocolResult(
protocol_text="# Meeting Protocol\n",
exact_prompt="prompt",
model_metadata={"model": kwargs["model"]},
runtime_metadata={"request_count": 1},
raw_response={"response": "# Meeting Protocol\n"},
)
class MvpApiTests(unittest.TestCase):
def config(self, root: Path):
audio = root / "meeting.wav"
model = root / "model.bin"
audio.write_bytes(b"audio")
model.write_bytes(b"model")
return MvpMeetingConfig(
audio_file=audio,
whisper_model=model,
output_root=root / "runs",
model="test:model",
)
def test_post_diarization_checkpoint_skips_protocol_generation(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
config = replace(self.config(root), diarization="cpu", stop_after_diarization=True)
with (
patch.object(mvp_api, "prepare_audio", side_effect=fake_prepare),
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe),
patch.object(mvp_api, "diarize_audio", side_effect=fake_diarize),
patch.object(mvp_api, "generate_direct_protocol") as generate,
):
result = mvp_api.run_mvp_meeting(config, meeting_context=context_data())
self.assertEqual(result.exit_code, 0)
self.assertIsNone(result.protocol_path)
generate.assert_not_called()
metadata = json.loads((result.run_dir / "run_metadata.json").read_text())
self.assertEqual(metadata["status"], "awaiting_speaker_review")
self.assertEqual(metadata["speaker_review"]["speaker_count"], 2)
self.assertTrue((result.run_dir / "diarization/transcript_diarized.json").is_file())
def test_programmatic_context_is_persisted_without_source_yaml(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
config = self.config(root)
events = []
with (
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe),
patch.object(mvp_api, "prepare_audio", side_effect=fake_prepare),
patch.object(
mvp_api, "generate_direct_protocol", side_effect=fake_protocol
) as protocol_generator,
patch("src.meeting_lab.protocol.history._git_provenance", return_value={}),
patch.object(subprocess, "run") as subprocess_run,
):
result = mvp_api.run_mvp_meeting(
config, meeting_context=context_data(), progress_sink=events.append
)
self.assertEqual(result.exit_code, 0)
context_path = result.run_dir / "context/meeting_context.yaml"
self.assertTrue(context_path.is_file())
persisted = load_meeting_context(context_path)
self.assertEqual(persisted.meeting_id, "programmatic-test")
self.assertEqual(persisted.speaker_mappings, {"SPEAKER_00": "person-1"})
subprocess_run.assert_not_called()
self.assertEqual(
[(event.stage, event.status) for event in events],
[
("preparing", "started"),
("preparing", "completed"),
("transcription", "started"),
("transcription", "completed"),
("protocol_generation", "started"),
("protocol_generation", "completed"),
("completed", "completed"),
],
)
self.assertTrue(all(event.progress is None for event in events))
self.assertEqual(protocol_generator.call_args.kwargs["num_ctx"], 32_768)
self.assertEqual(
protocol_generator.call_args.kwargs["safe_input_token_budget"],
29_000,
)
def test_protocol_only_regeneration_reuses_diarized_artifacts(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
run_dir = root / "existing-run"
diarization_dir = run_dir / "diarization"
diarization_dir.mkdir(parents=True)
source = diarization_dir / "transcript_diarized.json"
source.write_text(
json.dumps(
{
"text": "SPEAKER_00: Existing statement.\n",
"segments": [
{
"start": 0.0,
"end": 1.0,
"speaker_id": "SPEAKER_00",
"text": "Existing statement.",
}
],
"speaker_labels_anonymous": True,
}
),
encoding="utf-8",
)
source_before = source.read_bytes()
mapped_context = context_data()
mapped_context["speaker_mappings"] = {"SPEAKER_00": "person-1"}
with (
patch.object(mvp_api, "prepare_audio") as preparation,
patch.object(mvp_api, "transcribe_audio") as transcription,
patch.object(mvp_api, "diarize_audio") as diarization,
patch.object(
mvp_api, "generate_direct_protocol", side_effect=fake_protocol
) as protocol,
):
result = mvp_api.regenerate_mvp_protocol(
run_dir,
meeting_context=mapped_context,
model="qwen3.8:27b",
protocol_num_ctx=32_768,
protocol_safe_input_token_budget=29_000,
)
self.assertEqual(result.exit_code, 0)
self.assertEqual(result.protocol_path, run_dir / "protocol.md")
preparation.assert_not_called()
transcription.assert_not_called()
diarization.assert_not_called()
self.assertEqual(protocol.call_args.args[0], source)
self.assertEqual(protocol.call_args.kwargs["num_ctx"], 32_768)
self.assertEqual(source.read_bytes(), source_before)
persisted = load_meeting_context(run_dir / "context/meeting_context.yaml")
self.assertEqual(persisted.speaker_mappings, {"SPEAKER_00": "person-1"})
def test_regeneration_keeps_glossary_out_of_transcript_and_in_context(self):
with tempfile.TemporaryDirectory() as directory:
run_dir = Path(directory)
diarization_dir = run_dir / "diarization"
diarization_dir.mkdir()
source = diarization_dir / "transcript_diarized.json"
source.write_text(
json.dumps(
{
"text": "SPEAKER_00: Lumini.",
"segments": [
{
"start": 0.0,
"end": 1.0,
"speaker_id": "SPEAKER_00",
"text": "Lumini.",
}
],
"speaker_labels_anonymous": True,
}
),
encoding="utf-8",
)
source_before = source.read_bytes()
transcript_dir = run_dir / "transcript"
transcript_dir.mkdir()
whisper_source = transcript_dir / "transcript.json"
whisper_source.write_text(
json.dumps({"text": "Lumini.", "segments": []}), encoding="utf-8"
)
whisper_source_before = whisper_source.read_bytes()
context = context_data()
context["known_entities"] = {
"Authoritative terminology": ["Luminy (aliases: Lumini)"]
}
def generate_without_network(transcript, context_path, **kwargs):
return generate_direct_protocol(
transcript,
context_path,
model=kwargs["model"],
num_ctx=kwargs["num_ctx"],
num_thread=kwargs["num_thread"],
safe_input_token_budget=kwargs["safe_input_token_budget"],
glossary_aliases=kwargs["glossary_aliases"],
model_check=lambda *_: {},
generation_call=lambda *_args, **_kwargs: OllamaGeneration(
raw_response={"response": "# Protocol", "done": True},
text="# Protocol",
client_wall_time_seconds=0.1,
),
)
with (
patch.object(
mvp_api,
"generate_direct_protocol",
side_effect=generate_without_network,
),
):
mvp_api.regenerate_mvp_protocol(
run_dir,
meeting_context=context,
glossary_aliases={"Lumini": "Luminy"},
)
generation = run_dir / "protocol"
transcript_input = (generation / "transcript_input.txt").read_text()
exact_prompt = (generation / "exact_prompt.txt").read_text()
metadata = json.loads((generation / "runtime_metadata.json").read_text())
self.assertIn("SPEAKER_00: Lumini.", transcript_input)
self.assertNotIn("SPEAKER_00: Luminy.", transcript_input)
self.assertIn("Luminy (aliases: Lumini)", exact_prompt)
self.assertIn("SPEAKER_00", exact_prompt)
self.assertIn("Test Person", exact_prompt)
self.assertEqual(metadata["glossary_replacements"], [])
self.assertEqual(source.read_bytes(), source_before)
self.assertEqual(whisper_source.read_bytes(), whisper_source_before)
def test_failure_emits_terminal_failure_event(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
config = self.config(root)
events = []
with patch.object(
mvp_api,
"transcribe_audio",
side_effect=TranscriptionError("stopped"),
), patch.object(mvp_api, "prepare_audio", side_effect=fake_prepare):
result = mvp_api.run_mvp_meeting(config, progress_sink=events.append)
self.assertEqual(result.exit_code, 2)
self.assertEqual(events[-1].stage, "failed")
self.assertEqual(events[-1].status, "failed")
self.assertIn("transcription", events[-1].message)
metadata = json.loads((result.run_dir / "run_metadata.json").read_text())
self.assertEqual(metadata["status"], "failed")
def test_cli_defaults_and_wrapper_delegate_without_subprocess(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
config = self.config(root)
args = cli.parse_args(
[str(config.audio_file), "--whisper-model", str(config.whisper_model)]
)
expected = MvpRunResult(0, root / "run", root / "run/protocol.md")
with patch.object(cli, "run_mvp_meeting", return_value=expected) as api:
actual = cli.run(args, context_override=context_data())
self.assertEqual(actual, (0, expected.run_dir, expected.protocol_path))
delegated = api.call_args.args[0]
self.assertEqual(delegated.diarization, "off")
self.assertEqual(delegated.language, "de")
self.assertEqual(delegated.whisper_executable, "whisper-cli")
self.assertEqual(delegated.ffmpeg_executable, "ffmpeg")
self.assertTrue(delegated.audio_normalization)
self.assertEqual(delegated.protocol_num_ctx, 32_768)
self.assertEqual(delegated.protocol_safe_input_token_budget, 29_000)
self.assertEqual(api.call_args.kwargs["meeting_context"], context_data())
def test_cli_explicit_audio_normalization_values_are_propagated(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
config = self.config(root)
enabled = cli.config_from_args(
cli.parse_args(
[
str(config.audio_file),
"--whisper-model",
str(config.whisper_model),
"--audio-normalization",
]
)
)
disabled = cli.config_from_args(
cli.parse_args(
[
str(config.audio_file),
"--whisper-model",
str(config.whisper_model),
"--no-audio-normalization",
]
)
)
self.assertTrue(enabled.audio_normalization)
self.assertFalse(disabled.audio_normalization)
def test_transcription_receives_prepared_wav_for_encoded_inputs(self):
for suffix in (".flac", ".m4a"):
with self.subTest(suffix=suffix), tempfile.TemporaryDirectory() as directory:
root = Path(directory)
config = self.config(root)
encoded = config.audio_file.with_suffix(suffix)
config.audio_file.rename(encoded)
config = replace(config, audio_file=encoded)
received = []
def capture_transcribe(audio, *args, received_paths=received, **kwargs):
received_paths.append(audio)
return fake_transcribe(audio, *args, **kwargs)
with (
patch.object(mvp_api, "prepare_audio", side_effect=fake_prepare),
patch.object(mvp_api, "transcribe_audio", side_effect=capture_transcribe),
patch.object(mvp_api, "generate_direct_protocol", side_effect=fake_protocol),
):
result = mvp_api.run_mvp_meeting(
config, meeting_context=context_data()
)
self.assertEqual(result.exit_code, 0)
self.assertEqual(received, [result.run_dir / "audio" / "prepared.wav"])
manifest = json.loads(
(result.run_dir / "audio" / "input_manifest.json").read_text()
)
self.assertEqual(manifest["format"], suffix.removeprefix("."))
self.assertEqual(
manifest["prepared_audio"]["prepared_audio_path"],
str((result.run_dir / "audio" / "prepared.wav").resolve()),
)
if __name__ == "__main__":
unittest.main()
+444
View File
@@ -0,0 +1,444 @@
import json
import shutil
import tempfile
import unittest
from datetime import datetime
from pathlib import Path
from unittest.mock import Mock, patch
from scripts import run_mvp_meeting
from src.meeting_lab.audio import PreparedAudio
from src.meeting_lab.audio.preparation import (
DEFAULT_NORMALIZATION_FILTER,
DEFAULT_NORMALIZATION_METHOD,
)
from src.meeting_lab.orchestration import mvp as mvp_api
from src.meeting_lab.protocol.generate_direct_protocol import DirectProtocolResult
from src.meeting_lab.diarization.backend import DiarizationResult
from src.meeting_lab.transcription.whisper import TranscriptionError, TranscriptionResult
VALID_CONTEXT = """schema_version: "1"
meeting:
meeting_id: "mvp-test"
title: "MVP Test"
language: "de"
participants: []
mentioned_people: []
organization:
departments: []
known_entities: {}
"""
def protocol_result(model: str = "chosen:model") -> DirectProtocolResult:
text = "# Protokoll\n\nUnverändert. \n"
return DirectProtocolResult(
protocol_text=text,
exact_prompt="exact prompt\n",
model_metadata={"model": model},
runtime_metadata={"model": model, "request_count": 1, "client_wall_time_seconds": 0.5},
raw_response={"response": text, "done": True},
transcript_input="selected transcript\n",
)
def fake_transcribe(
audio_path: Path,
model_path: Path,
output_dir: Path,
language: str,
*,
executable: str,
threads: str | int,
) -> TranscriptionResult:
output_dir.mkdir(parents=True, exist_ok=True)
raw = output_dir / "whisper_raw.json"
transcript = output_dir / "transcript.json"
text = output_dir / "transcript.txt"
metadata = output_dir / "runtime_metadata.json"
raw.write_text('{"transcription": []}\n', encoding="utf-8")
transcript.write_text(
json.dumps(
{
"text": "Besprechungstext.",
"segments": [
{"id": 0, "start": 0.0, "end": 1.0, "text": "Besprechungstext."}
],
}
)
+ "\n",
encoding="utf-8",
)
text.write_text("Besprechungstext.\n", encoding="utf-8")
metadata.write_text('{"runtime_seconds": 1.25}\n', encoding="utf-8")
return TranscriptionResult(output_dir, raw, transcript, text, metadata, 1.25)
def fake_diarize(audio_path, output_dir, device_mode, **kwargs):
output_dir.mkdir(parents=True, exist_ok=True)
metadata = {
"backend": "pyannote.audio",
"model": "pyannote/speaker-diarization-community-1",
"requested_device_mode": device_mode,
"actual_device": "cuda",
"device_name": "Fake GPU",
"runtime_seconds": 2.5,
"speaker_count": 1,
"credentials_persisted": False,
}
paths = {
"metadata": output_dir / "metadata.json",
"ordinary": output_dir / "diarization.rttm",
"exclusive": output_dir / "exclusive_diarization.rttm",
"turns": output_dir / "turns.json",
"exclusive_turns": output_dir / "exclusive_turns.json",
}
paths["metadata"].write_text(json.dumps(metadata), encoding="utf-8")
paths["ordinary"].write_text("", encoding="utf-8")
paths["exclusive"].write_text("", encoding="utf-8")
paths["turns"].write_text("[]", encoding="utf-8")
paths["exclusive_turns"].write_text(
json.dumps([{"start": 0, "end": 10, "speaker_id": "SPEAKER_00"}]),
encoding="utf-8",
)
return DiarizationResult(
output_dir,
paths["metadata"],
paths["ordinary"],
paths["exclusive"],
paths["turns"],
paths["exclusive_turns"],
metadata,
)
def fake_prepare(source, destination, **kwargs):
destination.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(source, destination)
normalization_enabled = kwargs.get("normalization_enabled", True)
return PreparedAudio(
source,
source.suffix.removeprefix("."),
destination,
"ffmpeg",
kwargs.get("ffmpeg_executable", "ffmpeg"),
normalization_enabled,
DEFAULT_NORMALIZATION_METHOD if normalization_enabled else None,
DEFAULT_NORMALIZATION_FILTER if normalization_enabled else None,
)
class MvpOrchestratorTests(unittest.TestCase):
def setUp(self) -> None:
patcher = patch.object(mvp_api, "prepare_audio", side_effect=fake_prepare)
self.prepare_audio = patcher.start()
self.addCleanup(patcher.stop)
def create_inputs(self, root: Path) -> tuple[Path, Path, Path]:
audio = root / "team meeting.wav"
whisper_model = root / "ggml-model.bin"
context = root / "source-context.yaml"
audio.write_bytes(b"audio")
whisper_model.write_bytes(b"model")
context.write_text(VALID_CONTEXT, encoding="utf-8")
return audio, whisper_model, context
def args(self, root: Path, extra: list[str] | None = None):
audio, whisper_model, context = self.create_inputs(root)
values = [
str(audio),
"--whisper-model", str(whisper_model),
"--context", str(context),
"--output-root", str(root / "runs"),
"--language", "de",
"--model", "chosen:model",
"--ollama-endpoint", "http://ollama.test:11434",
]
if extra:
values.extend(extra)
return run_mvp_meeting.parse_args(values)
def test_successful_full_orchestration_and_artifact_layout(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
args = self.args(root)
with (
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe) as whisper,
patch.object(
mvp_api,
"generate_direct_protocol",
return_value=protocol_result(),
) as protocol,
):
code, run_dir, protocol_path = run_mvp_meeting.run(args)
self.assertEqual(code, 0)
self.assertIsNotNone(run_dir)
self.assertEqual(protocol_path, run_dir / "protocol.md")
expected = {
"run_metadata.json",
"audio/input_manifest.json",
"audio/prepared.wav",
"audio/preparation_metadata.json",
"transcript/whisper_raw.json",
"transcript/transcript.json",
"transcript/transcript.txt",
"transcript/runtime_metadata.json",
"context/meeting_context.yaml",
"protocol/exact_prompt.txt",
"protocol/transcript_input.txt",
"protocol/raw_response.json",
"protocol/runtime_metadata.json",
"protocol.md",
}
self.assertTrue(all((run_dir / item).is_file() for item in expected))
metadata = json.loads((run_dir / "run_metadata.json").read_text())
self.assertEqual(metadata["status"], "completed")
self.assertEqual(metadata["model"], "chosen:model")
self.assertIsNone(metadata["failure"])
self.assertEqual(whisper.call_count, 1)
self.assertEqual(
whisper.call_args.args[0], run_dir / "audio" / "prepared.wav"
)
self.assertEqual(protocol.call_count, 1)
self.assertTrue(
self.prepare_audio.call_args.kwargs["normalization_enabled"]
)
preparation = json.loads(
(run_dir / "audio" / "preparation_metadata.json").read_text()
)
self.assertTrue(preparation["normalization_enabled"])
def test_cli_can_disable_audio_normalization_without_bypassing_preparation(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
args = self.args(root, ["--no-audio-normalization"])
with (
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe) as whisper,
patch.object(
mvp_api,
"generate_direct_protocol",
return_value=protocol_result(),
),
):
code, run_dir, _ = run_mvp_meeting.run(args)
self.assertEqual(code, 0)
self.prepare_audio.assert_called_once()
self.assertFalse(
self.prepare_audio.call_args.kwargs["normalization_enabled"]
)
self.assertEqual(
whisper.call_args.args[0], run_dir / "audio" / "prepared.wav"
)
preparation = json.loads(
(run_dir / "audio" / "preparation_metadata.json").read_text()
)
self.assertFalse(preparation["normalization_enabled"])
self.assertIsNone(preparation["normalization_method"])
self.assertIsNone(preparation["normalization_filter"])
def test_context_model_endpoint_and_whisper_options_are_forwarded(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
args = self.args(
root,
[
"--whisper-executable",
"/tools/whisper-cli",
"--ffmpeg-executable",
"/tools/ffmpeg",
"--threads",
"4",
],
)
with (
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe) as whisper,
patch.object(
mvp_api,
"generate_direct_protocol",
return_value=protocol_result(),
) as protocol,
):
code, run_dir, _ = run_mvp_meeting.run(args)
self.assertEqual(code, 0)
self.assertEqual(whisper.call_args.args[3], "de")
self.assertEqual(whisper.call_args.kwargs["executable"], "/tools/whisper-cli")
self.assertEqual(whisper.call_args.kwargs["threads"], "4")
self.assertEqual(
self.prepare_audio.call_args.kwargs["ffmpeg_executable"],
"/tools/ffmpeg",
)
self.assertEqual(protocol.call_args.args[1], run_dir / "context/meeting_context.yaml")
self.assertEqual(protocol.call_args.kwargs["model"], "chosen:model")
self.assertEqual(
protocol.call_args.kwargs["endpoint"], "http://ollama.test:11434"
)
def test_diarization_is_off_by_default_and_preserves_protocol_input(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
args = self.args(root)
with (
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe),
patch.object(mvp_api, "diarize_audio") as diarization,
patch.object(
mvp_api,
"generate_direct_protocol",
return_value=protocol_result(),
) as protocol,
):
code, run_dir, _ = run_mvp_meeting.run(args)
self.assertEqual(code, 0)
diarization.assert_not_called()
self.assertEqual(protocol.call_args.args[0], run_dir / "transcript/transcript.json")
self.assertFalse((run_dir / "diarization").exists())
def test_diarization_cli_propagates_and_uses_derived_protocol_input(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
args = self.args(
root,
[
"--diarization", "gpu",
"--diarization-runtime", "container",
"--diarization-container-image", "rocm/test",
"--diarization-container-arg=--device=/dev/kfd",
],
)
with (
patch.object(
mvp_api, "transcribe_audio", side_effect=fake_transcribe
),
patch.object(
mvp_api, "diarize_audio", side_effect=fake_diarize
) as diarization,
patch.object(
mvp_api,
"generate_direct_protocol",
return_value=protocol_result(),
) as protocol,
):
code, run_dir, _ = run_mvp_meeting.run(args)
self.assertEqual(code, 0)
self.assertEqual(diarization.call_args.args[2], "gpu")
self.assertEqual(
diarization.call_args.args[0], run_dir / "audio" / "prepared.wav"
)
self.assertEqual(diarization.call_args.kwargs["runtime"], "container")
self.assertEqual(
diarization.call_args.kwargs["container_args"], ("--device=/dev/kfd",)
)
derived = run_dir / "diarization/transcript_diarized.json"
self.assertEqual(protocol.call_args.args[0], derived)
self.assertIn("SPEAKER_00", derived.read_text(encoding="utf-8"))
self.assertEqual(
json.loads((run_dir / "transcript/transcript.json").read_text())["text"],
"Besprechungstext.",
)
run_metadata = json.loads((run_dir / "run_metadata.json").read_text())
self.assertTrue(run_metadata["diarization"]["enabled"])
self.assertNotIn("HF_TOKEN", json.dumps(run_metadata))
def test_whisper_failure_is_recorded_and_protocol_is_not_called(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
args = self.args(root)
with (
patch.object(
mvp_api,
"transcribe_audio",
side_effect=TranscriptionError("whisper stopped"),
),
patch.object(mvp_api, "generate_direct_protocol") as protocol,
):
code, run_dir, protocol_path = run_mvp_meeting.run(args)
metadata = json.loads((run_dir / "run_metadata.json").read_text())
self.assertEqual(code, 2)
self.assertIsNone(protocol_path)
self.assertEqual(metadata["status"], "failed")
self.assertEqual(metadata["failure"]["stage"], "whisper")
self.assertIn("whisper stopped", metadata["failure"]["message"])
protocol.assert_not_called()
self.assertTrue((run_dir / "audio/input_manifest.json").is_file())
def test_protocol_failure_preserves_transcript_and_failure_metadata(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
args = self.args(root)
with (
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe),
patch.object(
mvp_api,
"generate_direct_protocol",
side_effect=ValueError("generation stopped"),
),
):
code, run_dir, protocol_path = run_mvp_meeting.run(args)
metadata = json.loads((run_dir / "run_metadata.json").read_text())
self.assertEqual(code, 2)
self.assertIsNone(protocol_path)
self.assertEqual(metadata["failure"]["stage"], "protocol")
self.assertTrue((run_dir / "transcript/transcript.json").is_file())
self.assertFalse((run_dir / "protocol.md").exists())
def test_unique_run_directories_do_not_overwrite(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
fixed = datetime(2026, 8, 21, 9, 15, 30)
first = run_mvp_meeting.create_unique_run_dir(root, "team meeting", lambda: fixed)
marker = first / "keep.txt"
marker.write_text("keep", encoding="utf-8")
second = run_mvp_meeting.create_unique_run_dir(root, "team meeting", lambda: fixed)
self.assertEqual(first.name, "team_meeting_20260821_091530")
self.assertEqual(second.name, "team_meeting_20260821_091530_01")
self.assertEqual(marker.read_text(), "keep")
def test_semantic_pipeline_functions_are_never_invoked(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
args = self.args(root)
with (
patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe),
patch.object(
mvp_api,
"generate_direct_protocol",
return_value=protocol_result(),
),
patch("src.meeting_lab.chunking.chunk_transcript.build_chunks") as chunking,
patch("src.meeting_lab.extraction.extract_chunks.extract_input") as extraction,
patch("src.meeting_lab.consolidation.consolidate_facts.call_ollama") as consolidation,
):
code, _, _ = run_mvp_meeting.run(args)
self.assertEqual(code, 0)
chunking.assert_not_called()
extraction.assert_not_called()
consolidation.assert_not_called()
def test_main_returns_nonzero_for_whisper_failure(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
args = self.args(root)
argv = [
str(args.audio_file),
"--whisper-model", str(args.whisper_model),
"--context", str(args.context),
"--output-root", str(args.output_root),
]
with patch.object(
mvp_api,
"transcribe_audio",
side_effect=TranscriptionError("failed"),
):
self.assertEqual(run_mvp_meeting.main(argv), 2)
if __name__ == "__main__":
unittest.main()
+169
View File
@@ -0,0 +1,169 @@
"""Publication fault injection without inference or audio processing."""
import json
import multiprocessing
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from src.meeting_lab.models.meeting_context import create_meeting_context, load_meeting_context
from src.meeting_lab.protocol import history
from src.meeting_lab.orchestration import mvp
from src.meeting_lab.protocol.generate_direct_protocol import DirectProtocolResult
from tests import test_mvp_api as fixtures
def result(text):
return DirectProtocolResult(
protocol_text=text,
exact_prompt="prompt " + text,
transcript_input="Lumini original",
model_metadata={"name": "test"},
runtime_metadata={
"glossary_aliases_configured": {"Lumini": "Luminy"},
"glossary_replacements": [],
"num_thread": 10,
},
raw_response={"response": text},
)
def interrupt_publication(root):
original = history._replace_link
def interrupted(path, target):
if path.name == "current":
os._exit(77)
original(path, target)
history._replace_link = interrupted
history.persist_protocol_generation(
Path(root), result("new"), context=create_meeting_context(fixtures.context_data())
)
class HistoryTests(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory()
self.addCleanup(self.temporary.cleanup)
self.root = Path(self.temporary.name)
self.context = create_meeting_context(fixtures.context_data())
def persist(self, text):
return history.persist_protocol_generation(self.root, result(text), context=self.context)
def snapshot(self):
names = [
"protocol.md",
"context/meeting_context.yaml",
*["protocol/" + n for n in history._DIAGNOSTICS],
]
return {name: (self.root / name).read_bytes() for name in names}
def test_success_numbering_provenance_and_immutable_context(self):
self.persist("first")
first = self.snapshot()
self.persist("second")
self.assertEqual((self.root / "protocol.md").read_text(), "second")
old = self.root / "protocol/generations/001"
self.assertEqual((old / "protocol.md").read_bytes(), first["protocol.md"])
self.assertEqual(
(old / "meeting_context.yaml").read_bytes(), first["context/meeting_context.yaml"]
)
current = (self.root / "protocol/current").resolve()
self.assertEqual(current.name, "002")
metadata = json.loads((current / "runtime_metadata.json").read_text())
self.assertEqual(metadata["generation_index"], 2)
self.assertEqual(metadata["speaker_mapping"], self.context.speaker_mappings)
self.assertEqual(metadata["glossary_aliases_configured"], {"Lumini": "Luminy"})
self.assertEqual(metadata["glossary_replacements"], [])
self.assertEqual(metadata["num_thread"], 10)
self.assertEqual(metadata["source_run_id"], self.root.name)
self.assertEqual(
json.loads((current / "model_metadata.json").read_text()), {"name": "test"}
)
self.assertIn("sha", metadata["git"]["meeting_lab"])
self.assertEqual(
load_meeting_context(current / "meeting_context.yaml").data, self.context.data
)
for name in history._DIAGNOSTICS:
self.assertEqual((self.root / "protocol" / name).resolve(), current / name)
def test_failure_writing_record_preserves_previous(self):
self.persist("first")
before = self.snapshot()
with patch.object(history, "_write_json", side_effect=OSError("disk full")):
with self.assertRaises(OSError):
self.persist("second")
self.assertEqual(self.snapshot(), before)
self.persist("third")
self.assertEqual((self.root / "protocol/current").resolve().name, "002")
def test_failure_at_atomic_publication_preserves_all_stable_paths(self):
self.persist("first")
before = self.snapshot()
original = history.os.replace
def fail_current(source, destination):
if Path(destination).name == "current":
raise OSError("publication stopped")
original(source, destination)
with patch.object(history.os, "replace", side_effect=fail_current):
with self.assertRaises(OSError):
self.persist("second")
self.assertEqual(self.snapshot(), before)
self.assertFalse((self.root / "protocol/generations/002").exists())
def test_process_interruption_before_pointer_swap_preserves_current(self):
self.persist("first")
before = self.snapshot()
worker = multiprocessing.get_context("fork").Process(
target=interrupt_publication, args=(str(self.root),)
)
worker.start()
worker.join(timeout=10)
self.assertFalse(worker.is_alive())
self.assertEqual(worker.exitcode, 77)
self.assertEqual(self.snapshot(), before)
self.persist("third")
self.assertEqual((self.root / "protocol.md").read_text(), "third")
def test_legacy_migration_and_failure_during_link_conversion(self):
(self.root / "protocol").mkdir()
(self.root / "context").mkdir()
(self.root / "protocol.md").write_text("legacy")
for name in history._DIAGNOSTICS:
(self.root / "protocol" / name).write_text("legacy " + name)
(self.root / "context/meeting_context.yaml").write_text("legacy context")
before = self.snapshot()
original = history._replace_link
def fail_conversion(path, target):
if path.name == "raw_response.json":
raise OSError("link conversion stopped")
original(path, target)
with patch.object(history, "_replace_link", side_effect=fail_conversion):
with self.assertRaises(OSError):
self.persist("new")
self.assertEqual(self.snapshot(), before)
self.persist("new")
self.assertEqual((self.root / "protocol/generations/001/protocol.md").read_text(), "legacy")
self.assertEqual((self.root / "protocol/current").resolve().name, "002")
def test_failed_model_does_not_change_context_or_latest_generation(self):
self.persist("first")
before = self.snapshot()
(self.root / "transcript").mkdir()
(self.root / "transcript/transcript.json").write_text('{"text":"original"}')
different = fixtures.context_data()
different["speaker_mappings"] = {}
with patch.object(
mvp, "generate_direct_protocol", side_effect=RuntimeError("model stopped")
):
with self.assertRaises(RuntimeError):
mvp.regenerate_mvp_protocol(self.root, meeting_context=different)
self.assertEqual(self.snapshot(), before)
+58
View File
@@ -0,0 +1,58 @@
"""Protocol thread configuration through the public pipeline boundaries."""
import tempfile
import unittest
from dataclasses import replace
from pathlib import Path
from unittest.mock import Mock, patch
from src.meeting_lab.llm import ollama
from src.meeting_lab.orchestration import mvp
from tests import test_mvp_api as fixtures
class ProtocolThreadTests(unittest.TestCase):
def test_initial_and_regeneration_forward_default_and_explicit_threads(self):
for threads in (None, 4, 10, 16):
with self.subTest(threads=threads), tempfile.TemporaryDirectory() as directory:
config = replace(
fixtures.MvpApiTests().config(Path(directory)), protocol_num_thread=threads
)
with (
patch.object(mvp, "prepare_audio", side_effect=fixtures.fake_prepare),
patch.object(mvp, "transcribe_audio", side_effect=fixtures.fake_transcribe),
patch.object(
mvp, "generate_direct_protocol", side_effect=fixtures.fake_protocol
) as generate,
):
result = mvp.run_mvp_meeting(config, meeting_context=fixtures.context_data())
self.assertEqual(result.exit_code, 0)
self.assertEqual(generate.call_args.kwargs["num_thread"], threads)
mvp.regenerate_mvp_protocol(
result.run_dir,
meeting_context=fixtures.context_data(),
protocol_num_thread=threads,
)
self.assertEqual(generate.call_args.kwargs["num_thread"], threads)
def test_explicit_payload_and_invalid_values(self):
response = Mock()
response.json.return_value = {"response": "protocol"}
with patch.object(ollama.requests, "post", return_value=response) as post:
ollama.generate_once(
"url", "model", "prompt", timeout=1, num_ctx=32, num_predict=8, num_thread=10
)
self.assertEqual(post.call_args.kwargs["json"]["options"]["num_thread"], 10)
post.reset_mock()
for value in (0, -1, True, 1.5):
with self.assertRaises(ValueError):
ollama.generate_once(
"url",
"model",
"prompt",
timeout=1,
num_ctx=32,
num_predict=8,
num_thread=value,
)
post.assert_not_called()
+352
View File
@@ -0,0 +1,352 @@
import json
import tempfile
import unittest
from pathlib import Path
from unittest.mock import Mock, patch
from src.meeting_lab.diarization.alignment import diarized_transcript_text
from src.meeting_lab.llm import ollama
from src.meeting_lab.llm.ollama import OllamaGeneration
from src.meeting_lab.protocol.direct_protocol_prompt import (
COMPACT_DIARIZED_PROTOCOL_INSTRUCTION,
build_direct_protocol_prompt,
)
from src.meeting_lab.protocol.generate_direct_protocol import (
DirectProtocolError,
estimate_input_tokens,
generate_direct_protocol,
)
from src.meeting_lab.protocol.transcript_input import compact_diarized_transcript
def segments() -> list[dict[str, object]]:
return [
{"id": 0, "start": 0.0, "end": 1.0, "text": "First.", "speaker_id": "SPEAKER_01"},
{"id": 1, "start": 1.0, "end": 2.0, "text": "Second.", "speaker_id": "SPEAKER_01"},
{"id": 2, "start": 2.0, "end": 3.0, "text": "Third.", "speaker_id": "SPEAKER_04"},
{"id": 3, "start": 3.0, "end": 4.0, "text": "Unassigned.", "speaker_id": None},
{"id": 4, "start": 4.0, "end": 5.0, "text": "Last.", "speaker_id": "SPEAKER_01"},
]
def diarized_document(repetitions: int = 1) -> dict[str, object]:
source = segments() * repetitions
return {
"text": diarized_transcript_text(source),
"segments": source,
"speaker_labels_anonymous": True,
"alignment_source": "exclusive_diarization",
}
def completed_generation() -> OllamaGeneration:
text = "# Meeting Protocol\n\nComplete."
return OllamaGeneration(
raw_response={
"response": text,
"done": True,
"done_reason": "stop",
"prompt_eval_count": 100,
"eval_count": 10,
},
text=text,
client_wall_time_seconds=0.1,
)
class CompactDiarizedTranscriptTests(unittest.TestCase):
def test_adjacent_segments_group_and_transitions_remain_separate(self) -> None:
compact = compact_diarized_transcript(segments())
self.assertEqual(
[block.speaker_id for block in compact.blocks],
["SPEAKER_01", "SPEAKER_04", "SPEAKER_UNASSIGNED", "SPEAKER_01"],
)
self.assertEqual(compact.blocks[0].segment_texts, ("First.", "Second."))
self.assertEqual(compact.blocks[-1].segment_texts, ("Last.",))
self.assertEqual(compact.text.count("SPEAKER_01:"), 2)
def test_every_segment_text_and_order_are_preserved(self) -> None:
source = segments()
compact = compact_diarized_transcript(source)
self.assertEqual(compact.source_segment_count, len(source))
self.assertEqual(compact.represented_segment_count, len(source))
self.assertEqual(
compact.segment_texts,
tuple(str(segment["text"]) for segment in source),
)
self.assertEqual(compact.segment_texts[0], "First.")
self.assertEqual(compact.segment_texts[-1], "Last.")
self.assertIn("SPEAKER_UNASSIGNED: Unassigned.", compact.text)
def test_compact_form_is_materially_smaller_than_per_segment_format(self) -> None:
source = [
{
"start": index,
"end": index + 1,
"text": "Repeated transcript content.",
"speaker_id": "SPEAKER_01",
}
for index in range(100)
]
compact = compact_diarized_transcript(source).text
verbose = diarized_transcript_text(source)
self.assertLess(len(compact), len(verbose) * 0.6)
class ProtocolInputBudgetTests(unittest.TestCase):
def _write(self, root: Path, document: dict[str, object]) -> Path:
path = root / "transcript.json"
path.write_text(json.dumps(document), encoding="utf-8")
return path
def test_token_estimate_uses_utf8_bytes_for_non_ascii_safety(self) -> None:
self.assertEqual(estimate_input_tokens("ä" * 44), 20)
def test_compact_diarized_representation_selected_within_budget(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
document = diarized_document()
transcript = self._write(root, document)
compact = compact_diarized_transcript(document["segments"]).text
budget = estimate_input_tokens(
build_direct_protocol_prompt(
compact,
instruction=COMPACT_DIARIZED_PROTOCOL_INSTRUCTION,
)
)
call = Mock(return_value=completed_generation())
result = generate_direct_protocol(
transcript,
safe_input_token_budget=budget,
model_check=Mock(return_value={}),
generation_call=call,
)
self.assertEqual(
result.runtime_metadata["selected_transcript_representation"],
"diarized_compact",
)
self.assertFalse(result.runtime_metadata["fallback_used"])
self.assertTrue(result.runtime_metadata["diarization_enabled"])
self.assertEqual(result.runtime_metadata["safe_input_token_budget"], budget)
self.assertEqual(result.transcript_input, compact)
self.assertEqual(call.call_count, 1)
def test_glossary_aliases_do_not_mutate_protocol_input_or_source_artifact(
self,
) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
document = {
"text": "Lumini und Carbofool.",
"segments": [
{
"start": 0.0,
"end": 1.0,
"speaker_id": "SPEAKER_00",
"text": "Lumini und Carbofool.",
}
],
"speaker_labels_anonymous": True,
}
transcript = self._write(root, document)
source_before = transcript.read_bytes()
result = generate_direct_protocol(
transcript,
glossary_aliases={"Lumini": "Luminy", "Carbofool": "Carbofol"},
model_check=Mock(return_value={}),
generation_call=Mock(return_value=completed_generation()),
)
self.assertEqual(transcript.read_bytes(), source_before)
self.assertIn("SPEAKER_00: Lumini und Carbofool.", result.transcript_input)
self.assertIn("SPEAKER_00: Lumini und Carbofool.", result.exact_prompt)
self.assertEqual(result.runtime_metadata["glossary_replacements"], [])
self.assertEqual(
result.runtime_metadata["glossary_aliases_configured"],
{"Carbofool": "Carbofol", "Lumini": "Luminy"},
)
def test_mapped_speakers_and_statements_reach_final_ollama_payload(self) -> None:
diarized = {
"text": "",
"segments": [
{
"start": 0.0,
"end": 1.0,
"speaker_id": "SPEAKER_00",
"text": "We will run the trial on Wednesday.",
},
{
"start": 1.0,
"end": 2.0,
"speaker_id": "SPEAKER_01",
"text": "I will prepare the raw materials before then.",
},
{
"start": 2.0,
"end": 3.0,
"speaker_id": "SPEAKER_00",
"text": "Good. Anna owns the material preparation.",
},
],
"speaker_labels_anonymous": True,
"alignment_source": "exclusive_diarization",
}
context = {
"schema_version": "1",
"meeting": {
"meeting_id": "speaker-test",
"title": "Speaker test",
"language": "en",
},
"participants": [
{"participant_id": "martin", "display_name": "Martin"},
{"participant_id": "anna", "display_name": "Anna"},
],
"speaker_mappings": {"SPEAKER_00": "martin", "SPEAKER_01": "anna"},
"mentioned_people": [],
"organization": {"departments": []},
"known_entities": {},
}
response = Mock()
response.raise_for_status.return_value = None
response.json.return_value = {"response": "# Meeting Protocol\n", "done": True}
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
transcript = self._write(root, diarized)
context_path = root / "context.yaml"
context_path.write_text(json.dumps(context), encoding="utf-8")
with patch.object(ollama.requests, "post", return_value=response) as post:
result = generate_direct_protocol(
transcript,
context_path,
model="qwen3.8:27b",
model_check=Mock(return_value={}),
)
prompt = post.call_args.kwargs["json"]["prompt"]
self.assertEqual(result.exact_prompt, prompt)
self.assertIn("- SPEAKER_00: Martin (participant_id: martin)", prompt)
self.assertIn("- SPEAKER_01: Anna (participant_id: anna)", prompt)
self.assertIn("SPEAKER_00: We will run the trial on Wednesday.", prompt)
self.assertIn(
"SPEAKER_01: I will prepare the raw materials before then.", prompt
)
self.assertIn("SPEAKER_00: Good. Anna owns the material preparation.", prompt)
self.assertNotIn("Martin: We will run the trial on Wednesday.", prompt)
self.assertNotIn("Anna: I will prepare the raw materials before then.", prompt)
self.assertIn("autoritativen SPEAKER_XX-zu-Teilnehmer-Zuordnungen", prompt)
self.assertIn("Ich-Zusage", prompt)
self.assertIn("nur erwähnten Personen", prompt)
self.assertIn("keine persönliche Verantwortung", prompt)
def test_plain_fallback_selected_when_diarized_compact_exceeds_budget(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
alternating_segments = [
{
"id": index,
"start": float(index),
"end": float(index + 1),
"text": "Word.",
"speaker_id": f"SPEAKER_{index % 2:02d}",
}
for index in range(200)
]
document = {
"text": diarized_transcript_text(alternating_segments),
"segments": alternating_segments,
"speaker_labels_anonymous": True,
"alignment_source": "exclusive_diarization",
}
transcript = self._write(root, document)
compact = compact_diarized_transcript(document["segments"]).text
plain = " ".join(str(segment["text"]) for segment in document["segments"])
compact_estimate = estimate_input_tokens(
build_direct_protocol_prompt(
compact,
instruction=COMPACT_DIARIZED_PROTOCOL_INSTRUCTION,
)
)
plain_estimate = estimate_input_tokens(build_direct_protocol_prompt(plain))
self.assertLess(plain_estimate, compact_estimate)
result = generate_direct_protocol(
transcript,
safe_input_token_budget=plain_estimate,
model_check=Mock(return_value={}),
generation_call=Mock(return_value=completed_generation()),
)
self.assertEqual(
result.runtime_metadata["selected_transcript_representation"],
"plain_transcript_fallback",
)
self.assertTrue(result.runtime_metadata["fallback_used"])
self.assertEqual(result.transcript_input, plain)
self.assertNotIn("SPEAKER_00", result.exact_prompt)
self.assertNotIn("SPEAKER_01", result.exact_prompt)
self.assertFalse(result.runtime_metadata["speaker_attribution_available"])
self.assertEqual(
result.runtime_metadata["speaker_attribution_loss_reason"],
"plain_transcript_fallback",
)
def test_oversized_plain_transcript_fails_before_any_network_call(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
transcript = self._write(
root,
{"text": "large input " * 1000, "segments": []},
)
model_check = Mock()
generation_call = Mock()
with self.assertRaisesRegex(
DirectProtocolError,
"No LLM request was made; silent truncation is not allowed",
):
generate_direct_protocol(
transcript,
safe_input_token_budget=1,
model_check=model_check,
generation_call=generation_call,
)
model_check.assert_not_called()
generation_call.assert_not_called()
def test_existing_plain_path_and_metadata_remain_direct(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
transcript = self._write(
root,
{"text": "Plain original transcript.", "segments": []},
)
result = generate_direct_protocol(
transcript,
model_check=Mock(return_value={}),
generation_call=Mock(return_value=completed_generation()),
)
self.assertEqual(result.transcript_input, "Plain original transcript.")
self.assertEqual(
result.runtime_metadata["selected_transcript_representation"],
"plain_transcript",
)
self.assertFalse(result.runtime_metadata["fallback_used"])
self.assertFalse(result.runtime_metadata["diarization_enabled"])
self.assertIsNone(result.runtime_metadata["speaker_attribution_available"])
if __name__ == "__main__":
unittest.main()
+221
View File
@@ -0,0 +1,221 @@
import json
import subprocess
import tempfile
import unittest
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import Mock, patch
from scripts import transcribe_meeting
from src.meeting_lab.transcription.whisper import (
TranscriptionError,
compact_transcript,
transcribe_audio,
)
RAW_RESULT = {
"params": {"language": "de"},
"transcription": [
{
"timestamps": {"from": "00:00:00,000", "to": "00:00:01,200"},
"offsets": {"from": 0, "to": 1200},
"text": " Hallo ",
},
{
"timestamps": {"from": "00:00:01,200", "to": "00:00:02,500"},
"offsets": {"from": 1200, "to": 2500},
"text": "Welt",
},
],
}
class FakeRunner:
def __init__(self, *, returncode: int = 0, write_raw: bool = True) -> None:
self.returncode = returncode
self.write_raw = write_raw
self.commands: list[list[str]] = []
def __call__(self, command, **kwargs):
command = list(command)
self.commands.append(command)
if self.write_raw:
prefix = Path(command[command.index("-of") + 1])
prefix.with_suffix(".json").write_text(
json.dumps(RAW_RESULT, ensure_ascii=False), encoding="utf-8"
)
return subprocess.CompletedProcess(
command, self.returncode, stdout="", stderr="backend failure"
)
class TranscriptionTests(unittest.TestCase):
def create_inputs(self, root: Path) -> tuple[Path, Path]:
audio = root / "meeting.wav"
model = root / "ggml-model.bin"
audio.write_bytes(b"audio")
model.write_bytes(b"model")
return audio, model
def test_missing_audio_file_is_rejected_before_execution(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
model = root / "model.bin"
model.write_bytes(b"model")
runner = FakeRunner()
with self.assertRaisesRegex(TranscriptionError, "Audio file does not exist"):
transcribe_audio(root / "missing.wav", model, root / "out", runner=runner)
self.assertEqual(runner.commands, [])
def test_invalid_parameters_are_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
audio, model = self.create_inputs(root)
with self.assertRaisesRegex(TranscriptionError, "model does not exist"):
transcribe_audio(audio, root / "missing.bin", root / "out")
with self.assertRaisesRegex(TranscriptionError, "Language must"):
transcribe_audio(audio, model, root / "out", " ")
with self.assertRaisesRegex(TranscriptionError, "executable must"):
transcribe_audio(audio, model, root / "out", executable="")
with self.assertRaisesRegex(TranscriptionError, "Threads must"):
transcribe_audio(audio, model, root / "out", threads="zero")
def test_auto_threads_use_detected_physical_core_count_and_flash_attention(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
audio, model = self.create_inputs(root)
runner = FakeRunner()
result = transcribe_audio(
audio,
model,
root / "output",
runner=runner,
thread_detector=lambda: 6,
)
command = runner.commands[0]
self.assertEqual(command[command.index("-t") + 1], "6")
self.assertIn("-fa", command)
metadata = json.loads(result.runtime_metadata.read_text())
self.assertEqual(metadata["threads"], 6)
self.assertEqual(metadata["threads_option"], "auto")
self.assertTrue(metadata["flash_attention"])
self.assertEqual(metadata["whisper_executable"], "whisper-cli")
def test_explicit_thread_count_is_forwarded_unchanged(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
audio, model = self.create_inputs(root)
runner = FakeRunner()
result = transcribe_audio(
audio, model, root / "output", threads=12, runner=runner
)
command = runner.commands[0]
self.assertEqual(command[command.index("-t") + 1], "12")
metadata = json.loads(result.runtime_metadata.read_text())
self.assertEqual(metadata["threads"], 12)
self.assertEqual(metadata["threads_option"], "12")
def test_cli_forwards_explicit_thread_count_to_wrapper(self) -> None:
fake_result = Mock(
transcript_json=Path("output/transcript.json"), runtime_seconds=1.0
)
with patch.object(
transcribe_meeting, "transcribe_audio", return_value=fake_result
) as transcribe:
code = transcribe_meeting.main(
[
"meeting.wav",
"--model", "model.bin",
"--output-dir", "output",
"--threads", "6",
]
)
self.assertEqual(code, 0)
self.assertEqual(transcribe.call_args.kwargs["threads"], "6")
def test_compact_transcript_has_established_schema(self) -> None:
self.assertEqual(
compact_transcript(RAW_RESULT),
{
"text": "Hallo Welt",
"segments": [
{"id": 0, "start": 0.0, "end": 1.2, "text": "Hallo"},
{"id": 1, "start": 1.2, "end": 2.5, "text": "Welt"},
],
},
)
def test_success_preserves_raw_output_and_creates_metadata(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
audio, model = self.create_inputs(root)
runner = FakeRunner()
times = iter([10.0, 12.25])
result = transcribe_audio(
audio,
model,
root / "output",
"de",
runner=runner,
monotonic=lambda: next(times),
now=lambda: datetime(2026, 8, 21, 8, 30, tzinfo=timezone.utc),
)
self.assertEqual(json.loads(result.raw_output.read_text()), RAW_RESULT)
self.assertEqual(json.loads(result.transcript_json.read_text())["text"], "Hallo Welt")
self.assertEqual(
result.transcript_text.read_text(),
"[00:00:00.000 - 00:00:01.200] Hallo\n"
"[00:00:01.200 - 00:00:02.500] Welt\n",
)
metadata = json.loads(result.runtime_metadata.read_text())
self.assertEqual(metadata["backend"], "whisper.cpp")
self.assertEqual(metadata["language"], "de")
self.assertEqual(metadata["duration_seconds"], 2.5)
self.assertEqual(metadata["runtime_seconds"], 2.25)
self.assertEqual(metadata["timestamp"], "2026-08-21T08:30:00+00:00")
self.assertEqual(metadata["input_file"], str(audio.resolve()))
self.assertEqual(metadata["model"], str(model.resolve()))
def test_output_paths_are_deterministic(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
audio, model = self.create_inputs(root)
output = root / "chosen-output"
result = transcribe_audio(audio, model, output, runner=FakeRunner())
self.assertEqual(result.raw_output, output / "whisper_raw.json")
self.assertEqual(result.transcript_json, output / "transcript.json")
self.assertEqual(result.transcript_text, output / "transcript.txt")
self.assertEqual(result.runtime_metadata, output / "runtime_metadata.json")
self.assertEqual(
set(json.loads(result.runtime_metadata.read_text())["output_files"].values()),
{"whisper_raw.json", "transcript.json", "transcript.txt", "runtime_metadata.json"},
)
def test_backend_failure_raises_and_preserves_available_raw_output(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
audio, model = self.create_inputs(root)
output = root / "output"
with self.assertRaisesRegex(TranscriptionError, "exit code 7.*backend failure"):
transcribe_audio(audio, model, output, runner=FakeRunner(returncode=7))
self.assertEqual(json.loads((output / "whisper_raw.json").read_text()), RAW_RESULT)
self.assertFalse((output / "transcript.json").exists())
self.assertFalse((output / "runtime_metadata.json").exists())
def test_missing_backend_output_is_reported(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
audio, model = self.create_inputs(root)
with self.assertRaisesRegex(TranscriptionError, "without producing JSON"):
transcribe_audio(
audio, model, root / "output", runner=FakeRunner(write_raw=False)
)
if __name__ == "__main__":
unittest.main()