From 889a4fe32d95e1ae534ff803a95fd6de5828c970 Mon Sep 17 00:00:00 2001 From: Martin Tazl Date: Tue, 21 Jul 2026 17:21:01 +0200 Subject: [PATCH] Detect topic boundaries as continuous segments --- .../segmentation/segment_topics.py | 684 +++++++++++------- 1 file changed, 432 insertions(+), 252 deletions(-) diff --git a/src/meeting_lab/segmentation/segment_topics.py b/src/meeting_lab/segmentation/segment_topics.py index 15606ef..33faeae 100644 --- a/src/meeting_lab/segmentation/segment_topics.py +++ b/src/meeting_lab/segmentation/segment_topics.py @@ -1,20 +1,23 @@ #!/usr/bin/env python3 """ -Segment a normalized meeting transcript into discussion topics using Ollama. +Detect topic boundaries in a normalized meeting transcript using Ollama. -The script numbers the transcript blocks and asks the language model only to -identify the thematic structure of the discussion. +The language model only identifies positions at which the discussion topic +changes. Python then converts those boundaries into continuous, complete and +non-overlapping segments. -It does not extract facts, decisions, todos, questions or summaries. +The script does not: -Examples: +- collect or merge topics, +- identify resumed topics, +- extract facts, +- extract decisions, +- extract todos, +- summarize the discussion. + +Example: python src/meeting_lab/segmentation/segment_topics.py \ samples/chunks/chunk_01_normalized.txt - - python src/meeting_lab/segmentation/segment_topics.py \ - samples/chunks/chunk_01_normalized.txt \ - --model qwen3:8b \ - --output experiments/topic-segmentation/chunk_01_topics.json """ from __future__ import annotations @@ -32,131 +35,209 @@ import requests DEFAULT_MODEL = "qwen3:8b" DEFAULT_ENDPOINT = "http://localhost:11434/api/generate" +DEFAULT_TARGET_BLOCK_CHARS = 600 SYSTEM_INSTRUCTION = """ -Du analysierst die thematische Struktur eines Meeting-Transkripts. +Du analysierst den chronologischen Verlauf eines Meeting-Transkripts. -Deine einzige Aufgabe ist die Themensegmentierung. +Deine einzige Aufgabe ist es, Stellen zu erkennen, an denen das Gespräch +inhaltlich zu einem anderen Thema wechselt. -Du sollst erkennen: +Du erhältst nummerierte Textblöcke. -1. wann ein Thema beginnt, -2. wann ein Thema endet, -3. wann zu einem anderen Thema gewechselt wird, -4. wann ein früheres Thema erneut aufgenommen wird. +Gib ausschließlich die Nummern der Blöcke zurück, nach denen ein echter +Themenwechsel stattfindet. + +Beispiel: + +Wenn die Blöcke 1 bis 4 ein Thema behandeln und mit Block 5 ein neues Thema +beginnt, gib 4 als Grenze zurück. Regeln: -1. Extrahiere keine Fakten. -2. Extrahiere keine Aufgaben. -3. Extrahiere keine Entscheidungen. -4. Extrahiere keine offenen Fragen. -5. Fasse die Diskussion nicht zusammen. -6. Bewerte die Aussagen nicht. -7. Erfinde keine Themen, die im Text nicht behandelt werden. -8. Verwende möglichst kurze und sachliche Themenbezeichnungen. -9. Jeder Block muss genau einem Thema zugeordnet werden. -10. Mehrere direkt aufeinanderfolgende Blöcke desselben Themas bilden ein Segment. -11. Wird ein früheres Thema später wieder aufgenommen, verwende dieselbe topic_id. -12. Kleine Rückfragen oder kurze Ergänzungen bleiben beim aktuellen Thema, - sofern sie kein eigenständiges Thema eröffnen. -13. Organisatorische Übergänge dürfen als eigenes Thema erfasst werden, wenn - sie einen erkennbaren Gesprächsabschnitt bilden. -14. Gib ausschließlich gültiges JSON aus. -15. Gib kein Markdown und keine Erläuterungen außerhalb des JSON aus. +1. Gib nur echte inhaltliche Themenwechsel zurück. +2. Kleine Rückfragen sind kein Themenwechsel. +3. Kurze Ergänzungen sind kein Themenwechsel. +4. Sprecherwechsel sind kein Themenwechsel. +5. Zustimmung oder Ablehnung sind kein Themenwechsel. +6. Eine kurze Abschweifung ist nur dann ein eigenes Segment, wenn sie einen + erkennbaren Gesprächsabschnitt bildet. +7. Organisatorische Einleitungen oder Verabschiedungen dürfen eigene Segmente + bilden. +8. Fasse keine Inhalte zusammen. +9. Vergib keine Themenbezeichnungen. +10. Extrahiere keine Fakten, Aufgaben, Entscheidungen oder Fragen. +11. Die letzte Blocknummer darf nicht als Themenwechsel ausgegeben werden. +12. Die Blocknummern müssen aufsteigend und eindeutig sein. +13. Gib ausschließlich gültiges JSON aus. +14. Gib kein Markdown und keine zusätzlichen Erläuterungen aus. """.strip() -OUTPUT_SCHEMA = { - "topics": [ - { - "topic_id": "topic_001", - "title": "Kurze sachliche Themenbezeichnung", - "segments": [ - { - "start_block": 1, - "end_block": 5, - } - ], +TOPIC_CHANGE_JSON_SCHEMA: dict[str, Any] = { + "type": "object", + "properties": { + "topic_changes": { + "type": "array", + "items": { + "type": "integer", + "minimum": 1, + }, } - ] + }, + "required": ["topic_changes"], + "additionalProperties": False, } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( - description="Segment a meeting transcript into discussion topics." + description=( + "Detect topic boundaries in a normalized meeting transcript." + ) ) + parser.add_argument( "input_file", type=Path, - help="Normalized transcript or transcript chunk", + help="Normalized transcript file", ) + parser.add_argument( "-o", "--output", type=Path, - help="Output JSON file; default: _topics.json", + help="Output JSON file; default: _segments.json", ) + parser.add_argument( "--model", default=DEFAULT_MODEL, help=f"Ollama model name (default: {DEFAULT_MODEL})", ) + parser.add_argument( "--endpoint", default=DEFAULT_ENDPOINT, help=f"Ollama generate endpoint (default: {DEFAULT_ENDPOINT})", ) + parser.add_argument( "--timeout", type=int, default=1800, help="HTTP timeout in seconds (default: 1800)", ) + parser.add_argument( "--temperature", type=float, default=0.0, help="Sampling temperature (default: 0.0)", ) + + parser.add_argument( + "--target-block-chars", + type=int, + default=DEFAULT_TARGET_BLOCK_CHARS, + help=( + "Approximate size of analysis blocks in characters " + f"(default: {DEFAULT_TARGET_BLOCK_CHARS})" + ), + ) + return parser.parse_args() -def split_into_blocks(text: str) -> list[str]: +def split_into_blocks( + text: str, + target_chars: int = DEFAULT_TARGET_BLOCK_CHARS, +) -> list[str]: """ - Split a transcript into stable, numbered discussion blocks. + Combine short transcript utterances into larger analysis blocks. - Blank-line-separated paragraphs are preferred. If the transcript contains - no blank lines, individual non-empty lines are used as blocks. + Whisper-derived transcripts often contain one short utterance per + paragraph. Adjacent utterances are therefore grouped until the approximate + target size is reached. """ - normalized = text.replace("\r\n", "\n").replace("\r", "\n").strip() + if target_chars < 1: + raise ValueError( + "target_chars must be greater than zero." + ) - paragraphs = [ + normalized = ( + text.replace("\r\n", "\n") + .replace("\r", "\n") + .strip() + ) + + units = [ part.strip() for part in re.split(r"\n\s*\n+", normalized) if part.strip() ] - if len(paragraphs) > 1: - return paragraphs + if not units: + units = [ + line.strip() + for line in normalized.splitlines() + if line.strip() + ] - return [line.strip() for line in normalized.splitlines() if line.strip()] + blocks: list[str] = [] + current_units: list[str] = [] + current_length = 0 + + for unit in units: + separator_length = 1 if current_units else 0 + + projected_length = ( + current_length + + separator_length + + len(unit) + ) + + if ( + current_units + and projected_length > target_chars + ): + blocks.append("\n".join(current_units)) + + current_units = [unit] + current_length = len(unit) + + else: + current_units.append(unit) + current_length = projected_length + + if current_units: + blocks.append("\n".join(current_units)) + + return blocks -def render_numbered_blocks(blocks: list[str]) -> str: +def render_numbered_blocks( + blocks: list[str], +) -> str: return "\n\n".join( f"[BLOCK {number}]\n{block}" - for number, block in enumerate(blocks, start=1) + for number, block in enumerate( + blocks, + start=1, + ) ) -def build_prompt(source_name: str, blocks: list[str]) -> str: - schema_text = json.dumps(OUTPUT_SCHEMA, ensure_ascii=False, indent=2) - numbered_transcript = render_numbered_blocks(blocks) +def build_prompt( + source_name: str, + blocks: list[str], +) -> str: + numbered_transcript = render_numbered_blocks( + blocks + ) - return f"""{SYSTEM_INSTRUCTION} + return f"""Bestimme die Stellen, an denen sich das Gesprächsthema ändert. Quelldatei: {source_name} @@ -164,24 +245,19 @@ Quelldatei: Anzahl der Blöcke: {len(blocks)} -Erwartete JSON-Struktur: -{schema_text} +Erwartete Ausgabe: -Zusätzliche Anforderungen: +{{ + "topic_changes": [4, 12, 18] +}} -- Alle topic_id-Werte müssen dem Muster topic_001, topic_002 usw. folgen. -- Die Nummerierung beginnt mit topic_001. -- Die topic_id-Werte müssen innerhalb der Ausgabe eindeutig sein. -- start_block und end_block beziehen sich auf die angegebenen Blocknummern. -- start_block darf nicht größer als end_block sein. -- Kein Block darf fehlen. -- Kein Block darf mehreren Themen gleichzeitig zugeordnet werden. -- Direkt aufeinanderfolgende Bereiche desselben Themas sollen als ein Segment - ausgegeben werden. -- Nicht zusammenhängende Bereiche desselben Themas sollen als mehrere Segmente - innerhalb desselben Themas ausgegeben werden. -- Themenbezeichnungen sollen knapp sein und keine Zusammenfassung enthalten. -- Der Schlüssel "topics" muss immer vorhanden sein. +Bedeutung: + +- Der Wert 4 bedeutet, dass nach Block 4 ein neues Thema beginnt. +- Der Wert 12 bedeutet, dass nach Block 12 ein neues Thema beginnt. +- Der letzte Block darf nicht in der Liste stehen. +- Falls es keinen Themenwechsel gibt, gib eine leere Liste zurück. +- Gib nur eindeutige, aufsteigend sortierte Blocknummern zurück. TRANSKRIPT: --- BEGINN TRANSKRIPT --- @@ -199,20 +275,26 @@ def call_ollama( ) -> tuple[str, dict[str, Any]]: payload = { "model": model, + "system": SYSTEM_INSTRUCTION, "prompt": prompt, "stream": False, - "format": "json", + "format": TOPIC_CHANGE_JSON_SCHEMA, + "think": False, "options": { "temperature": temperature, + "seed": 42, + "num_ctx": 32768, }, } started = time.perf_counter() + response = requests.post( endpoint, json=payload, timeout=timeout, ) + elapsed = time.perf_counter() - started response.raise_for_status() @@ -220,197 +302,226 @@ def call_ollama( data = response.json() response_text = data.get("response") - if not isinstance(response_text, str) or not response_text.strip(): - raise ValueError("Ollama returned no usable response text.") + if ( + not isinstance(response_text, str) + or not response_text.strip() + ): + raise ValueError( + "Ollama returned no usable response text." + ) metadata = { "model": data.get("model", model), - "elapsed_seconds": round(elapsed, 3), - "total_duration_ns": data.get("total_duration"), - "load_duration_ns": data.get("load_duration"), - "prompt_eval_count": data.get("prompt_eval_count"), - "prompt_eval_duration_ns": data.get("prompt_eval_duration"), - "eval_count": data.get("eval_count"), - "eval_duration_ns": data.get("eval_duration"), + "elapsed_seconds": round( + elapsed, + 3, + ), + "total_duration_ns": data.get( + "total_duration" + ), + "load_duration_ns": data.get( + "load_duration" + ), + "prompt_eval_count": data.get( + "prompt_eval_count" + ), + "prompt_eval_duration_ns": data.get( + "prompt_eval_duration" + ), + "eval_count": data.get( + "eval_count" + ), + "eval_duration_ns": data.get( + "eval_duration" + ), } return response_text.strip(), metadata -def parse_json_response(text: str) -> dict[str, Any]: +def parse_json_response( + text: str, +) -> dict[str, Any]: try: parsed = json.loads(text) + except json.JSONDecodeError: - # Fallback for models that wrap JSON in Markdown or add text around it. - match = re.search(r"\{.*\}", text, flags=re.DOTALL) + match = re.search( + r"\{.*\}", + text, + flags=re.DOTALL, + ) + if not match: raise - parsed = json.loads(match.group(0)) + + parsed = json.loads( + match.group(0) + ) if not isinstance(parsed, dict): raise ValueError( - "The model response is valid JSON but not a JSON object." + "The model response is not a JSON object." ) return parsed -def validate_topic_id(topic_id: Any) -> str: - if not isinstance(topic_id, str): - raise ValueError("Every topic must contain a string topic_id.") - - if not re.fullmatch(r"topic_\d{3}", topic_id): - raise ValueError( - f"Invalid topic_id {topic_id!r}; expected format topic_001." - ) - - return topic_id - - -def validate_segment( - segment: Any, - topic_id: str, - block_count: int, -) -> dict[str, int]: - if not isinstance(segment, dict): - raise ValueError( - f"Topic {topic_id} contains a segment that is not an object." - ) - - start_block = segment.get("start_block") - end_block = segment.get("end_block") - - if not isinstance(start_block, int) or isinstance(start_block, bool): - raise ValueError( - f"Topic {topic_id} contains an invalid start_block." - ) - - if not isinstance(end_block, int) or isinstance(end_block, bool): - raise ValueError( - f"Topic {topic_id} contains an invalid end_block." - ) - - if start_block < 1 or end_block > block_count: - raise ValueError( - f"Topic {topic_id} references blocks outside 1..{block_count}." - ) - - if start_block > end_block: - raise ValueError( - f"Topic {topic_id} contains start_block > end_block." - ) - - return { - "start_block": start_block, - "end_block": end_block, - } - - -def normalize_topics( +def validate_topic_changes( result: dict[str, Any], block_count: int, -) -> list[dict[str, Any]]: - raw_topics = result.get("topics") +) -> list[int]: + raw_changes = result.get("topic_changes") - if not isinstance(raw_topics, list): - raise ValueError("Model output does not contain a topics list.") + if not isinstance(raw_changes, list): + raise ValueError( + "Model output does not contain " + "a topic_changes list." + ) - topics: list[dict[str, Any]] = [] - seen_topic_ids: set[str] = set() + changes: list[int] = [] - for raw_topic in raw_topics: - if not isinstance(raw_topic, dict): - raise ValueError("Every topic must be a JSON object.") + for value in raw_changes: + if ( + not isinstance(value, int) + or isinstance(value, bool) + ): + raise ValueError( + "Every topic change must be " + "an integer block number." + ) - topic_id = validate_topic_id(raw_topic.get("topic_id")) + if value < 1: + raise ValueError( + f"Invalid topic change: {value}" + ) - if topic_id in seen_topic_ids: - raise ValueError(f"Duplicate topic_id: {topic_id}") - seen_topic_ids.add(topic_id) + if value >= block_count: + raise ValueError( + f"Topic change {value} is invalid. " + f"The last block is {block_count}." + ) - title = raw_topic.get("title") - if not isinstance(title, str) or not title.strip(): - raise ValueError(f"Topic {topic_id} has no usable title.") + changes.append(value) - raw_segments = raw_topic.get("segments") - if not isinstance(raw_segments, list) or not raw_segments: - raise ValueError(f"Topic {topic_id} has no segments.") + if len(changes) != len(set(changes)): + raise ValueError( + "The model returned duplicate " + "topic changes." + ) - segments = [ - validate_segment(segment, topic_id, block_count) - for segment in raw_segments - ] - segments.sort(key=lambda item: item["start_block"]) + if changes != sorted(changes): + raise ValueError( + "The topic changes are not " + "sorted in ascending order." + ) - topics.append( + return changes + + +def build_segments( + topic_changes: list[int], + block_count: int, +) -> list[dict[str, int | str]]: + """ + Convert topic boundaries into continuous and non-overlapping segments. + """ + segments: list[dict[str, int | str]] = [] + + start_block = 1 + + for number, end_block in enumerate( + topic_changes, + start=1, + ): + segments.append( { - "topic_id": topic_id, - "title": title.strip(), - "segments": segments, + "segment_id": ( + f"segment_{number:03d}" + ), + "start_block": start_block, + "end_block": end_block, } ) - if not topics: - raise ValueError("The model returned no topics.") + start_block = end_block + 1 - topics.sort( - key=lambda topic: min( - segment["start_block"] for segment in topic["segments"] - ) + final_number = len(segments) + 1 + + segments.append( + { + "segment_id": ( + f"segment_{final_number:03d}" + ), + "start_block": start_block, + "end_block": block_count, + } ) - return topics + return segments -def validate_block_coverage( - topics: list[dict[str, Any]], +def validate_segments( + segments: list[dict[str, int | str]], block_count: int, ) -> None: - assignments: dict[int, str] = {} - - for topic in topics: - topic_id = topic["topic_id"] - - for segment in topic["segments"]: - for block_number in range( - segment["start_block"], - segment["end_block"] + 1, - ): - previous_topic = assignments.get(block_number) - - if previous_topic is not None: - raise ValueError( - f"Block {block_number} is assigned to both " - f"{previous_topic} and {topic_id}." - ) - - assignments[block_number] = topic_id - - missing_blocks = [ - block_number - for block_number in range(1, block_count + 1) - if block_number not in assignments - ] - - if missing_blocks: - formatted = ", ".join(str(number) for number in missing_blocks) + if not segments: raise ValueError( - f"The following blocks were not assigned to a topic: {formatted}" + "No segments were created." + ) + + expected_start = 1 + + for segment in segments: + start_block = segment["start_block"] + end_block = segment["end_block"] + + if not isinstance(start_block, int): + raise ValueError( + "Invalid segment start block." + ) + + if not isinstance(end_block, int): + raise ValueError( + "Invalid segment end block." + ) + + if start_block != expected_start: + raise ValueError( + "Segments are not continuous." + ) + + if end_block < start_block: + raise ValueError( + "A segment ends before it starts." + ) + + expected_start = end_block + 1 + + if expected_start != block_count + 1: + raise ValueError( + "Segments do not cover all blocks." ) def build_output( source_name: str, blocks: list[str], - topics: list[dict[str, Any]], + topic_changes: list[int], + segments: list[dict[str, int | str]], metadata: dict[str, Any], + target_block_chars: int, ) -> dict[str, Any]: return { "source": { "file": source_name, "block_count": len(blocks), + "target_block_chars": ( + target_block_chars + ), }, - "topics": topics, + "topic_changes": topic_changes, + "segments": segments, "_run": metadata, } @@ -419,8 +530,15 @@ def write_raw_response( output_path: Path, raw_text: str, ) -> Path: - raw_path = output_path.with_suffix(".raw.txt") - raw_path.write_text(raw_text + "\n", encoding="utf-8") + raw_path = output_path.with_suffix( + ".raw.txt" + ) + + raw_path.write_text( + raw_text + "\n", + encoding="utf-8", + ) + return raw_path @@ -430,7 +548,8 @@ def main() -> int: try: if not args.input_file.is_file(): raise FileNotFoundError( - f"Input file not found: {args.input_file}" + f"Input file not found: " + f"{args.input_file}" ) transcript = args.input_file.read_text( @@ -438,28 +557,50 @@ def main() -> int: ).strip() if not transcript: - raise ValueError("The input file is empty.") + raise ValueError( + "The input file is empty." + ) - blocks = split_into_blocks(transcript) + blocks = split_into_blocks( + transcript, + target_chars=( + args.target_block_chars + ), + ) if not blocks: - raise ValueError("No discussion blocks could be created.") + raise ValueError( + "No analysis blocks could " + "be created." + ) - output_path = args.output or args.input_file.with_name( - f"{args.input_file.stem}_topics.json" + output_path = ( + args.output + or args.input_file.with_name( + f"{args.input_file.stem}" + "_segments.json" + ) + ) + + output_path.parent.mkdir( + parents=True, + exist_ok=True, ) - output_path.parent.mkdir(parents=True, exist_ok=True) prompt = build_prompt( source_name=args.input_file.name, blocks=blocks, ) - print(f"Model: {args.model}") - print(f"Input: {args.input_file}") - print(f"Blocks: {len(blocks)}") - print(f"Output: {output_path}") - print("Segmenting discussion topics ...") + print(f"Model: {args.model}") + print(f"Input: {args.input_file}") + print(f"Blocks: {len(blocks)}") + print( + "Block target: " + f"{args.target_block_chars} characters" + ) + print(f"Output: {output_path}") + print("Detecting topic boundaries ...") raw_text, metadata = call_ollama( endpoint=args.endpoint, @@ -470,28 +611,53 @@ def main() -> int: ) try: - parsed = parse_json_response(raw_text) - topics = normalize_topics( - result=parsed, + parsed = parse_json_response( + raw_text + ) + + topic_changes = ( + validate_topic_changes( + result=parsed, + block_count=len(blocks), + ) + ) + + segments = build_segments( + topic_changes=topic_changes, block_count=len(blocks), ) - validate_block_coverage( - topics=topics, + + validate_segments( + segments=segments, block_count=len(blocks), ) - except (json.JSONDecodeError, ValueError) as exc: - raw_path = write_raw_response(output_path, raw_text) + + except ( + json.JSONDecodeError, + ValueError, + ) as exc: + raw_path = write_raw_response( + output_path, + raw_text, + ) + raise ValueError( - f"Model output could not be validated. " - f"Raw output saved to: {raw_path}. " + "Model output could not be " + "validated. " + f"Raw output saved to: " + f"{raw_path}. " f"Reason: {exc}" ) from exc result = build_output( source_name=args.input_file.name, blocks=blocks, - topics=topics, + topic_changes=topic_changes, + segments=segments, metadata=metadata, + target_block_chars=( + args.target_block_chars + ), ) output_path.write_text( @@ -504,40 +670,54 @@ def main() -> int: encoding="utf-8", ) - segment_count = sum( - len(topic["segments"]) for topic in topics + print( + f"Done in " + f"{metadata['elapsed_seconds']:.1f} " + "seconds." ) - resumed_topic_count = sum( - 1 for topic in topics if len(topic["segments"]) > 1 + print( + f"Topic changes: " + f"{len(topic_changes)}" + ) + print( + f"Segments: " + f"{len(segments)}" ) - - print(f"Done in {metadata['elapsed_seconds']:.1f} seconds.") - print(f"Topics: {len(topics)}") - print(f"Segments: {segment_count}") - print(f"Resumed topics: {resumed_topic_count}") return 0 except requests.ConnectionError: print( - "Error: Ollama is not reachable. Is `ollama serve` running?", + "Error: Ollama is not reachable. " + "Is `ollama serve` running?", file=sys.stderr, ) return 1 + except requests.Timeout: print( "Error: The Ollama request timed out.", file=sys.stderr, ) return 1 + except requests.HTTPError as exc: print( - f"Error: Ollama returned an HTTP error: {exc}", + "Error: Ollama returned an HTTP " + f"error: {exc}", file=sys.stderr, ) return 1 - except (OSError, UnicodeError, ValueError) as exc: - print(f"Error: {exc}", file=sys.stderr) + + except ( + OSError, + UnicodeError, + ValueError, + ) as exc: + print( + f"Error: {exc}", + file=sys.stderr, + ) return 1