Skip to content

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 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
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 (adopter 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 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
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 (adopter 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 adopter 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, adopter 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 (adopter ask public-render-skill-helper): promoted from the v7.5.1 private _render_skill_for_prompt to a public helper.

v7.7.5 (adopter 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/skill_render.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def render_skill_for_prompt(entry: Any) -> str:
    r"""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, adopter 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 (adopter ask `public-render-skill-helper`): promoted from
    the v7.5.1 private ``_render_skill_for_prompt`` to a public
    helper.

    v7.7.5 (adopter 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.
    """
    # TA2.3: the skill-metadata readers live in the kernel-contracts module
    # that owns the vocabulary, not in the facade-compiler gate node that
    # enforces it -- ``capability -> facade-compiler`` is ``no``. Same two
    # functions, same definitions.
    from symfonic.core.contracts.skill_metadata import (
        extract_bare_identifier as _extract_bare_identifier,
    )
    from symfonic.core.contracts.skill_metadata import (
        skill_action_tool as _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