Detect topic boundaries as continuous segments

This commit is contained in:
2026-07-21 17:21:01 +02:00
parent f234efca48
commit 889a4fe32d
+432 -252
View File
@@ -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: <input>_topics.json",
help="Output JSON file; default: <input>_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