feat(memory): dosage-aware guideline retrieval (core + top-k)

altk_evolve/cli/cli.py

@@ -407,6 +407,41 @@ def consolidate_entities(
         raise typer.Exit(1)
 
 
+@entities_app.command("select")
+def select_entities(
+    namespace: Annotated[str, typer.Argument(help="Namespace to select guidelines from")],
+    task: Annotated[str, typer.Option("--task", "-q", help="Task instruction to retrieve guidelines for")],
+    top_k: Annotated[Optional[int], typer.Option("--top-k", "-k", help="Max task-specific guidelines beyond the core")] = None,
+    core_support: Annotated[Optional[int], typer.Option("--core-support", "-c", help="Support threshold for the always-on core")] = None,
+):
+    """Select an always-on core plus the top-k task-relevant guidelines (dosage-aware retrieval)."""
+    client = get_client()
+
+    try:
+        selection = client.select_guidelines(namespace, task, top_k=top_k, core_support=core_support)
+    except NamespaceNotFoundException:
+        console.print(f"[red]Namespace '{namespace}' not found.[/red]")
+        raise typer.Exit(1)
+    except ValueError as e:
+        console.print(f"[red]Retrieval unavailable:[/red] {e}")
+        console.print("[yellow]Configure the embedding model/backend before selecting guidelines.[/yellow]")
+        raise typer.Exit(1)
+
+    if not selection.all:
+        console.print("[yellow]No guidelines found for this task.[/yellow]")
+        return
+
+    console.print(f"[bold]Core ({len(selection.core)}):[/bold]")
+    for entity in selection.core:
+        console.print(f"  • {entity.content}")
+    console.print(f"\n[bold]Retrieved ({len(selection.retrieved)}):[/bold]")
+    for entity in selection.retrieved:
+        console.print(f"  • {entity.content}")
+    console.print(
+        f"\n[dim]Total: {len(selection.all)} guidelines ({len(selection.core)} core + {len(selection.retrieved)} retrieved)[/dim]"
+    )
+
+
 # =============================================================================
 # Sync Commands
 # =============================================================================

altk_evolve/config/evolve.py

@@ -1,3 +1,4 @@
+from pydantic import Field, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from typing import Literal
 
@@ -17,6 +18,30 @@ class EvolveConfig(BaseSettings):
     # time, not by deleting entities here.
     consolidation_mode: Literal["none", "lossless", "lossy"] = "lossless"
     lossy_target_num_guidelines: int = 12
+    # Dosage-aware retrieval knobs (see docs: capability-dependent dosage).
+    #   static    - inject the whole playbook (best for strong models; current default behavior)
+    #   retrieval - inject core (support >= core_support) + top-k task-relevant guidelines
+    # Default is "static" so existing get_guidelines callers are unaffected; set "retrieval"
+    # to opt get_guidelines into the dosage-aware path.
+    injection_mode: Literal["static", "retrieval"] = "static"
+    retrieval_top_k: int = Field(default=10, ge=0)
+    core_support: int = Field(default=3, ge=1)
+    # Non-destructive sup2/sup3 floor on the candidate pool. Constrained to <= core_support
+    # (see validator below) so it can never drop a guideline that qualifies for the core.
+    min_support: int = Field(default=1, ge=1)
+    retrieval_similarity_key: Literal["source_task", "guideline_text"] = "source_task"
+    retrieval_near_core_thresh: float = Field(default=0.75, ge=0.0, le=1.0)
+    retrieval_dedup_thresh: float = Field(default=0.90, ge=0.0, le=1.0)
+    evidence_filter: Literal["all", "success", "failure"] = "all"
+
+    @model_validator(mode="after")
+    def _check_support_thresholds(self) -> "EvolveConfig":
+        if self.min_support > self.core_support:
+            raise ValueError(
+                f"min_support ({self.min_support}) must be <= core_support ({self.core_support}); "
+                "a floor above the core threshold would drop guidelines that qualify for the core."
+            )
+        return self
 
 
 # to reload settings call evolve_config.__init__()

altk_evolve/frontend/client/evolve_client.py

@@ -1,4 +1,5 @@
 import logging
+from typing import TYPE_CHECKING, cast
 
 from altk_evolve.backend.base import BaseEntityBackend
 from altk_evolve.config.evolve import EvolveConfig
@@ -7,9 +8,23 @@
 from altk_evolve.schema.exceptions import NamespaceAlreadyExistsException, NamespaceNotFoundException
 from altk_evolve.schema.guidelines import ConsolidationResult
 
+if TYPE_CHECKING:
+    from altk_evolve.llm.guidelines.retrieval import GuidelineSelection, SimilarityKey
+
 logger = logging.getLogger(__name__)
 
 
+def _filter_by_evidence(entities: list[RecordedEntity], evidence_filter: str) -> list[RecordedEntity]:
+    """Keep guidelines matching an evidence polarity. Unknown evidence (None) is always kept."""
+    if evidence_filter == "success":
+        allowed = {"success", "both", None}
+    elif evidence_filter == "failure":
+        allowed = {"failure", "both", None}
+    else:  # "all"
+        return entities
+    return [e for e in entities if (e.metadata or {}).get("evidence") in allowed]
+
+
 class EvolveClient:
     """Wrapper client around evolve entity backends."""
 
@@ -246,6 +261,61 @@ def consolidate_guidelines(self, namespace_id: str, threshold: float | None = No
             support_after=support_after,
         )
 
+    def select_guidelines(
+        self,
+        namespace_id: str,
+        task_query: str,
+        top_k: int | None = None,
+        core_support: int | None = None,
+        min_support: int | None = None,
+        evidence_filter: str | None = None,
+        limit: int = 10000,
+    ) -> "GuidelineSelection":
+        """Select the always-on core plus the top-k task-relevant guidelines for a task.
+
+        This is the dosage-aware alternative to injecting the whole playbook: the core
+        (guidelines with support >= ``core_support``) is always returned, plus up to
+        ``top_k`` further guidelines whose source task is most similar to ``task_query``.
+        Unset arguments fall back to the corresponding ``config`` values.
+
+        Args:
+            namespace_id: Namespace to select guidelines from.
+            task_query: The current task instruction to retrieve for.
+            top_k: Max retrieved (non-core) guidelines. Defaults to ``config.retrieval_top_k``.
+            core_support: Support threshold for the always-on core. Defaults to config.
+            min_support: Non-destructive support floor on the candidate pool. Defaults to config.
+            evidence_filter: ``"all"`` / ``"success"`` / ``"failure"``. Defaults to config.
+            limit: Max guideline entities to fetch.
+
+        Returns:
+            A ``GuidelineSelection`` (core + retrieved).
+        """
+        from altk_evolve.llm.guidelines.retrieval import GuidelineSelection, select_guidelines
+
+        top_k = top_k if top_k is not None else getattr(self.config, "retrieval_top_k", 10)
+        core_support = core_support if core_support is not None else getattr(self.config, "core_support", 3)
+        min_support = min_support if min_support is not None else getattr(self.config, "min_support", 1)
+        evidence_filter = evidence_filter or getattr(self.config, "evidence_filter", "all")
+        similarity_key = getattr(self.config, "retrieval_similarity_key", "source_task")
+        near_core_thresh = getattr(self.config, "retrieval_near_core_thresh", 0.75)
+        dedup_thresh = getattr(self.config, "retrieval_dedup_thresh", 0.90)
+
+        entities = self.get_all_entities(namespace_id, filters={"type": "guideline"}, limit=limit)
+        entities = _filter_by_evidence(entities, str(evidence_filter))
+        if not entities:
+            return GuidelineSelection(core=[], retrieved=[])
+
+        return select_guidelines(
+            entities,
+            task_query,
+            top_k=int(top_k),
+            core_support=int(core_support),
+            min_support=int(min_support),
+            similarity_key=cast("SimilarityKey", similarity_key),
+            near_core_thresh=float(near_core_thresh),
+            dedup_thresh=float(dedup_thresh),
+        )
+
     # Convenience methods for common patterns
     def namespace_exists(self, namespace_id: str) -> bool:
         """Check if a namespace exists."""

altk_evolve/frontend/mcp/mcp_server.py

@@ -305,15 +305,74 @@ def get_guidelines(
     Provide a task description and receive applicable best practices and guidelines.
     This tool is maintained for backward compatibility. Use 'get_entities' for more generic queries.
 
+    Honors ``EvolveConfig.injection_mode``: 'static' (default) returns the whole guideline set,
+    while 'retrieval' routes through the dosage-aware core + top-k path (see get_relevant_guidelines).
+
     Args:
         task: A description of the task you want guidelines for
         user_id: Optional caller user ID. Logged for attribution; does not filter results.
         namespace_id: Optional namespace override. Falls back to the configured default.
         session_id: Optional session/thread ID. Logged for attribution; does not filter results.
     """
+    from altk_evolve.config.evolve import evolve_config
+
+    if evolve_config.injection_mode == "retrieval":
+        return get_relevant_guidelines(task, user_id=user_id, namespace_id=namespace_id, session_id=session_id)
     return get_entities_logic(task, "guideline", user_id=user_id, namespace_id=namespace_id, session_id=session_id)
 
 
+@mcp.tool()
+def get_relevant_guidelines(
+    task: str,
+    top_k: int | None = None,
+    core_support: int | None = None,
+    namespace_id: str | None = None,
+    user_id: str | None = None,
+    session_id: str | None = None,
+) -> str:
+    """
+    Get a dosage-aware set of guidelines for a task: an always-on core plus the top-k
+    guidelines most relevant to this task (retrieved by similarity to the tasks they were
+    learned from).
+
+    Prefer this over 'get_guidelines'/'get_entities' for weaker models, where injecting the
+    whole playbook can hurt and a small, targeted set helps.
+
+    Args:
+        task: A description of the task you want guidelines for.
+        top_k: Max task-specific guidelines to add beyond the core. Defaults to config.
+        core_support: Support threshold for the always-on core. Defaults to config.
+        namespace_id: Optional namespace override. Falls back to the configured default.
+        user_id: Optional caller user ID. Logged for attribution; does not filter results.
+        session_id: Optional session/thread ID. Logged for attribution; does not filter results.
+    """
+    from altk_evolve.llm.guidelines.retrieval import format_selection
+
+    resolved_ns = _resolve_namespace(namespace_id)
+    # Log only non-sensitive metadata at INFO; task is arbitrary user text (see save_trajectory).
+    logger.info(
+        "get_relevant_guidelines (namespace=%s, top_k=%s, core_support=%s, task_len=%s, user_present=%s, session_present=%s)",
+        resolved_ns,
+        top_k,
+        core_support,
+        len(task),
+        user_id is not None,
+        session_id is not None,
+    )
+    logger.debug("get_relevant_guidelines task=%s", task)
+    client = get_client()
+    try:
+        selection = client.select_guidelines(resolved_ns, task, top_k=top_k, core_support=core_support)
+    except NamespaceNotFoundException:
+        _evict_namespace(resolved_ns)
+        resolved_ns = _resolve_namespace(namespace_id)
+        selection = client.select_guidelines(resolved_ns, task, top_k=top_k, core_support=core_support)
+    except ValueError as e:
+        logger.error("Retrieval unavailable for get_relevant_guidelines: %s", e)
+        return f"Guideline retrieval unavailable: {e}"
+    return format_selection(selection)
+
+
 def _empty_store_user_facts_response(user_id: str) -> str:
     return json.dumps({"user_id": user_id, "stored_count": 0, "updates": []})
 

altk_evolve/llm/guidelines/retrieval.py

@@ -0,0 +1,152 @@
+"""Dosage-aware guideline selection: always-on core + per-task retrieved guidelines.
+
+Instead of injecting the whole playbook every task (best only for strong models), this
+selects a small, task-relevant dose (best for weaker models — the capability-dependent
+dosage finding):
+
+    core      = guidelines whose ``support`` >= ``core_support`` (they recurred across many
+                tasks, so they generalise) — always included.
+    retrieved = up to ``top_k`` further guidelines whose SOURCE task (``task_description``)
+                is most similar to the current task ("a lesson from a task like this one"),
+                after dropping ones already covered by the core or duplicating each other.
+
+``min_support`` applies a non-destructive support threshold (the sup2/sup3 filter) to the
+candidate pool without deleting anything from the store.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import Literal
+
+import numpy as np
+
+from altk_evolve.llm.guidelines.clustering import _get_sentence_transformer
+from altk_evolve.schema.core import RecordedEntity
+
+logger = logging.getLogger(__name__)
+
+SimilarityKey = Literal["source_task", "guideline_text"]
+
+
+@dataclass(frozen=True)
+class GuidelineSelection:
+    """Result of :func:`select_guidelines` — core (always-on) plus retrieved guidelines."""
+
+    core: list[RecordedEntity] = field(default_factory=list)
+    retrieved: list[RecordedEntity] = field(default_factory=list)
+
+    @property
+    def all(self) -> list[RecordedEntity]:
+        return [*self.core, *self.retrieved]
+
+
+def _support(entity: RecordedEntity) -> int:
+    try:
+        return max(1, int((entity.metadata or {}).get("support", 1) or 1))
+    except (TypeError, ValueError):
+        return 1
+
+
+def _key_text(entity: RecordedEntity, similarity_key: SimilarityKey) -> str:
+    if similarity_key == "source_task":
+        return str((entity.metadata or {}).get("task_description", "") or entity.content)
+    return str(entity.content)
+
+
+def _embed(texts: list[str], embedding_model: str) -> np.ndarray:
+    model = _get_sentence_transformer(embedding_model)
+    return np.asarray(model.encode(texts, normalize_embeddings=True))
+
+
+def select_guidelines(
+    entities: list[RecordedEntity],
+    task_query: str,
+    *,
+    top_k: int = 10,
+    core_support: int = 3,
+    min_support: int = 1,
+    similarity_key: SimilarityKey = "source_task",
+    near_core_thresh: float = 0.75,
+    dedup_thresh: float = 0.90,
+    embedding_model: str | None = None,
+) -> GuidelineSelection:
+    """Select the always-on core plus the top-``top_k`` task-relevant guidelines.
+
+    Args:
+        entities: Candidate guideline entities (carrying ``support`` and, for
+            ``similarity_key="source_task"``, ``task_description`` in metadata).
+        task_query: The current task instruction to retrieve for.
+        top_k: Maximum number of retrieved (non-core) guidelines.
+        core_support: Guidelines with support >= this are always included.
+        min_support: Non-destructive sup2/sup3 floor applied to the whole pool *before* the
+            core/candidate split. Callers should keep ``min_support <= core_support`` (enforced
+            by ``EvolveConfig``) so the floor never drops a guideline that qualifies for the core.
+        similarity_key: Retrieve by the source ``task_description`` (default) or by the
+            guideline text itself. For ``"source_task"``, a candidate lacking a
+            ``task_description`` falls back to being ranked by its own content.
+        near_core_thresh: Drop a candidate whose content cosine to any core guideline is
+            >= this (already covered by the core).
+        dedup_thresh: Drop a candidate whose content cosine to an already-kept candidate is
+            >= this.
+        embedding_model: SentenceTransformer model name. Defaults to the configured model.
+
+    Returns:
+        A :class:`GuidelineSelection` (core first, then retrieved by descending relevance).
+    """
+    if embedding_model is None:
+        from altk_evolve.config.milvus import milvus_other_settings
+
+        embedding_model = milvus_other_settings.embedding_model
+
+    pool = [e for e in entities if _support(e) >= min_support]
+    core = [e for e in pool if _support(e) >= core_support]
+    candidates = [e for e in pool if _support(e) < core_support]
+
+    if top_k <= 0 or not candidates:
+        return GuidelineSelection(core=core, retrieved=[])
+
+    core_content_emb = _embed([str(e.content) for e in core], embedding_model) if core else None
+    cand_content_emb = _embed([str(e.content) for e in candidates], embedding_model)
+    cand_key_emb = (
+        cand_content_emb
+        if similarity_key == "guideline_text"
+        else _embed([_key_text(e, similarity_key) for e in candidates], embedding_model)
+    )
+    query_emb = _embed([task_query], embedding_model)[0]
+
+    # Drop candidates already covered by the core, then dedup the remainder, then rank by
+    # similarity of the source task to the current task.
+    kept: list[tuple[int, float]] = []  # (candidate index, relevance score)
+    kept_content: list[np.ndarray] = []
+    order = sorted(range(len(candidates)), key=lambda i: float(cand_key_emb[i] @ query_emb), reverse=True)
+    for i in order:
+        content_vec = cand_content_emb[i]
+        if core_content_emb is not None and float(np.max(core_content_emb @ content_vec)) >= near_core_thresh:
+            continue
+        if kept_content and float(np.max(np.stack(kept_content) @ content_vec)) >= dedup_thresh:
+            continue
+        kept.append((i, float(cand_key_emb[i] @ query_emb)))
+        kept_content.append(content_vec)
+        if len(kept) >= top_k:
+            break
+
+    retrieved = [candidates[i] for i, _ in kept]
+    return GuidelineSelection(core=core, retrieved=retrieved)
+
+
+def format_selection(selection: GuidelineSelection) -> str:
+    """Render a selection as an injectable guidelines block (mirrors the retrieved-tips agent)."""
+    lines = [
+        "Guidelines learned from past task attempts (follow these carefully; they address the most common mistakes):",
+        "",
+    ]
+    for entity in selection.core:
+        lines.append(f"- {entity.content}")
+    if selection.retrieved:
+        lines.append("")
+        lines.append("Additional task-specific guidelines retrieved from similar past tasks:")
+        for entity in selection.retrieved:
+            lines.append(f"- {entity.content}")
+    return "\n".join(lines)