martin/meeting-lab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement Semantic Consolidator V0

Browse Source 
- add deterministic canonicalization support for extraction items
- add facts-only semantic consolidation using local Ollama
- preserve source evidence and validate complete fact coverage
- add conservative merge rules and non-LLM tests
- record the first validated real-life consolidation benchmark
- document current scope, limitations and next evaluation step
This commit is contained in:
admin
2026-07-31 11:25:46 +02:00
parent 6e34334506
commit 90aa34d5d0
20 changed files with 5517 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_canonicalize.py
+169
View File
@@ -0,0 +1,169 @@
import json
import tempfile
import unittest
from pathlib import Path
from src.meeting_lab.consolidation.canonicalize import (
canonicalize_extractions,
find_extraction_files,
load_json_object,
)
EMPTY_EXTRACTION = {
"facts": [],
"decisions": [],
"todos": [],
"questions": [],
"positions": [],
"technical": [],
}
def write_extraction(directory: Path, name: str, data: dict) -> Path:
path = directory / name
path.write_text(json.dumps(data), encoding="utf-8")
return path
class CanonicalizeTests(unittest.TestCase):
def test_stable_file_ordering(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
write_extraction(root, "chunk_10_extraction.json", EMPTY_EXTRACTION)
write_extraction(root, "chunk_02_extraction.json", EMPTY_EXTRACTION)
write_extraction(root, "chunk_01_extraction.json", EMPTY_EXTRACTION)
self.assertEqual(
[path.name for path in find_extraction_files(root)],
[
"chunk_01_extraction.json",
"chunk_02_extraction.json",
"chunk_10_extraction.json",
],
)
def test_required_category_validation(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
data = dict(EMPTY_EXTRACTION)
data.pop("technical")
write_extraction(root, "chunk_01_extraction.json", data)
with self.assertRaisesRegex(ValueError, "technical"):
canonicalize_extractions(root)
def test_stable_item_ids_and_string_parsing(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
data = dict(EMPTY_EXTRACTION)
data["facts"] = ["Ada | Revenue increased | clear | Revenue increased"]
data["decisions"] = ["Ship the patch | Agreed to ship"]
data["todos"] = ["Update docs | Mira | Friday | I will update docs"]
data["questions"] = ["Which plan? | Which plan should we use?"]
data["positions"] = ["Kai | Prefer option B | I prefer option B"]
data["technical"] = ["API | Uses v2 mapping | clear | API uses v2"]
write_extraction(root, "chunk_01_extraction.json", data)
output = canonicalize_extractions(root)
items = output["items"]
self.assertEqual(
[item["item_id"] for item in items],
[
"fact_0001",
"decision_0001",
"action_item_0001",
"open_question_0001",
"position_0001",
"technical_detail_0001",
],
)
fact = items[0]
self.assertEqual(fact["speaker"], "Ada")
self.assertEqual(fact["text"], "Revenue increased")
self.assertEqual(fact["status"], "clear")
todo = items[2]
self.assertEqual(todo["responsible"], "Mira")
self.assertEqual(todo["deadline"], "Friday")
self.assertEqual(todo["evidence"], "I will update docs")
def test_source_reference_preservation(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
data = dict(EMPTY_EXTRACTION)
data["decisions"] = ["Decision one | Evidence one"]
write_extraction(root, "chunk_01_extraction.json", data)
item = canonicalize_extractions(root)["items"][0]
self.assertEqual(item["source_file"], "chunk_01_extraction.json")
self.assertEqual(item["source_index"], 0)
self.assertEqual(item["original_value"], "Decision one | Evidence one")
self.assertEqual(
item["source_references"],
[
{
"source_file": "chunk_01_extraction.json",
"source_index": 0,
"evidence": "Evidence one",
"original_value": "Decision one | Evidence one",
}
],
)
def test_exact_duplicate_handling(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
first = dict(EMPTY_EXTRACTION)
second = dict(EMPTY_EXTRACTION)
first["decisions"] = ["Same decision | Same evidence"]
second["decisions"] = ["Same decision | Same evidence"]
write_extraction(root, "chunk_01_extraction.json", first)
write_extraction(root, "chunk_02_extraction.json", second)
output = canonicalize_extractions(root)
self.assertEqual(len(output["items"]), 1)
self.assertEqual(output["stats"]["exact_duplicates_merged"], 1)
self.assertEqual(output["items"][0]["duplicate_count"], 2)
self.assertEqual(len(output["items"][0]["source_references"]), 2)
def test_no_merging_of_merely_similar_statements(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
data = dict(EMPTY_EXTRACTION)
data["decisions"] = [
"Ship the patch Friday | Agreed to ship Friday",
"Ship the patch next week | Agreed to ship next week",
]
write_extraction(root, "chunk_01_extraction.json", data)
output = canonicalize_extractions(root)
self.assertEqual(len(output["items"]), 2)
self.assertEqual(output["stats"]["exact_duplicates_merged"], 0)
def test_invalid_json_handling(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
(root / "chunk_01_extraction.json").write_text("{invalid", encoding="utf-8")
with self.assertRaisesRegex(ValueError, "Invalid JSON"):
load_json_object(root / "chunk_01_extraction.json")
def test_empty_categories(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
write_extraction(root, "chunk_01_extraction.json", EMPTY_EXTRACTION)
output = canonicalize_extractions(root)
self.assertEqual(output["items"], [])
self.assertEqual(output["stats"]["input_item_count"], 0)
self.assertEqual(output["stats"]["output_item_count"], 0)
if __name__ == "__main__":
unittest.main()
tests/test_consolidate_facts.py
+215
View File
@@ -0,0 +1,215 @@
import unittest
from src.meeting_lab.consolidation.consolidate_facts import (
ConsolidationValidationError,
DEFAULT_NUM_PREDICT,
build_consolidated_output,
build_ollama_payload,
parse_model_json,
validate_consolidated_output,
validate_model_groups,
)
def canonicalized_fixture():
fact_one = {
"item_id": "fact_0001",
"category": "fact",
"text": "The lead maintains the list.",
"evidence": "lead maintains the list",
"source_file": "chunk_01_extraction.json",
"source_index": 0,
"original_value": "The lead maintains the list. | evidence",
"source_references": [
{
"source_file": "chunk_01_extraction.json",
"source_index": 0,
"evidence": "lead maintains the list",
"original_value": "The lead maintains the list. | evidence",
}
],
"duplicate_count": 1,
}
fact_two = {
"item_id": "fact_0002",
"category": "fact",
"text": "The head maintains the project list.",
"evidence": "head maintains the project list",
"source_file": "chunk_02_extraction.json",
"source_index": 0,
"original_value": "The head maintains the project list. | evidence",
"source_references": [
{
"source_file": "chunk_02_extraction.json",
"source_index": 0,
"evidence": "head maintains the project list",
"original_value": "The head maintains the project list. | evidence",
}
],
"duplicate_count": 1,
}
decision = {
"item_id": "decision_0001",
"category": "decision",
"text": "Ship it.",
"evidence": "Agreed.",
"source_file": "chunk_01_extraction.json",
"source_index": 0,
"original_value": "Ship it. | Agreed.",
"source_references": [],
"duplicate_count": 1,
}
return {
"schema_version": "1",
"source_files": ["chunk_01_extraction.json", "chunk_02_extraction.json"],
"items": [fact_one, fact_two, decision],
}
class ConsolidateFactsTests(unittest.TestCase):
def test_payload_construction_disables_streaming_and_thinking_by_default(self):
payload = build_ollama_payload(
model="qwen3.5:9B",
prompt="prompt",
num_ctx=32768,
num_predict=DEFAULT_NUM_PREDICT,
think=False,
)
self.assertEqual(payload["model"], "qwen3.5:9B")
self.assertEqual(payload["prompt"], "prompt")
self.assertIs(payload["stream"], False)
self.assertIs(payload["think"], False)
self.assertEqual(payload["format"], "json")
self.assertEqual(payload["options"]["temperature"], 0.0)
self.assertEqual(payload["options"]["num_ctx"], 32768)
self.assertEqual(payload["options"]["num_predict"], DEFAULT_NUM_PREDICT)
def test_payload_construction_can_enable_thinking_explicitly(self):
payload = build_ollama_payload(
model="qwen3.5:9B",
prompt="prompt",
num_ctx=16384,
num_predict=1024,
think=True,
)
self.assertIs(payload["think"], True)
self.assertEqual(payload["options"]["num_ctx"], 16384)
self.assertEqual(payload["options"]["num_predict"], 1024)
def test_grouping_validation_accepts_complete_singletons(self):
groups = validate_model_groups(
{
"groups": [
{
"canonical_text": "The lead maintains the list.",
"source_item_ids": ["fact_0001"],
"merge_reason": "Singleton.",
},
{
"canonical_text": "The head maintains the project list.",
"source_item_ids": ["fact_0002"],
"merge_reason": "Singleton.",
},
]
},
{"fact_0001", "fact_0002"},
)
self.assertEqual(len(groups), 2)
def test_no_missing_source_ids(self):
with self.assertRaisesRegex(ConsolidationValidationError, "Missing"):
validate_model_groups(
{
"groups": [
{
"canonical_text": "Only one.",
"source_item_ids": ["fact_0001"],
"merge_reason": "Singleton.",
}
]
},
{"fact_0001", "fact_0002"},
)
def test_no_duplicate_source_ids(self):
with self.assertRaisesRegex(ConsolidationValidationError, "multiple"):
validate_model_groups(
{
"groups": [
{
"canonical_text": "One.",
"source_item_ids": ["fact_0001"],
"merge_reason": "Singleton.",
},
{
"canonical_text": "Again.",
"source_item_ids": ["fact_0001"],
"merge_reason": "Singleton.",
},
]
},
{"fact_0001"},
)
def test_merged_group_validation(self):
groups = validate_model_groups(
{
"groups": [
{
"canonical_text": "The lead maintains the project list.",
"source_item_ids": ["fact_0001", "fact_0002"],
"merge_reason": "Same proposition.",
}
]
},
{"fact_0001", "fact_0002"},
)
output = build_consolidated_output(canonicalized_fixture(), groups)
validate_consolidated_output(canonicalized_fixture(), output)
self.assertEqual(output["items"][0]["source_item_ids"], ["fact_0001", "fact_0002"])
def test_preservation_of_non_fact_categories(self):
fixture = canonicalized_fixture()
groups = validate_model_groups(
{
"groups": [
{
"canonical_text": "The lead maintains the project list.",
"source_item_ids": ["fact_0001", "fact_0002"],
"merge_reason": "Same proposition.",
}
]
},
{"fact_0001", "fact_0002"},
)
output = build_consolidated_output(fixture, groups)
self.assertEqual(output["items"][1:], fixture["items"][2:])
def test_invalid_model_json(self):
with self.assertRaisesRegex(ConsolidationValidationError, "Invalid model JSON"):
parse_model_json("{invalid")
def test_unknown_source_item_ids(self):
with self.assertRaisesRegex(ConsolidationValidationError, "unknown"):
validate_model_groups(
{
"groups": [
{
"canonical_text": "Unknown.",
"source_item_ids": ["fact_9999"],
"merge_reason": "Bad ID.",
}
]
},
{"fact_0001"},
)
if __name__ == "__main__":
unittest.main()