feat: Implement document synthesis endpoint with hierarchical summarization

app.py

@@ -9,9 +9,11 @@ Provides a web UI for the FractalStat RAG system with GPU acceleration.
 import gradio as gr
 import os
 import time
+from typing import List, Dict, Any
 
 from warbler_cda.answer_generator import AnswerGenerator
 from warbler_cda.remote_pack_loader import RemotePackLoader
+from warbler_cda.summarization_ladder import SummarizationLadder
 
 # Import the HuggingFace Spaces GPU decorator
 try:
@@ -401,6 +403,78 @@ def get_system_stats() -> str:
     return output
 
 
+def synthesize_document(
+    document_text: str,
+    micro_window_size: int = 5,
+    macro_trigger_count: int = 3,
+) -> str:
+    """Generate a hierarchical synthesis artifact for a document-sized input."""
+    cleaned = document_text.strip()
+    if not cleaned:
+        return "Please provide document text to synthesize."
+
+    fragments: List[Dict[str, Any]] = []
+    units = [part.strip() for part in cleaned.splitlines() if part.strip()]
+    if len(units) < 2:
+        units = [sentence.strip() for sentence in cleaned.split(".") if sentence.strip()]
+
+    for index, unit in enumerate(units, start=1):
+        fragments.append(
+            {
+                "id": f"fragment_{index}",
+                "text": unit,
+                "heat": min(1.0, 0.4 + (len(unit) / 400.0)),
+            }
+        )
+
+    if not fragments:
+        return "Unable to derive synthesis fragments from the provided text."
+
+    ladder = SummarizationLadder(
+        config={
+            "micro_window_size": micro_window_size,
+            "macro_trigger_count": macro_trigger_count,
+        }
+    )
+    report = ladder.process_fragments(fragments)
+
+    lines = ["## Synthesis Output", "", f"**Input Fragments:** {len(fragments)}", ""]
+
+    if ladder.macro_distillations:
+        latest_macro = ladder.macro_distillations[-1]
+        lines.extend([
+            "### Macro Distillation",
+            "",
+            latest_macro.distilled_essence,
+            "",
+        ])
+
+    if ladder.micro_summaries:
+        lines.append("### Micro Summaries")
+        lines.append("")
+        for micro in list(ladder.micro_summaries)[-3:]:
+            lines.append(f"- {micro.compressed_text}")
+        lines.append("")
+
+    metrics = ladder.get_compression_metrics()["current_state"]
+    lines.extend([
+        "### Metrics",
+        "",
+        f"- Active Micro Summaries: {metrics['micro_summaries_active']}",
+        f"- Macro Distillations: {metrics['macro_distillations_total']}",
+        f"- Compression Ratio: {metrics['compression_ratio']:.2f}",
+        "",
+    ])
+
+    if report["new_macro_distillations"]:
+        lines.append("### New Distillation IDs")
+        lines.append("")
+        for item in report["new_macro_distillations"]:
+            lines.append(f"- {item['distillation_id']}")
+
+    return "\n".join(lines)
+
+
 # Create Gradio interface
 with gr.Blocks(title="Warbler CDA - FractalStat RAG") as demo:
     gr.Markdown("""
@@ -598,36 +672,66 @@ with gr.Blocks(title="Warbler CDA - FractalStat RAG") as demo:
         )
     
     with gr.Tab("System Stats"):
-        stats_output = gr.Markdown(get_system_stats())
+        stats_output = gr.Markdown("Click **Refresh Stats** to load system metrics.")
         stats_btn = gr.Button("Refresh Stats")
         def refresh_stats():
             return get_system_stats()
         stats_btn.click(fn=refresh_stats, outputs=stats_output)
-        demo.load(fn=refresh_stats, outputs=stats_output)
+
+    with gr.Tab("Synthesis"):
+        synthesis_input = gr.Textbox(
+            label="Document Text",
+            placeholder="Paste notes, docs, or source material to produce a hierarchical synthesis...",
+            lines=12,
+        )
+        with gr.Row():
+            synthesis_micro_window = gr.Slider(
+                minimum=2,
+                maximum=8,
+                value=5,
+                step=1,
+                label="Micro Window Size",
+            )
+            synthesis_macro_trigger = gr.Slider(
+                minimum=2,
+                maximum=5,
+                value=3,
+                step=1,
+                label="Macro Trigger Count",
+            )
+        synthesis_btn = gr.Button("Synthesize", variant="primary")
+        synthesis_output = gr.Markdown(
+            "Paste a document and click **Synthesize** to generate a TLDA-style macro distillation."
+        )
+        synthesis_btn.click(
+            fn=synthesize_document,
+            inputs=[synthesis_input, synthesis_micro_window, synthesis_macro_trigger],
+            outputs=synthesis_output,
+        )
     
     with gr.Tab("About"):
-        gr.Markdown("""
-        ## About Warbler CDA
-        
-        Warbler CDA is a production-ready RAG system featuring:
-        
-        - **8D FractalStat Addressing**: Multi-dimensional intelligence for superior retrieval
-        - **Semantic Anchors**: Persistent memory with provenance tracking
-        - **Bob the Skeptic**: Automatic bias detection and validation
-        - **Narrative Coherence**: Quality analysis beyond simple similarity
-        
-        ### Performance
-        
-        - 84% test coverage with 587 passing tests
-        - 9-28s query response time
-        - 0.88 average relevance score
-        - 75-83% narrative coherence
-        
-        ### Links
-        
-        - [Source Code](https://gitlab.com/tiny-walnut-games/the-seed)
-        - [Documentation](https://gitlab.com/tiny-walnut-games/the-seed/-/tree/main/warbler-cda-package)
-        - [Performance Report](https://gitlab.com/tiny-walnut-games/the-seed/-/blob/main/warbler-cda-package/WARBLER_CDA_PERFORMANCE_REPORT.md)
+        gr.HTML("""
+                <h2>About Warbler CDA</h2>
+                <p>Warbler CDA is a production-ready RAG system featuring:</p>
+                <ul>
+                    <li><strong>8D FractalStat Addressing</strong>: Multi-dimensional intelligence for superior retrieval</li>
+                    <li><strong>Semantic Anchors</strong>: Persistent memory with provenance tracking</li>
+                    <li><strong>Bob the Skeptic</strong>: Automatic bias detection and validation</li>
+                    <li><strong>Narrative Coherence</strong>: Quality analysis beyond simple similarity</li>
+                </ul>
+                <h3>Performance</h3>
+                <ul>
+                    <li>84% test coverage with 587 passing tests</li>
+                    <li>9-28s query response time</li>
+                    <li>0.88 average relevance score</li>
+                    <li>75-83% narrative coherence</li>
+                </ul>
+                <h3>Links</h3>
+                <ul>
+                    <li><a href="https://gitlab.com/tiny-walnut-games/the-seed" target="_blank">Source Code</a></li>
+                    <li><a href="https://gitlab.com/tiny-walnut-games/the-seed/-/tree/main/warbler-cda-package" target="_blank">Documentation</a></li>
+                    <li><a href="https://gitlab.com/tiny-walnut-games/the-seed/-/blob/main/warbler-cda-package/WARBLER_CDA_PERFORMANCE_REPORT.md" target="_blank">Performance Report</a></li>
+                </ul>
         """)
 
 if __name__ == "__main__":

warbler_cda/api/service.py

@@ -1,60 +1,3 @@
-# --- TLDA-style Synthesis Imports and Endpoint (at end of file) ---
-from warbler_cda.summarization_ladder import SummarizationLadder
-from typing import List, Dict, Any, Optional
-from pydantic import BaseModel
-
-# Synthesis pipeline instance (lazy init)
-_synthesis_ladder: Optional[SummarizationLadder] = None
-
-class SynthesisRequest(BaseModel):
-    """Request model for synthesis (TLDA-style)"""
-    fragments: List[Dict[str, Any]]
-    config: Optional[Dict[str, Any]] = None
-
-class SynthesisResult(BaseModel):
-    """Response model for synthesis results"""
-    micro_summaries: List[Dict[str, Any]]
-    macro_distillations: List[Dict[str, Any]]
-    metrics: Dict[str, Any]
-
-def _init_synthesis_ladder(config=None):
-    global _synthesis_ladder
-    if _synthesis_ladder is None:
-        _synthesis_ladder = SummarizationLadder(config=config)
-    return _synthesis_ladder
-
-@app.post("/synthesize", response_model=SynthesisResult)
-async def synthesize(request: SynthesisRequest):
-    """Synthesize a document or fragment list using the SummarizationLadder pipeline."""
-    ladder = _init_synthesis_ladder(request.config)
-    report = ladder.process_fragments(request.fragments)
-    # Collect micro/macro summaries for response
-    micro_summaries = [
-        {
-            "summary_id": ms.summary_id,
-            "compressed_text": ms.compressed_text,
-            "window_size": ms.window_size,
-            "heat_aggregate": ms.heat_aggregate,
-            "age_seconds": ms.get_age_seconds(),
-        }
-        for ms in list(ladder.micro_summaries)
-    ]
-    macro_distillations = [
-        {
-            "distillation_id": md.distillation_id,
-            "distilled_essence": md.distilled_essence,
-            "consolidation_ratio": md.consolidation_ratio,
-            "anchor_reinforcements": md.anchor_reinforcements,
-        }
-        for md in ladder.macro_distillations
-    ]
-    metrics = ladder.get_compression_metrics()
-    return SynthesisResult(
-        micro_summaries=micro_summaries,
-        macro_distillations=macro_distillations,
-        metrics=metrics,
-    )
-
 """
 EXP-09 CLI API Service - FractalStat Retrieval API with Concurrency Support.
 
@@ -77,6 +20,7 @@ from warbler_cda.retrieval_api import RetrievalAPI, RetrievalQuery, RetrievalMod
 from warbler_cda.fractalstat_rag_bridge import FractalStatRAGBridge
 from warbler_cda.pack_loader import PackLoader
 from warbler_cda.answer_generator import AnswerGenerator
+from warbler_cda.summarization_ladder import SummarizationLadder
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
@@ -101,11 +45,6 @@ app = FastAPI(
     lifespan=lifespan,
 )
 
-# --- TLDA-style Synthesis Imports and Endpoint (immediately after app definition) ---
-from warbler_cda.summarization_ladder import SummarizationLadder
-from typing import List, Dict, Any, Optional
-from pydantic import BaseModel
-
 # Synthesis pipeline instance (lazy init)
 _synthesis_ladder: Optional[SummarizationLadder] = None
 
@@ -126,38 +65,6 @@ def _init_synthesis_ladder(config=None):
         _synthesis_ladder = SummarizationLadder(config=config)
     return _synthesis_ladder
 
-@app.post("/synthesize", response_model=SynthesisResult)
-async def synthesize(request: SynthesisRequest):
-    """Synthesize a document or fragment list using the SummarizationLadder pipeline."""
-    ladder = _init_synthesis_ladder(request.config)
-    report = ladder.process_fragments(request.fragments)
-    # Collect micro/macro summaries for response
-    micro_summaries = [
-        {
-            "summary_id": ms.summary_id,
-            "compressed_text": ms.compressed_text,
-            "window_size": ms.window_size,
-            "heat_aggregate": ms.heat_aggregate,
-            "age_seconds": ms.get_age_seconds(),
-        }
-        for ms in list(ladder.micro_summaries)
-    ]
-    macro_distillations = [
-        {
-            "distillation_id": md.distillation_id,
-            "distilled_essence": md.distilled_essence,
-            "consolidation_ratio": md.consolidation_ratio,
-            "anchor_reinforcements": md.anchor_reinforcements,
-        }
-        for md in ladder.macro_distillations
-    ]
-    metrics = ladder.get_compression_metrics()
-    return SynthesisResult(
-        micro_summaries=micro_summaries,
-        macro_distillations=macro_distillations,
-        metrics=metrics,
-    )
-
 # Global state
 _api_instance: Optional[RetrievalAPI] = None
 _answer_generator: Optional[AnswerGenerator] = None
@@ -891,6 +798,39 @@ async def reset_metrics():
     return {"status": "metrics reset"}
 
 
+@app.post("/synthesize", response_model=SynthesisResult)
+async def synthesize(request: SynthesisRequest):
+    """Synthesize fragments through the hierarchical summarization ladder."""
+    ladder = _init_synthesis_ladder(request.config)
+    ladder.process_fragments(request.fragments)
+
+    micro_summaries = [
+        {
+            "summary_id": summary.summary_id,
+            "compressed_text": summary.compressed_text,
+            "window_size": summary.window_size,
+            "heat_aggregate": summary.heat_aggregate,
+            "age_seconds": summary.get_age_seconds(),
+        }
+        for summary in list(ladder.micro_summaries)
+    ]
+    macro_distillations = [
+        {
+            "distillation_id": distillation.distillation_id,
+            "distilled_essence": distillation.distilled_essence,
+            "consolidation_ratio": distillation.consolidation_ratio,
+            "anchor_reinforcements": distillation.anchor_reinforcements,
+        }
+        for distillation in ladder.macro_distillations
+    ]
+
+    return SynthesisResult(
+        micro_summaries=micro_summaries,
+        macro_distillations=macro_distillations,
+        metrics=ladder.get_compression_metrics(),
+    )
+
+
 if __name__ == "__main__":
     import uvicorn