|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +from collections import defaultdict |
| 4 | +from typing import Any |
| 5 | + |
| 6 | + |
| 7 | +class PatternReconciler: |
| 8 | + """P0 reconciler that groups gene seeds by category into simple pattern families.""" |
| 9 | + |
| 10 | + def reconcile(self, gene_seeds: list[dict[str, Any]]) -> list[dict[str, Any]]: |
| 11 | + grouped: dict[str, list[dict[str, Any]]] = defaultdict(list) |
| 12 | + for seed in gene_seeds: |
| 13 | + grouped[seed["category"]].append(seed) |
| 14 | + |
| 15 | + patterns: list[dict[str, Any]] = [] |
| 16 | + for category, seeds in grouped.items(): |
| 17 | + pattern_id = f"rp:{category}:1" |
| 18 | + labels = [seed["label"] for seed in seeds] |
| 19 | + confidences = [float(seed.get("confidence", 0.0)) for seed in seeds] |
| 20 | + avg_confidence = sum(confidences) / len(confidences) if confidences else 0.0 |
| 21 | + source_observations: list[str] = [] |
| 22 | + for seed in seeds: |
| 23 | + source_observations.extend(seed.get("source_observations", [])) |
| 24 | + |
| 25 | + patterns.append( |
| 26 | + { |
| 27 | + "pattern_id": pattern_id, |
| 28 | + "category": category, |
| 29 | + "label": f"reconciled-{category}-pattern", |
| 30 | + "constituent_gene_seeds": [seed["gene_seed_id"] for seed in seeds], |
| 31 | + "source_observations": sorted(set(source_observations)), |
| 32 | + "reconciliation_note": f"Grouped {len(seeds)} gene seed(s) with shared category '{category}' into a P0 pattern family.", |
| 33 | + "conflict_resolved": False, |
| 34 | + "resolution_strategy": "group_by_category", |
| 35 | + "unified_intent": "; ".join(labels), |
| 36 | + "confidence": round(avg_confidence, 3), |
| 37 | + "tags": sorted({tag for seed in seeds for tag in seed.get("tags", [])}), |
| 38 | + } |
| 39 | + ) |
| 40 | + |
| 41 | + return sorted(patterns, key=lambda p: p["pattern_id"]) |
0 commit comments