symfonic.tools.hydrate_context

hydrate_context

hydrate_context: LangChain tool for on-demand memory loading in JIT mode.

In JIT Context mode (FrameworkConfig.jit_context=True) the system prompt is minimal and does not pre-load memory context. The LLM calls this tool to fetch relevant semantic nodes on demand before composing its response.

The hydration is RELATIONAL: it not only returns nodes matching the domain but also expands via SpreadingActivation to include the heaviest-weighted neighbors in the graph. This gives the agent full relational context rather than isolated facts.

create_hydrate_context_tool

create_hydrate_context_tool(
    orchestrator: Any,
    scope: Any,
    *,
    max_neighbors: int = 5,
    min_importance: float = 0.0,
) -> Any

Return a LangChain tool bound to a specific orchestrator and scope.

The tool queries the semantic memory layer for nodes whose label prefix matches the requested domain, then expands the result via spreading activation to include the heaviest-weighted graph neighbors.

Parameters:

Name	Type	Description	Default
orchestrator	Any	MemoryOrchestrator instance with a semantic layer.	required
scope	Any	FrameworkTenantScope providing tenant isolation.	required
max_neighbors	int	Cap on neighbors fetched per seed node (default 5).	5
min_importance	float	Skip neighbors below this importance (default 0.0).	0.0

Returns:

Type	Description
Any	A LangChain @tool-decorated async function ready for registration.

Source code in src/symfonic/tools/hydrate_context.py

def create_hydrate_context_tool(
    orchestrator: Any,
    scope: Any,
    *,
    max_neighbors: int = 5,
    min_importance: float = 0.0,
) -> Any:
    """Return a LangChain tool bound to a specific orchestrator and scope.

    The tool queries the semantic memory layer for nodes whose label prefix
    matches the requested domain, then expands the result via spreading
    activation to include the heaviest-weighted graph neighbors.

    Args:
        orchestrator: MemoryOrchestrator instance with a semantic layer.
        scope: FrameworkTenantScope providing tenant isolation.
        max_neighbors: Cap on neighbors fetched per seed node (default 5).
        min_importance: Skip neighbors below this importance (default 0.0).

    Returns:
        A LangChain @tool-decorated async function ready for registration.
    """
    from langchain_core.tools import tool

    from symfonic.memory.retrieval.spreading import SpreadingActivation
    from symfonic.memory.types import MemoryLayer

    @tool
    async def hydrate_context(domain: str, top_k: int = 5) -> str:
        """Load relational memory context on demand for a given domain or topic.

        Use this tool in JIT mode before answering a query that requires
        prior memory context.  Queries the semantic memory layer for nodes
        matching the domain, then expands via spreading activation to pull
        in the heaviest-weighted graph neighbors.

        Args:
            domain: Topic or label prefix to search for (e.g. "user", "project").
            top_k: Maximum number of seed memory nodes to retrieve (default 5).

        Returns:
            Formatted context string with matching seed nodes plus their
            1st-degree graph neighbors, or a message indicating no relevant
            context was found.
        """
        # Clamp caller-supplied top_k so an LLM hallucinating top_k=10_000
        # can't trigger a huge semantic retrieval + O(n) graph rehydration.
        try:
            top_k_int = int(top_k) if top_k is not None else 5
        except (TypeError, ValueError):
            top_k_int = 5
        effective_top_k = max(1, min(top_k_int, MAX_TOP_K))
        if effective_top_k != top_k_int:
            logger.info(
                "hydrate_context: clamped top_k=%s to %s (MAX_TOP_K=%s)",
                top_k_int, effective_top_k, MAX_TOP_K,
            )

        try:
            semantic_layer = orchestrator.get_layer(MemoryLayer.SEMANTIC)
            if semantic_layer is None:
                return (
                    f"[hydrate_context] No semantic layer available for "
                    f"domain='{domain}'"
                )

            resolved_scope = (
                scope()
                if callable(scope) and not hasattr(scope, "to_memory_scope")
                else scope
            )
            memory_scope = resolved_scope.to_memory_scope()
            entries = await semantic_layer.retrieve(
                memory_scope, domain, top_k=effective_top_k,
            )

            if not entries:
                return (
                    f"[hydrate_context] No memory nodes found for "
                    f"domain='{domain}'"
                )

            lines: list[str] = [
                f"[hydrate_context] Domain: {domain} ({len(entries)} seeds)"
            ]

            # Re-hydrate MemoryEntry.node_id into full MemoryNode so
            # spreading activation has access to .id/.label.  Same pattern
            # as engine._run_spreading_activation.
            graph_store = getattr(semantic_layer, "_graph", None)
            seed_nodes: list[Any] = []
            for entry in entries:
                if hasattr(entry, "id") and hasattr(entry, "label"):
                    seed_nodes.append(entry)
                elif (
                    graph_store is not None
                    and hasattr(entry, "node_id")
                    and entry.node_id is not None
                ):
                    try:
                        node = await graph_store.get_node(
                            memory_scope, entry.node_id,
                        )
                        if node is not None:
                            seed_nodes.append(node)
                    except Exception:
                        logger.debug(
                            "hydrate_context: re-hydration failed for %s",
                            entry.node_id, exc_info=True,
                        )

                content = getattr(entry, "content", str(entry))
                layer_name = getattr(entry, "layer", "semantic")
                metadata = getattr(entry, "metadata", {}) or {}
                line = f"  [{layer_name}] {content}"
                if metadata:
                    extra: list[str] = []
                    for key in ("name", "description", "role"):
                        val = metadata.get(key)
                        if val and isinstance(val, str):
                            extra.append(f"{key}={val}")
                    if extra:
                        line += f" ({', '.join(extra[:3])})"
                lines.append(line)

            # Spreading activation: expand with heaviest-weighted neighbors.
            # The SpreadingActivation engine already sorts neighbors by
            # importance descending, giving us the "heaviest" relations.
            if graph_store is not None and seed_nodes:
                try:
                    spreader = SpreadingActivation(
                        graph_store,
                        max_neighbors_per_node=max_neighbors,
                        min_importance=min_importance,
                    )
                    seen_ids = {str(getattr(n, "id", "")) for n in seed_nodes}
                    neighbors = await spreader.expand(
                        memory_scope, seed_nodes, already_seen_ids=seen_ids,
                    )
                    if neighbors:
                        lines.append(
                            f"  --- {len(neighbors)} related neighbors "
                            "(by weight) ---"
                        )
                        for n in neighbors[: max_neighbors * 2]:
                            source_label = getattr(n.source_node, "label", "?")
                            target_label = getattr(n.node, "label", "?")
                            importance = getattr(n.node, "importance", 5.0)
                            target_content = str(
                                (n.node.properties or {}).get(
                                    "content", target_label
                                )
                            )
                            lines.append(
                                f"  [linked:{importance:.1f}] {source_label}"
                                f" -> {n.relationship}: {target_content}"
                            )
                except Exception:
                    logger.debug(
                        "hydrate_context spreading activation failed",
                        exc_info=True,
                    )

            return _truncate_output("\n".join(lines))

        except Exception as exc:
            logger.warning(
                "hydrate_context failed for domain=%s: %s",
                domain, exc, exc_info=True,
            )
            return (
                f"[hydrate_context] Error loading context for "
                f"domain='{domain}': {exc}"
            )

    return hydrate_context