symfonic.memory.layers.procedural.router

router

CapabilityRouter -- LLM-driven tool selection from procedural memory.

Queries the PROCEDURAL layer for relevant skills, searches the ToolCatalog, and uses the LLM to select the minimal set of tools needed for a request.

CapabilityRouter

CapabilityRouter(
    procedural_layer: ProceduralLayer,
    catalog: ToolCatalog,
    llm: Any,
    max_tools: int = 5,
    *,
    observability_hook: ObservabilityHook | None = None,
)

Routes user queries to the appropriate tools via LLM selection.

Combines procedural memory (learned skills) with a static tool catalog to select the minimal set of tools needed for a given query.

Source code in src/symfonic/memory/layers/procedural/router.py

def __init__(
    self,
    procedural_layer: ProceduralLayer,
    catalog: ToolCatalog,
    llm: Any,
    max_tools: int = 5,
    *,
    observability_hook: ObservabilityHook | None = None,
) -> None:
    self._procedural = procedural_layer
    self._catalog = catalog
    self._llm = llm
    self._max_tools = max_tools
    # v8.3.2 observability-bug fix (G5): the procedural-router LLM call
    # emitted ONLY on the merge-dependent callback manager (same bug
    # class as the v8.3.1 critic).  When a hook is supplied it also
    # emits on the always-on ``ObservabilityHook``.  ``None`` resolves
    # to ``NoOpObservabilityHook`` so the default is byte-identical.
    # NOTE: the live caller (``MemoryOrchestrator``) does NOT yet thread
    # a real hook -- see v8.3.2 report; this lets direct callers wire
    # one without invasive orchestrator plumbing.
    if observability_hook is None:
        from symfonic.core.observability.hooks import NoOpObservabilityHook

        observability_hook = NoOpObservabilityHook()
    self._observability_hook = observability_hook

route async

route(
    scope: TenantScope,
    query: str,
    conversation_history: list[Any] | None = None,
    *,
    callback_manager: CallbackManager | None = None,
    run_id: str = "",
    model_name: str = "procedural_router",
) -> RoutingResult

Route a query to appropriate tools.

Steps: 1. Query procedural layer for relevant skills. 2. Search catalog for matching tools. 3. Use LLM with structured output to select top tools. 4. Return only minimal schemas for selected tools.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
query	str	The user's query or request.	required
conversation_history	list[Any] | None	Optional recent conversation for context.	None
callback_manager	CallbackManager | None	v7.4.3 (Jarvio Ask 5) -- when provided, fires on_llm_end(node_name="procedural_router") after the LLM call so the (opt-in) capability router's tokens are visible to billing / OTel. None preserves byte- identical pre-7.4.3 behaviour.	None
run_id	str	Engine run identifier for callback correlation.	''
model_name	str	Optional model identifier stamped on the LLMEndEvent. Defaults to "procedural_router".	'procedural_router'

Returns:

Type	Description
RoutingResult	RoutingResult with intent, selected tools, and reasoning.

Source code in src/symfonic/memory/layers/procedural/router.py

async def route(
    self,
    scope: TenantScope,
    query: str,
    conversation_history: list[Any] | None = None,
    *,
    callback_manager: CallbackManager | None = None,
    run_id: str = "",
    model_name: str = "procedural_router",
) -> RoutingResult:
    """Route a query to appropriate tools.

    Steps:
    1. Query procedural layer for relevant skills.
    2. Search catalog for matching tools.
    3. Use LLM with structured output to select top tools.
    4. Return only minimal schemas for selected tools.

    Args:
        scope: Tenant scope for isolation.
        query: The user's query or request.
        conversation_history: Optional recent conversation for context.
        callback_manager: v7.4.3 (Jarvio Ask 5) -- when provided, fires
            ``on_llm_end(node_name="procedural_router")`` after the
            LLM call so the (opt-in) capability router's tokens are
            visible to billing / OTel.  ``None`` preserves byte-
            identical pre-7.4.3 behaviour.
        run_id: Engine run identifier for callback correlation.
        model_name: Optional model identifier stamped on the
            LLMEndEvent.  Defaults to ``"procedural_router"``.

    Returns:
        RoutingResult with intent, selected tools, and reasoning.
    """
    # Step 1: Query procedural layer for relevant skills
    skills = await self._procedural.query_skills(scope, query, top_k=5)
    skill_context = _format_skill_context(skills)

    # Step 2: Format tool catalog
    all_tools = self._catalog.list_all()
    catalog_text = "\n".join(
        f"- {t['tool_id']}: {t['description']}" for t in all_tools
    ) or "No tools registered."

    # Step 3: Build prompt and call LLM
    prompt_template = load_prompt("router")
    prompt = prompt_template.format(
        max_tools=self._max_tools,
        tool_catalog=catalog_text,
        skill_context=skill_context,
    )

    try:
        from symfonic.core.callbacks.emit import (
            _extract_usage,
            emit_llm_end_from_result,
            emit_llm_start,
            llm_timing,
            resolve_model_name,
        )

        _resolved_model = resolve_model_name(self._llm, model_name)
        _router_prompt = prompt + f"\n\nUser query: {query}"

        # v8.3.2 (G5): emit on the always-on ObservabilityHook BEFORE
        # the call (mirrors react.py:1078 / metacognitive.py:412).
        try:
            await self._observability_hook.on_llm_start(
                _resolved_model, [_router_prompt], run_id
            )
        except Exception:
            logger.exception("ObservabilityHook.on_llm_start failed")

        # v8.4.0 CALLBACK-PATH fix: open the OTel span on the merged
        # manager BEFORE the ainvoke so the existing END (:306) closes
        # it.  Same node_name; byte-identical no-op when no manager.
        if callback_manager is not None and not callback_manager.is_noop:
            await emit_llm_start(
                callback_manager,
                model=_resolved_model,
                messages=[_router_prompt],
                run_id=run_id,
                node_name="procedural_router",
            )

        # v7.15.0: capture timing so LLMEndEvent carries duration.
        async with llm_timing() as _timing:
            response = await self._llm.ainvoke(_router_prompt)

        # v8.3.2 (G5): emit on the always-on ObservabilityHook AFTER
        # the call (mirrors react.py:1433 / metacognitive.py:447).
        try:
            await self._observability_hook.on_llm_end(
                _resolved_model,
                str(getattr(response, "content", response)),
                _extract_usage(response),
                run_id,
            )
        except Exception:
            logger.exception("ObservabilityHook.on_llm_end failed")

        # v7.4.3 Jarvio Ask 5: emit on_llm_end so the (opt-in)
        # LLM-backed router's tokens show up in billing / OTel.
        # v7.14.0: stamp the resolved SKU instead of the
        # caller-passed name.  See callbacks/emit.py:resolve_model_name
        # for the provider-attribute survey + fallback semantics.
        await emit_llm_end_from_result(
            callback_manager,
            model=_resolved_model,
            result=response,
            run_id=run_id,
            node_name="procedural_router",
            timing=_timing,
        )

        content = response.content if hasattr(response, "content") else str(response)
        return self._parse_response(content, all_tools)
    except Exception:
        logger.exception("Router LLM call failed for tenant %s", scope.tenant_id)
        return RoutingResult(intent=query, reasoning="LLM routing failed, no tools selected")

RoutingResult

Bases: BaseModel

Result of capability routing.

Contains the detected intent, selected tool IDs and their minimal schemas, and the reasoning behind the selection.

render_skill_for_prompt

render_skill_for_prompt(entry: Any) -> str

Format a single skill for the router prompt's RELEVANT SKILLS section.

Three metadata['precondition'] shapes are honoured:

• Non-empty str (v7.5.1 shape) -- appended as a single indented PRE-FLIGHT: line. Renders byte-identical to pre-7.6.2 output; prompt-cache parity is preserved.
• Non-empty list[str] (v7.6.2 shape, Jarvio multi-precondition ask) -- one indented PRE-FLIGHT: line per entry, in supplied order. Empty / non-string entries are skipped silently. The router prompt's ordering clause at prompts/router.txt treats every PRE-FLIGHT line as a hard ordering constraint over required_tool_ids.
• Anything else (None, empty string, whitespace-only, non-string non-list metadata) -- the bullet renders single-line and no PRE-FLIGHT marker is emitted.

Skills without preconditions render byte-identical to pre-7.5.1 so prompt-cache parity holds for adopters who never use the authored tier.

v7.7.4 (Jarvio ask public-render-skill-helper): promoted from the v7.5.1 private _render_skill_for_prompt to a public helper.

v7.7.5 (Jarvio L2-prose architectural ask, Option D): bare- identifier preconditions render with a strictly stronger imperative phrasing -- PRE-FLIGHT REQUIRED: BEFORE calling \`X\`, YOU MUST first call \`Y\`. -- to lift compliance on the L2 path for adopters who don't opt into the v7.7.5 L1 render knob. Free-text preconditions fall through to the v7.5.1 PRE-FLIGHT: phrasing byte-identical (free text cannot be safely rewritten without inventing tool names). The bare-id discriminator is shared with the v7.7.4 gate via _extract_bare_identifier; the action tool is resolved via _skill_action_tool. Skills with no resolvable bare-id action tool fall back to free-text phrasing even when the precondition itself is a bare id -- refusing to invent.

Source code in src/symfonic/memory/layers/procedural/router.py

def render_skill_for_prompt(entry: Any) -> str:
    """Format a single skill for the router prompt's RELEVANT SKILLS section.

    Three ``metadata['precondition']`` shapes are honoured:

    * Non-empty ``str`` (v7.5.1 shape) -- appended as a single
      indented ``PRE-FLIGHT:`` line.  Renders byte-identical to
      pre-7.6.2 output; prompt-cache parity is preserved.
    * Non-empty ``list[str]`` (v7.6.2 shape, Jarvio multi-precondition
      ask) -- one indented ``PRE-FLIGHT:`` line per entry, in supplied
      order.  Empty / non-string entries are skipped silently.  The
      router prompt's ordering clause at ``prompts/router.txt`` treats
      every PRE-FLIGHT line as a hard ordering constraint over
      ``required_tool_ids``.
    * Anything else (``None``, empty string, whitespace-only,
      non-string non-list metadata) -- the bullet renders single-line
      and no PRE-FLIGHT marker is emitted.

    Skills without preconditions render byte-identical to pre-7.5.1 so
    prompt-cache parity holds for adopters who never use the authored
    tier.

    v7.7.4 (Jarvio ask `public-render-skill-helper`): promoted from
    the v7.5.1 private ``_render_skill_for_prompt`` to a public
    helper.

    v7.7.5 (Jarvio L2-prose architectural ask, Option D): bare-
    identifier preconditions render with a strictly stronger imperative
    phrasing -- ``PRE-FLIGHT REQUIRED: BEFORE calling \`X\`, YOU MUST
    first call \`Y\`.`` -- to lift compliance on the L2 path for
    adopters who don't opt into the v7.7.5 L1 render knob.  Free-text
    preconditions fall through to the v7.5.1 ``PRE-FLIGHT:`` phrasing
    byte-identical (free text cannot be safely rewritten without
    inventing tool names).  The bare-id discriminator is shared with
    the v7.7.4 gate via ``_extract_bare_identifier``; the action tool
    is resolved via ``_skill_action_tool``.  Skills with no resolvable
    bare-id action tool fall back to free-text phrasing even when the
    precondition itself is a bare id -- refusing to invent.
    """
    from symfonic.core.nodes.precondition_gate import (
        _extract_bare_identifier,
        _skill_action_tool,
    )

    steps = _skill_metadata_field(entry, "steps") or []
    content = getattr(entry, "content", "")
    line = f"- {content} (steps: {steps})"
    precondition = _skill_metadata_field(entry, "precondition")
    action_tool = _skill_action_tool(entry)
    if isinstance(precondition, str) and precondition.strip():
        pre_tool = _extract_bare_identifier(precondition)
        if pre_tool and action_tool:
            line += (
                f"\n  PRE-FLIGHT REQUIRED: BEFORE calling `{action_tool}`,"
                f" YOU MUST first call `{pre_tool}`."
            )
        else:
            line += f"\n  PRE-FLIGHT: {precondition.strip()}"
    elif isinstance(precondition, list):
        for item in precondition:
            if not isinstance(item, str) or not item.strip():
                continue
            pre_tool = _extract_bare_identifier(item)
            if pre_tool and action_tool:
                line += (
                    f"\n  PRE-FLIGHT REQUIRED: BEFORE calling"
                    f" `{action_tool}`, YOU MUST first call `{pre_tool}`."
                )
            else:
                line += f"\n  PRE-FLIGHT: {item.strip()}"
    return line