feat: cluster tips by task description similarity

kaizen/cli/cli.py

@@ -330,6 +330,73 @@ def show_entity(
         raise typer.Exit(1)
 
 
+@entities_app.command("consolidate")
+def consolidate_entities(
+    namespace: Annotated[str, typer.Argument(help="Namespace to consolidate entities in")],
+    threshold: Annotated[Optional[float], typer.Option("--threshold", "-t", help="Cosine similarity threshold (0-1)")] = None,
+    dry_run: Annotated[bool, typer.Option("--dry-run", help="Show clusters without modifying anything")] = True,
+):
+    """Cluster similar guideline entities by task description similarity."""
+    from kaizen.config.kaizen import kaizen_config
+
+    client = get_client()
+
+    effective_threshold = threshold if threshold is not None else kaizen_config.clustering_threshold
+
+    console.print(f"[bold]Clustering entities in '{namespace}'[/bold]")
+    console.print(f"  Threshold: {effective_threshold}")
+    console.print(f"  Dry run: {dry_run}")
+    console.print()
+
+    try:
+        clusters = client.cluster_tips(namespace, threshold=effective_threshold)
+    except NamespaceNotFoundException:
+        console.print(f"[red]Namespace '{namespace}' not found.[/red]")
+        raise typer.Exit(1)
+
+    if not clusters:
+        console.print("[yellow]No clusters found. Tips have dissimilar task descriptions.[/yellow]")
+        return
+
+    console.print(f"[green]Found {len(clusters)} cluster(s)[/green]\n")
+
+    for i, cluster in enumerate(clusters, 1):
+        table = Table(title=f"Cluster {i} ({len(cluster)} entities)")
+        table.add_column("ID", style="cyan", max_width=20)
+        table.add_column("Task Description", max_width=40)
+        table.add_column("Content", max_width=50)
+
+        for entity in cluster:
+            task_desc = (entity.metadata or {}).get("task_description", "")
+            if len(task_desc) > 40:
+                task_desc = task_desc[:37] + "..."
+            content_str = str(entity.content)
+            if len(content_str) > 50:
+                content_str = content_str[:47] + "..."
+            table.add_row(str(entity.id), task_desc, content_str)
+
+        console.print(table)
+        console.print()
+
+    total_entities = sum(len(c) for c in clusters)
+    console.print(f"[dim]Total: {total_entities} entities in {len(clusters)} clusters[/dim]")
+
+    if dry_run:
+        console.print("\n[yellow]Dry run — no changes made. Use --no-dry-run to consolidate.[/yellow]")
+        return
+
+    console.print("\n[bold]Consolidating clusters...[/bold]")
+    try:
+        result = client.consolidate_tips(namespace, threshold=effective_threshold)
+        console.print("[green]Consolidation complete:[/green]")
+        console.print(f"  Clusters combined: {result.clusters_found}")
+        console.print(f"  Tips before: {result.tips_before}")
+        console.print(f"  Tips after: {result.tips_after}")
+    except KaizenException as e:
+        console.print(f"[red]Consolidation failed: {e}[/red]")
+        raise typer.Exit(1)
+
+
 # =============================================================================
 # Sync Commands
 # =============================================================================

kaizen/config/kaizen.py

@@ -7,6 +7,7 @@ class KaizenConfig(BaseSettings):
     backend: Literal["milvus", "filesystem"] = "milvus"
     namespace_id: str = "kaizen"
     settings: BaseSettings | None = None
+    clustering_threshold: float = 0.80
 
 
 # to reload settings call kaizen_config.__init__()

kaizen/frontend/client/kaizen_client.py

@@ -1,9 +1,14 @@
+import logging
+
 from kaizen.schema.core import Entity, Namespace, RecordedEntity
 from kaizen.schema.exceptions import NamespaceNotFoundException
 from kaizen.schema.conflict_resolution import EntityUpdate
+from kaizen.schema.tips import ConsolidationResult
 from kaizen.config.kaizen import KaizenConfig
 from kaizen.backend.base import BaseEntityBackend
 
+logger = logging.getLogger(__name__)
+
 
 class KaizenClient:
     """Wrapper client around kaizen entity backends."""
@@ -70,6 +75,104 @@ def delete_entity_by_id(self, namespace_id: str, entity_id: str) -> None:
         """Delete a specific entity by its ID."""
         self.backend.delete_entity_by_id(namespace_id, entity_id)
 
+    def cluster_tips(self, namespace_id: str, threshold: float | None = None, limit: int = 10000) -> list[list[RecordedEntity]]:
+        """Cluster guideline entities by task description similarity.
+
+        Args:
+            namespace_id: Namespace to fetch entities from.
+            threshold: Cosine similarity threshold (0-1). Defaults to config value.
+            limit: Maximum number of guideline entities to fetch for clustering.
+
+        Returns:
+            List of clusters, each containing related RecordedEntity objects.
+        """
+        from kaizen.llm.tips.clustering import cluster_entities
+
+        if threshold is None:
+            threshold = self.config.clustering_threshold
+
+        entities = self.get_all_entities(namespace_id, filters={"type": "guideline"}, limit=limit)
+        if len(entities) >= limit:
+            logger.warning(
+                "Fetched %d entities (hit limit=%d); clustering results may be incomplete. Consider increasing the limit.",
+                len(entities),
+                limit,
+            )
+        return cluster_entities(entities, threshold=threshold)
+
+    def consolidate_tips(self, namespace_id: str, threshold: float | None = None) -> ConsolidationResult:
+        """Cluster similar tips and combine each cluster into consolidated guidelines.
+
+        Args:
+            namespace_id: Namespace to consolidate entities in.
+            threshold: Cosine similarity threshold (0-1). Defaults to config value.
+
+        Returns:
+            ConsolidationResult with counts of clusters, tips before, and tips after.
+        """
+        from kaizen.llm.tips.clustering import combine_cluster
+
+        clusters = self.cluster_tips(namespace_id, threshold=threshold)
+        clusters_found = 0
+        tips_before = 0
+        tips_after = 0
+
+        for cluster in clusters:
+            # Phase 1: combine + insert (skip cluster on failure)
+            try:
+                consolidated_tips = combine_cluster(cluster)
+
+                task_description = (cluster[0].metadata or {}).get("task_description", "")
+                new_entities = [
+                    Entity(
+                        content=tip.content,
+                        type="guideline",
+                        metadata={
+                            "task_description": task_description,
+                            "rationale": tip.rationale,
+                            "category": tip.category,
+                            "trigger": tip.trigger,
+                        },
+                    )
+                    for tip in consolidated_tips
+                ]
+                if not new_entities:
+                    logger.warning(
+                        "LLM returned no consolidated tips for cluster (IDs: %s); skipping deletion.",
+                        [e.id for e in cluster],
+                    )
+                    continue
+                self.update_entities(namespace_id, new_entities, enable_conflict_resolution=False)
+            except Exception:
+                logger.warning(
+                    "Failed to consolidate cluster of %d entities (IDs: %s); skipping.",
+                    len(cluster),
+                    [e.id for e in cluster],
+                    exc_info=True,
+                )
+                continue
+
+            clusters_found += 1
+            tips_before += len(cluster)
+            tips_after += len(consolidated_tips)
+
+            # Phase 2: delete originals (log errors but don't roll back insert)
+            for entity in cluster:
+                try:
+                    self.delete_entity_by_id(namespace_id, entity.id)
+                except Exception:
+                    logger.warning(
+                        "Failed to delete original entity %s after successful insert; skipping.",
+                        entity.id,
+                        exc_info=True,
+                    )
+
+        return ConsolidationResult(
+            clusters_found=clusters_found,
+            tips_before=tips_before,
+            tips_after=tips_after,
+        )
+
     # Convenience methods for common patterns
     def namespace_exists(self, namespace_id: str) -> bool:
         """Check if a namespace exists."""