hierarchical_demo.py•3.42 kB
"""
Example Script: Hierarchical Compression
Demonstrates N0 → N1 → N2 creation
"""
import asyncio
from continuo.embedding_service import EmbeddingService
from continuo.hierarchical_compression import HierarchicalCompressor
from continuo.memory_server import MemoryServer
def mock_llm_summarize(text: str) -> str:
"""Simulates LLM summary"""
# In practice, use OpenAI/Claude
words = text.split()
summary = " ".join(words[:30]) + "... [LLM summary]"
return summary
async def main():
"""Demonstrates hierarchical compression"""
print("=" * 60)
print("📦 Continuo - Hierarchical Compression - Demo")
print("=" * 60)
# Setup
embedding_service = EmbeddingService(provider="local")
compressor = HierarchicalCompressor(
max_working_set_size=3,
compression_ratio_n1=3,
compression_ratio_n2=2,
)
memory = MemoryServer(
embedding_service=embedding_service,
db_path="./compression_demo_db",
hierarchical_compressor=compressor,
)
# Create N0 chunks
print("\n[1/4] Creating N0 chunks (raw)...")
chunks = []
for i in range(6):
text = f"Chunk {i}: Implementation of feature {i} with technical details about architecture, patterns, and design decisions..."
doc_id = memory.store(text, {"index": i}, level="N0")
chunks.append({"id": doc_id, "text": text, "metadata": {"index": i}})
print(f" ✓ N0 chunk {i}")
# Create N1 micro-summaries
print("\n[2/4] Creating N1 micro-summaries...")
micro_summaries = []
for batch_idx in range(0, len(chunks), 3):
batch = chunks[batch_idx : batch_idx + 3]
summary = compressor.create_micro_summary(batch, llm_summarize_fn=mock_llm_summarize)
doc_id = memory.store(summary["text"], summary["metadata"], level="N1")
summary["id"] = doc_id
micro_summaries.append(summary)
print(f" ✓ N1 summary {len(micro_summaries)} (from {len(batch)} chunks)")
# Create N2 meta-summary
print("\n[3/4] Creating N2 meta-summary...")
meta_summary = compressor.create_meta_summary(
micro_summaries, llm_summarize_fn=mock_llm_summarize
)
doc_id = memory.store(meta_summary["text"], meta_summary["metadata"], level="N2")
print(" ✓ N2 meta-summary created")
# Demonstrate hierarchical search
print("\n[4/4] Testing hierarchical search...")
query = "implemented feature"
print(f"\n🔍 Query: '{query}'")
for level in ["N2", "N1", "N0"]:
results = memory.retrieve(query, top_k=2, level_filter=level)
print(f"\n Level {level}:")
for i, doc in enumerate(results["documents"], 1):
print(f" [{i}] {doc[:60]}...")
# Statistics
print("\n" + "=" * 60)
print("📊 Final Statistics")
print("=" * 60)
stats = memory.get_stats()
print(f" Total: {stats['total_documents']} docs")
print(f" N0 (chunks): {stats['by_level']['N0_chunks']}")
print(f" N1 (micro): {stats['by_level']['N1_micro_summaries']}")
print(f" N2 (meta): {stats['by_level']['N2_meta_summaries']}")
# Calculate compression rate
compression_rate = stats["by_level"]["N0_chunks"] / max(
1, stats["by_level"]["N2_meta_summaries"]
)
print(f" Compression rate: {compression_rate:.1f}x")
print("\n✅ Demo completed!")
if __name__ == "__main__":
asyncio.run(main())