Download src/agents/planner/groups.py from DataEyond/Agentic-Service-Data-Eyond-Catalog: direct link, hf CLI and curl.
- Browser
- Download file 5.59 kB
-
https://huggingface.co/spaces/DataEyond/Agentic-Service-Data-Eyond-Catalog/resolve/main/src/agents/planner/groups.py
- Command line
-
hf download hf://spaces/DataEyond/Agentic-Service-Data-Eyond-Catalog/src/agents/planner/groups.py
-
curl -L -o groups.py https://huggingface.co/spaces/DataEyond/Agentic-Service-Data-Eyond-Catalog/resolve/main/src/agents/planner/groups.py
5.59 kB
| """Tool groups β the planner sees a slice, not the whole registry (A17). | |
| With every analytics tool active the Planner-visible set is **18 tools**, inside the | |
| measured 15β20 band where tool-selection accuracy starts to degrade β and the driver is | |
| semantic similarity, not raw count, which is exactly what this registry has | |
| (`analyze_trend` vs `analyze_forecast`, `analyze_correlation` vs `analyze_importance`, | |
| `analyze_segmentation` vs `analyze_cluster`). Rather than turn tools off, the router | |
| names a GROUP and the Planner is shown only that group's tools plus the always-on core: | |
| roughly 8, not 18. | |
| **Three layers, and it matters which does what** (decided 2026-09-17): | |
| router (LLM, already runs) -> emits group_hint on its existing structured output | |
| -> THIS MODULE (deterministic, 0 LLM) -> filters the registry | |
| -> planner (LLM) -> plans against the slice, unchanged | |
| The filter is deliberately NOT the planner's job: to choose a group it would have to be | |
| shown every group, which is the thing being avoided, and a second LLM call would break | |
| the two-call invariant for a benefit that only pays above ~60 tools. It is not the | |
| router's job either β the router classifies, it does not filter. | |
| **Fail open, always.** The router runs ~95 lines before the catalog is read | |
| (`chat_handler`: classify at 334, catalog at 429), so it is CATALOG-BLIND: it can judge | |
| what a question is *shaped* like, never what the data can support. Feasibility stays the | |
| Planner's job (recipes R6 and R15). And because a filtered-out tool is one the Planner cannot | |
| ask for, a wrong hint is unrecoverable β so an absent, unknown, or empty hint yields the | |
| FULL registry. Slightly worse selection always beats an unrecoverable miss. | |
| Groups are derived from each spec's existing `category` rather than stored on `ToolSpec`: | |
| grouping is a planner concern (the 2026-09-15 discussion: *"ini tuh dari planner-nya"*), | |
| `category` is already almost the taxonomy, and deriving keeps the tool layer untouched. | |
| An unmapped category falls back to `explore`, so a new tool is visible-by-default rather | |
| than silently invisible β the safe direction for this failure. | |
| """ | |
| from __future__ import annotations | |
| from .contracts import ToolRegistry, ToolSpec | |
| # --------------------------------------------------------------------------- # | |
| # The taxonomy (A0, 2026-09-17) | |
| # --------------------------------------------------------------------------- # | |
| DATA_ACCESS = "data_access" | |
| EXPLORE = "explore" | |
| RELATIONSHIP = "relationship" | |
| TIMESERIES = "timeseries" | |
| GROUPING = "grouping" | |
| EVALUATE = "evaluate" | |
| UTILITY = "utility" | |
| #: Groups the router may name. `data_access` and `utility` are never hinted β they are | |
| #: always included (see `ALWAYS_ON`), so hinting them would mean nothing. | |
| HINTABLE_GROUPS: tuple[str, ...] = (EXPLORE, RELATIONSHIP, TIMESERIES, GROUPING, EVALUATE) | |
| #: Always in scope whatever the hint: the four data-access tools (nothing can run | |
| #: without them) and `render_chart` (any group's answer may carry a viz tail). | |
| ALWAYS_ON = (DATA_ACCESS, UTILITY) | |
| #: One line per group, for the router prompt. Written for a classifier that has the | |
| #: QUESTION and no catalog β phrased by what the user is asking for, not by method. | |
| GROUP_PURPOSE: dict[str, str] = { | |
| EXPLORE: "summarising, totalling or checking the data itself", | |
| RELATIONSHIP: "what relates to, explains or contributes to a measure", | |
| TIMESERIES: "movement over time, including projecting it forward", | |
| GROUPING: "splitting rows into segments or discovering groups", | |
| EVALUATE: "comparing two groups, or finding unusual rows", | |
| } | |
| _CATEGORY_TO_GROUP: dict[str, str] = { | |
| # data access β always on | |
| "catalog.introspection": DATA_ACCESS, | |
| "analytics.query": DATA_ACCESS, | |
| "retrieval.documents": DATA_ACCESS, | |
| # explore | |
| "analytics.descriptive": EXPLORE, | |
| "analytics.aggregation": EXPLORE, | |
| "analytics.quality": EXPLORE, | |
| # relationship | |
| "analytics.relationship": RELATIONSHIP, # correlation + importance | |
| "analytics.pareto": RELATIONSHIP, | |
| # timeseries | |
| "analytics.timeseries": TIMESERIES, # trend + forecast | |
| # grouping | |
| "analytics.segmentation": GROUPING, | |
| "analytics.clustering": GROUPING, | |
| # evaluate | |
| "analytics.comparison": EVALUATE, # ab_test | |
| "analytics.anomaly": EVALUATE, | |
| # utility β always on | |
| "analytics.combine": UTILITY, | |
| "analytics.visualization": UTILITY, | |
| } | |
| #: An unmapped category lands here. Visible-by-default: a tool nobody grouped should be | |
| #: selectable, not silently withheld. | |
| _UNGROUPED = EXPLORE | |
| def group_of(spec: ToolSpec) -> str: | |
| """The group a spec belongs to, derived from its `category`.""" | |
| return _CATEGORY_TO_GROUP.get(spec.category, _UNGROUPED) | |
| def scope(registry: ToolRegistry, hints: list[str] | None) -> ToolRegistry: | |
| """The slice of `registry` a turn hinted at `hints` should plan against. | |
| Returns the FULL registry when `hints` is empty, None, or names nothing known β | |
| the fail-open rule. Otherwise: everything in `ALWAYS_ON` plus every hinted group, | |
| preserving the registry's own ordering so the planner prompt stays stable. | |
| """ | |
| wanted = {h for h in (hints or []) if h in HINTABLE_GROUPS} | |
| if not wanted: | |
| return ToolRegistry(tools=list(registry.tools)) | |
| keep = wanted | set(ALWAYS_ON) | |
| return ToolRegistry(tools=[t for t in registry.tools if group_of(t) in keep]) | |
| def render_group_menu() -> str: | |
| """The group list for the router prompt, one line each.""" | |
| return "\n".join(f"- `{g}` β {GROUP_PURPOSE[g]}" for g in HINTABLE_GROUPS) | |