Engine-side helpers for wiring the IntentClassifier (v7.0 T07, v7.0.1).
Keeps the hot-path integration code outside engine.py (which sits
under the documented v7.1-T01 size waiver). The engine imports
run_intent_filter, enabled_layers_for_verdict, and
apply_tool_routing so the hydration + tool-binding paths stay
legible.
apply_tool_routing(
*,
verdict: IntentVerdict | None,
all_tools: Sequence[Any],
domain: DomainTemplate,
config: FrameworkConfig,
run_id: str,
callback_manager: Any | None,
) -> list[Any] | None
Populate resolved_tools for the current turn (v7.0.1).
Returns the list that the engine should write into
state["resolved_tools"], or None if no narrowing was
applied (off mode OR observe mode — observe emits
telemetry but does NOT narrow the bound list).
Behaviour matrix:
tool_routing_mode == "off" — returns None immediately.
Classifier is not consulted. Zero extra work on the hot path.tool_routing_mode == "observe" — narrower runs, event
fires, None is returned so react.py still binds the
full static manifest.tool_routing_mode == "enforce" — narrower runs, event
fires, narrowed list returned. When no narrowing occurred
(verdict missing / ambiguous / conservative fallback) returns
None so the engine doesn't waste a state write on the
full manifest.
Source code in src/symfonic/agent/triage/wiring.py
| async def apply_tool_routing(
*,
verdict: IntentVerdict | None,
all_tools: Sequence[Any],
domain: DomainTemplate,
config: FrameworkConfig,
run_id: str,
callback_manager: Any | None,
) -> list[Any] | None:
"""Populate ``resolved_tools`` for the current turn (v7.0.1).
Returns the list that the engine should write into
``state["resolved_tools"]``, or ``None`` if no narrowing was
applied (``off`` mode OR ``observe`` mode — ``observe`` emits
telemetry but does NOT narrow the bound list).
Behaviour matrix:
- ``tool_routing_mode == "off"`` — returns ``None`` immediately.
Classifier is not consulted. Zero extra work on the hot path.
- ``tool_routing_mode == "observe"`` — narrower runs, event
fires, ``None`` is returned so ``react.py`` still binds the
full static manifest.
- ``tool_routing_mode == "enforce"`` — narrower runs, event
fires, narrowed list returned. When no narrowing occurred
(verdict missing / ambiguous / conservative fallback) returns
``None`` so the engine doesn't waste a state write on the
full manifest.
"""
mode = config.tool_routing_mode
if mode == "off":
return None
if not all_tools:
return None
# v7.0.3 F4: thread the operator-configurable ``always_include_tools``
# through the narrower. Default ``("hydrate_context",)`` preserves
# v7.0.1 / v7.0.2 byte-for-byte. Domain templates and fabric can
# now pin additional low-cost tools without touching keyword maps.
always_include = _resolve_always_include(config)
narrowed = narrow_tools_for_intent(
verdict, all_tools, domain, always_include=always_include,
)
# Preserve identity short-circuit: when the helper returned the
# full manifest (fallback path) we skip the event AND the state
# write. ``narrow_tools_for_intent`` returns ``list(all_tools)``
# for None/ambiguous; compare by length + ordered names.
if _is_full_manifest(narrowed, all_tools):
return None
dropped = _diff_by_name(all_tools, narrowed)
if callback_manager is not None and callback_manager.has_hook(
"on_tool_routing_decision",
):
event = ToolRoutingDecisionEvent(
run_id=run_id,
mode=mode,
intent_label=verdict.label if verdict is not None else "unknown",
total_tools=len(all_tools),
narrowed_tool_names=[_tool_name(t) for t in narrowed],
saved_tokens_estimate=estimate_dropped_tokens(dropped),
)
try:
await callback_manager.on_tool_routing_decision(event)
except Exception:
logger.exception(
"on_tool_routing_decision dispatch raised; continuing",
)
if mode == "enforce":
return narrowed
# observe mode: telemetry only, keep full manifest bound.
return None
|
enabled_layers_for_verdict
enabled_layers_for_verdict(
config: FrameworkConfig, verdict: IntentVerdict | None
) -> set[str]
Return the layer set the hydration loop should honour.
off / observe: verdict is informational, return the
caller-configured layer set unchanged.enforce:
action — drop semantic / episodic / procedural so the turn
hydrates only from working + prospective.knowledge / ambiguous / None — caller-configured
layers unchanged (conservative fallback).
Source code in src/symfonic/agent/triage/wiring.py
| def enabled_layers_for_verdict(
config: FrameworkConfig,
verdict: IntentVerdict | None,
) -> set[str]:
"""Return the layer set the hydration loop should honour.
- ``off`` / ``observe``: verdict is informational, return the
caller-configured layer set unchanged.
- ``enforce``:
* ``action`` — drop semantic / episodic / procedural so the turn
hydrates only from working + prospective.
* ``knowledge`` / ``ambiguous`` / ``None`` — caller-configured
layers unchanged (conservative fallback).
"""
base = set(config.enabled_layers)
if config.intent_filter_mode != "enforce" or verdict is None:
return base
if verdict.label != "action":
return base
return base - _ACTION_SKIPPED_LAYERS
|
run_intent_filter
async
run_intent_filter(
*,
classifier: IntentClassifier | None,
config: FrameworkConfig,
domain: DomainTemplate,
query: str,
callbacks: Sequence[Any] | None,
) -> IntentVerdict | None
Classify query and fan the result out via callback handlers.
Returns the verdict on success; returns None when the classifier
is disabled or missing. Never raises — upstream engine logic
treats None as a soft miss and falls back to full hydration.
Source code in src/symfonic/agent/triage/wiring.py
| async def run_intent_filter(
*,
classifier: IntentClassifier | None,
config: FrameworkConfig,
domain: DomainTemplate,
query: str,
callbacks: Sequence[Any] | None,
) -> IntentVerdict | None:
"""Classify ``query`` and fan the result out via callback handlers.
Returns the verdict on success; returns ``None`` when the classifier
is disabled or missing. Never raises — upstream engine logic
treats ``None`` as a soft miss and falls back to full hydration.
"""
if classifier is None or not should_run_classifier(config):
return None
try:
verdict = classifier.classify(query, domain=domain)
except Exception:
logger.warning(
"IntentClassifier.classify raised; falling back to full hydration",
exc_info=True,
)
return None
# Only emit the intent_classified ExtensionEvent when the hydration-
# side filter is the reason we ran. If the classifier only fired
# because ``tool_routing_mode`` needed a verdict, the verdict is
# still returned to the caller but no ``intent_classified`` event
# is broadcast -- the tool-routing path emits its own
# ``ToolRoutingDecisionEvent`` via the callback manager instead.
if config.intent_filter_mode in {"observe", "enforce"}:
await _emit_intent_event(verdict, config, callbacks)
return verdict
|
should_run_classifier
should_run_classifier(config: FrameworkConfig) -> bool
True when either intent filter OR tool routing needs the verdict.
v7.0.1: tool_routing_mode consumes the same verdict the
hydration filter does. If either knob is in observe or
enforce we need a fresh classifier pass.
Source code in src/symfonic/agent/triage/wiring.py
| def should_run_classifier(config: FrameworkConfig) -> bool:
"""True when either intent filter OR tool routing needs the verdict.
v7.0.1: ``tool_routing_mode`` consumes the same verdict the
hydration filter does. If either knob is in ``observe`` or
``enforce`` we need a fresh classifier pass.
"""
return (
config.intent_filter_mode in {"observe", "enforce"}
or config.tool_routing_mode in {"observe", "enforce"}
)
|