Skip to content

symfonic.capabilities.tools.stages.intent

intent

Intent-routing stage — per-turn narrowing from an intent verdict.

Extracted from symfonic.agent.triage.tool_routing.narrow_tools_for_intent (v7.0.1), which remains the public entry point and now delegates here so there is one implementation of the decision matrix rather than two.

IntentRoutingStage 

IntentRoutingStage(*, verdict: Any = None, trigger_keywords: Mapping[str, Sequence[str]] | None = None, always_include: Sequence[str] = DEFAULT_ALWAYS_INCLUDE)

Narrow the turn's tools from an IntentVerdict-shaped object.

Decision matrix (unchanged from v7.0.1):

  • no verdict, or ambiguous — abstain. A noisy signal must not cost recall.
  • knowledge — keep only the always-include names.
  • action — keep the verdict's matched tools, plus every tool the domain declares no trigger keywords for (pre-v6.1.8 semantics: an absent entry means "always include"), plus always-include.
Source code in src/symfonic/capabilities/tools/stages/intent.py 
41
42
43
44
45
46
47
48
49
50
def __init__(
    self,
    *,
    verdict: Any = None,
    trigger_keywords: Mapping[str, Sequence[str]] | None = None,
    always_include: Sequence[str] = DEFAULT_ALWAYS_INCLUDE,
) -> None:
    self._verdict = verdict
    self._trigger_keywords = dict(trigger_keywords or {})
    self._always_include = tuple(name for name in always_include if name)

decide 

decide(tools: Sequence[Any]) -> StageOutcome

The decision matrix, synchronously.

Split out because the v7.0.1 entry point (narrow_tools_for_intent) is synchronous and is called from inside a running event loop. One implementation, two callers — the alternative was asyncio.run from a coroutine, which raises, or a second copy of the matrix, which drifts.

Source code in src/symfonic/capabilities/tools/stages/intent.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
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 decide(self, tools: Sequence[Any]) -> StageOutcome:
    """The decision matrix, synchronously.

    Split out because the v7.0.1 entry point
    (``narrow_tools_for_intent``) is synchronous and is called from
    inside a running event loop. One implementation, two callers —
    the alternative was ``asyncio.run`` from a coroutine, which
    raises, or a second copy of the matrix, which drifts.
    """
    tools = tuple(tools)
    if not tools:
        return StageOutcome.abstain("no candidate tools")

    verdict = self._verdict
    label = getattr(verdict, "label", None)
    if verdict is None or label == "ambiguous":
        return StageOutcome.abstain("no actionable verdict")

    always = set(self._always_include)

    if label == "knowledge":
        # Unnamed tools survive here exactly as they do on the action
        # path: they cannot be gated by a name-keyed policy, and the
        # v7.0.1 helper kept them.
        return StageOutcome.select(
            filter_by_names(tools, always),
            "knowledge turn: always-include only",
        )

    keep = set(always)
    keep.update(getattr(verdict, "matched_tool_keywords", None) or [])
    for tool in tools:
        name = tool_name(tool)
        if name is None:
            continue
        if not self._trigger_keywords.get(name):
            keep.add(name)

    return StageOutcome.select(
        filter_by_names(tools, keep), f"{label} turn: keyword intersection",
    )