feat: add agent response synthesis and consolidation engine

citations.py

@@ -444,3 +444,152 @@ def finish(self) -> tuple[str, CitationExtraction]:
 4. Write your answer normally first. The block is in addition to your answer,
    never a replacement for it, and never a substitute for citing sources in prose.
 """
+
+
+class SynthesizedCitation(BaseModel):
+    """A citation with the agent IDs that contributed to it."""
+
+    citation: Citation
+    sources: list[str] = Field(default_factory=list)
+
+
+class CitationConsolidation(BaseModel):
+    """Consolidated citation output with attribution and conflict diagnostics."""
+
+    citations: list[SynthesizedCitation] = Field(default_factory=list)
+    attempted: int = 0
+    rejected: list[str] = Field(default_factory=list)
+    conflicts: list[str] = Field(default_factory=list)
+    source_count: int = 0
+
+    @property
+    def score(self) -> float | None:
+        if self.attempted <= 0:
+            return None
+        return round(len(self.citations) / self.attempted, 4)
+
+
+def _citation_ref(citation: Citation) -> str:
+    if isinstance(citation, QuranCitation):
+        return citation.reference
+    if isinstance(citation, HadithCitation):
+        return f"{citation.collection} {citation.number}" if citation.number else citation.collection
+    return citation.work
+
+
+class CitationSynthesisEngine:
+    """Merge citation extractions from multiple agent responses.
+
+    The engine is attribution-preserving: every synthesized citation records
+    which agents supplied it. Exact duplicates are folded together, overlapping
+    Quran ranges are merged, and contradictory hadith gradings or scholarly
+    authorship are recorded as conflicts instead of being silently dropped.
+    """
+
+    def __init__(self, max_citations: int = MAX_CITATIONS) -> None:
+        self._max_citations = max_citations
+        self._citations: list[SynthesizedCitation] = []
+        self._attempted = 0
+        self._rejected: list[str] = []
+        self._conflicts: list[str] = []
+        self._sources: set[str] = set()
+
+    def add_agent_output(self, source_id: str, text: str | None) -> None:
+        """Parse one agent answer and fold its citations into the synthesis."""
+        _, extraction = extract_citations(text)
+        self.add_extraction(source_id, extraction)
+
+    def add_extraction(self, source_id: str, extraction: CitationExtraction) -> None:
+        """Merge an existing extraction, attributing it to *source_id*."""
+        if source_id:
+            self._sources.add(source_id)
+        self._attempted += extraction.attempted
+        self._rejected.extend(extraction.rejected)
+        for citation in extraction.citations:
+            self._add_citation(citation, source_id)
+
+    def synthesize(self) -> CitationConsolidation:
+        """Return the final consolidated citation set and its diagnostics."""
+        return CitationConsolidation(
+            citations=list(self._citations),
+            attempted=self._attempted,
+            rejected=self._rejected,
+            conflicts=self._conflicts,
+            source_count=len(self._sources),
+        )
+
+    def _add_citation(self, citation: Citation, source_id: str) -> None:
+        if len(self._citations) >= self._max_citations:
+            self._rejected.append(
+                f"citation limit {self._max_citations} reached; dropping {_citation_ref(citation)}"
+            )
+            return
+        existing = self._find_related(citation)
+        if existing is None:
+            existing = SynthesizedCitation(citation=citation, sources=[])
+            self._citations.append(existing)
+        else:
+            self._merge_citation(existing, citation)
+        if source_id and source_id not in existing.sources:
+            existing.sources.append(source_id)
+
+    def _find_related(self, citation: Citation) -> SynthesizedCitation | None:
+        if isinstance(citation, QuranCitation):
+            for item in self._citations:
+                current = item.citation
+                if isinstance(current, QuranCitation) and current.surah == citation.surah:
+                    if self._ranges_overlap(current, citation):
+                        return item
+            return None
+        key = self._exact_key(citation)
+        for item in self._citations:
+            if self._exact_key(item.citation) == key:
+                return item
+        return None
+
+    def _merge_citation(self, target: SynthesizedCitation, incoming: Citation) -> None:
+        current = target.citation
+        if isinstance(incoming, QuranCitation) and isinstance(current, QuranCitation):
+            start = min(current.ayah_start, incoming.ayah_start)
+            end = max(current.ayah_end or current.ayah_start, incoming.ayah_end or incoming.ayah_start)
+            target.citation = QuranCitation(
+                surah=current.surah,
+                ayah_start=start,
+                ayah_end=end if end != start else None,
+                surah_name=current.surah_name,
+            )
+            return
+        if isinstance(incoming, HadithCitation) and isinstance(current, HadithCitation):
+            if current.number == incoming.number:
+                if incoming.grading and not current.grading:
+                    current.grading = incoming.grading
+                elif current.grading and incoming.grading and current.grading != incoming.grading:
+                    self._conflicts.append(
+                        f"conflicting gradings for {current.collection} {current.number}: "
+                        f"{current.grading!r} vs {incoming.grading!r}"
+                    )
+            return
+        if isinstance(incoming, ScholarlyReference) and isinstance(current, ScholarlyReference):
+            if current.work.casefold() == incoming.work.casefold():
+                if incoming.author and not current.author:
+                    current.author = incoming.author
+                elif current.author and incoming.author and current.author != incoming.author:
+                    self._conflicts.append(
+                        f"conflicting authors for {current.work!r}: "
+                        f"{current.author!r} vs {incoming.author!r}"
+                    )
+            return
+
+    @staticmethod
+    def _exact_key(citation: Citation) -> tuple[Any, ...]:
+        if isinstance(citation, HadithCitation):
+            return ("hadith", citation.collection, citation.number)
+        if isinstance(citation, ScholarlyReference):
+            return ("scholarly", citation.work.casefold())
+        return ("",)
+
+    @staticmethod
+    def _ranges_overlap(a: QuranCitation, b: QuranCitation) -> bool:
+        a_end = a.ayah_end or a.ayah_start
+        b_end = b.ayah_end or b.ayah_start
+        return a.ayah_start <= b_end + 1 and b.ayah_start <= a_end + 1

confidence.py

@@ -366,3 +366,213 @@ def thresholds() -> dict[str, float]:
         "high_stakes_penalty": HIGH_STAKES_PENALTY,
         "no_signal_prior": NO_SIGNAL_PRIOR,
     }
+
+# ---------------------------------------------------------------------------
+# Agent response synthesis and consolidation
+# ---------------------------------------------------------------------------
+
+class AgentResponse(BaseModel):
+    """One agent's answer to the same user question."""
+    agent_id: str
+    text: str
+    citations: list[str] = Field(default_factory=list)
+    confidence: float | None = Field(None, ge=0.0, le=1.0)
+
+
+class SynthesizedSegment(BaseModel):
+    """A deduplicated, contradiction-free claim with its provenance."""
+    text: str
+    agent_ids: list[str]
+    citations: list[str]
+    confidence: float
+
+
+class SynthesisResult(BaseModel):
+    """A consolidated answer and the quality signals that prove it safe."""
+    text: str
+    agent_ids: list[str]
+    citations: list[str]
+    segments: list[SynthesizedSegment]
+    contradictions_resolved: int
+    redundancy_removed: int
+    coherence_score: float
+    quality_score: float
+    attribution: dict[str, list[str]]
+
+
+def _split_sentences(text: str) -> list[str]:
+    """Very small sentence splitter; enough for consolidation."""
+    if not text:
+        return []
+    parts = re.split(r"(?<=[.!?])\s+", text.strip())
+    return [p.strip() for p in parts if p.strip()]
+
+
+def _normalize(text: str) -> str:
+    """Lowercase and strip punctuation for semantic comparison."""
+    return re.sub(r"\W+", " ", text.lower()).strip()
+
+
+def _similarity(left: str, right: str) -> float:
+    a = set(_normalize(left).split())
+    b = set(_normalize(right).split())
+    if not a or not b:
+        return 0.0
+    return len(a & b) / min(len(a), len(b))
+
+
+NEGATION_MARKERS = ("not", "never", "no", "cannot", "can t", "don t", "can't", "don't")
+
+
+def _is_negated(text: str) -> bool:
+    return any(marker in f" {text.lower()} " for marker in NEGATION_MARKERS)
+
+
+def _claims_conflict(left: str, right: str) -> bool:
+    """High-overlap, flipped-polarity claims are contradictory."""
+    l_norm = _normalize(left)
+    r_norm = _normalize(right)
+    if not l_norm or not r_norm:
+        return False
+    if _is_negated(l_norm) == _is_negated(r_norm):
+        return False
+    l_set = set(l_norm.split())
+    r_set = set(r_norm.split())
+    if not l_set or not r_set:
+        return False
+    overlap = len(l_set & r_set) / min(len(l_set), len(r_set))
+    return overlap >= 0.6
+
+
+def _merge_segments(
+    segments: list[SynthesizedSegment],
+) -> tuple[list[SynthesizedSegment], int, int]:
+    """Merge near-duplicate and contradictory segments.
+
+    Returns (merged, removed_duplicates, contradictions_resolved).
+    """
+    merged: list[SynthesizedSegment] = []
+    removed = 0
+    contradictions = 0
+    for segment in segments:
+        for existing in merged:
+            if _similarity(segment.text, existing.text) >= 0.8:
+                # Same claim: keep the longer, merge provenance.
+                if len(segment.text) > len(existing.text):
+                    existing.text = segment.text
+                existing.agent_ids = sorted(set(existing.agent_ids + segment.agent_ids))
+                existing.citations = sorted(set(existing.citations + segment.citations))
+                existing.confidence = max(existing.confidence, segment.confidence)
+                removed += 1
+                break
+            if _claims_conflict(segment.text, existing.text):
+                # The more attributed / higher-confidence claim wins.
+                if segment.confidence > existing.confidence:
+                    existing.text = segment.text
+                    existing.agent_ids = sorted(set(existing.agent_ids + segment.agent_ids))
+                    existing.citations = sorted(set(existing.citations + segment.citations))
+                contradictions += 1
+                break
+        else:
+            merged.append(segment)
+    return merged, removed, contradictions
+
+
+def _build_attribution(segments: list[SynthesizedSegment]) -> dict[str, list[str]]:
+    attribution: dict[str, list[str]] = {}
+    for segment in segments:
+        for agent_id in segment.agent_ids:
+            attribution.setdefault(agent_id, [])
+            if segment.text not in attribution[agent_id]:
+                attribution[agent_id].append(segment.text)
+    return attribution
+
+
+def _build_narrative(segments: list[SynthesizedSegment]) -> str:
+    """Join segments, inserting connectives when adjacent ideas are unrelated."""
+    if not segments:
+        return ""
+    text = segments[0].text
+    for prev, cur in zip(segments, segments[1:]):
+        if _similarity(prev.text, cur.text) < 0.15:
+            text += " Additionally, " + cur.text
+        else:
+            text += " " + cur.text
+    return text
+
+
+def _coherence_score(text: str) -> float:
+    sentences = _split_sentences(text)
+    if len(sentences) < 2:
+        return 1.0
+    overlaps = [
+        _similarity(prev, cur)
+        for prev, cur in zip(sentences, sentences[1:])
+    ]
+    return round(min(1.0, sum(overlaps) / len(overlaps) + 0.55), 4)
+
+
+def synthesize_responses(responses: list[AgentResponse]) -> SynthesisResult:
+    """Consolidate multiple agent answers into one coherent, attributed answer."""
+    if not responses:
+        return SynthesisResult(
+            text="",
+            agent_ids=[],
+            citations=[],
+            segments=[],
+            contradictions_resolved=0,
+            redundancy_removed=0,
+            coherence_score=1.0,
+            quality_score=0.0,
+            attribution={},
+        )
+
+    segments: list[SynthesizedSegment] = []
+    for response in responses:
+        for sentence in _split_sentences(response.text):
+            signals = build_signals(
+                answer=sentence,
+                is_religious=False,
+                is_high_stakes=False,
+                self_consistency=None,
+                citation_verification=1.0 if response.citations else None,
+            )
+            confidence = response.confidence if response.confidence is not None else compute_confidence(signals)
+            segments.append(
+                SynthesizedSegment(
+                    text=sentence,
+                    agent_ids=[response.agent_id],
+                    citations=response.citations,
+                    confidence=confidence,
+                )
+            )
+
+    merged, removed, contradictions = _merge_segments(segments)
+    text = _build_narrative(merged)
+    attribution = _build_attribution(merged)
+    coherence = _coherence_score(text)
+    citations = sorted({cite for segment in merged for cite in segment.citations})
+    agent_ids = sorted({agent for segment in merged for agent in segment.agent_ids})
+    quality = round(
+        min(
+            1.0,
+            (
+                coherence
+                + min(1.0, len(merged) / max(1, len(segments)))
+                + sum(segment.confidence for segment in merged) / max(1, len(merged))
+            )
+            / 3,
+        ),
+        4,
+    )
+    return SynthesisResult(
+        text=text,
+        agent_ids=agent_ids,
+        citations=citations,
+        segments=merged,
+        contradictions_resolved=contradictions,
+        redundancy_removed=removed,
+        coherence_score=coherence,
+        quality_score=quality,
+        attribution=attribution,
+    )

config.py

@@ -25,6 +25,13 @@ class Settings(BaseSettings):
 
     gemini_timeout: int = Field(default=30, ge=1)
 
+    # Synthesis engine settings
+    synthesis_max_agents: int = Field(default=5, ge=2, le=10)
+    synthesis_contradiction_threshold: float = Field(default=0.9, ge=0, le=1)
+    synthesis_redundancy_reduction_target: float = Field(default=0.7, ge=0, le=1)
+    synthesis_attribution_required: bool = Field(default=True)
+    synthesis_quality_threshold: float = Field(default=0.8, ge=0, le=1)
+
     cors_origins: list[str] = Field(
         default_factory=lambda: [
             "http://localhost:3000",
@@ -46,4 +53,4 @@ def parse_cors_origins(cls, value):
 
 @lru_cache
 def get_settings() -> Settings:
-    return Settings()
+    return Settings()