# backend/src/agent/_tool_resolver.py # #region AgentChat.ToolResolver [C:2] [TYPE Module] [SEMANTICS agent-chat,tools,resolution] # @defgroup AgentChat Tool resolution helpers for the LangGraph agent. # @LAYER Service # @RELATION DEPENDS_ON -> [AgentChat.Tools] # @RATIONALE Centralised tool resolution prevents duplication of tool-name matching logic. # Deterministic intent matching (infer_tool_from_text, fast_confirmation_tool, # keyword lists, negation guard, classification sets) removed — LLM handles # all intent detection through LangGraph tool-calling. Only utility helpers # (tool call coercion, args normalization) remain. # @REJECTED Deterministic intent matching — fragile substring collisions, maintenance burden # of 24 keyword lists across 3 files, negation blindness in fast-track, and # inability to handle synonyms ("панели"≠"дашборды") or typos ("дашборд"). # @REJECTED Fast-track confirmation — bypasses LLM, causing negation blindness. # @REJECTED Tool risk classification sets — LLM decides which tools to call; # LangGraph interrupt_before handles HITL for dangerous tools at graph level. from typing import Any # ── Graph nodes — used by confirmation subsystem to distinguish tools from infrastructure ── _GRAPH_NODE_NAMES = {"agent", "tools", "__start__", "__end__"} # #region AgentChat.ToolResolver.KnownNames [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,catalog] # @ingroup AgentChat # @BRIEF Return registered LangChain tool names. # @POST Returns set of tool name strings; falls back to empty set on failure. def known_agent_tool_names() -> set[str]: try: from src.agent.tools import get_all_tools return {str(tool_obj.name) for tool_obj in get_all_tools() if getattr(tool_obj, "name", None)} except Exception: return set() # #endregion AgentChat.ToolResolver.KnownNames # #region AgentChat.ToolResolver.NormalizeArgs [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,args] # @ingroup AgentChat # @BRIEF Normalize raw tool arguments into a plain dict regardless of input format. # @POST Returns dict (empty dict for None/unparseable input). def normalize_tool_args(raw_args: Any) -> dict[str, Any]: if raw_args is None: return {} if isinstance(raw_args, dict): return raw_args if hasattr(raw_args, "model_dump"): dumped = raw_args.model_dump() return dumped if isinstance(dumped, dict) else {} if hasattr(raw_args, "dict"): dumped = raw_args.dict() return dumped if isinstance(dumped, dict) else {} return {} # #endregion AgentChat.ToolResolver.NormalizeArgs # #region AgentChat.ToolResolver.CoerceToolCall [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,coerce] # @ingroup AgentChat # @BRIEF Extract (tool_name, tool_args) tuple from a dict or object tool call. # @POST Returns (name, args) tuple; name may be None if unparseable. def coerce_tool_call(tool_call: Any) -> tuple[str | None, dict[str, Any]]: if isinstance(tool_call, dict): return ( tool_call.get("name") or tool_call.get("tool") or tool_call.get("id"), normalize_tool_args(tool_call.get("args") or tool_call.get("input")), ) return ( getattr(tool_call, "name", None), normalize_tool_args(getattr(tool_call, "args", None)), ) # #endregion AgentChat.ToolResolver.CoerceToolCall # #region AgentChat.ToolResolver.ExtractFromState [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,checkpoint] # @ingroup AgentChat # @BRIEF Extract pending tool name and args from the LangGraph checkpoint. # @POST Returns (tool_name, args) tuple; (None, {}) if nothing found. # @RATIONALE LLM handles intent — no fallback to keyword inference. def extract_tool_call_from_state(state, user_text: str = "") -> tuple[str | None, dict[str, Any]]: known_tools = known_agent_tool_names() try: messages = (state.values.get("messages") if hasattr(state, "values") else []) or [] for msg in reversed(messages[-5:]): if hasattr(msg, "tool_calls") and msg.tool_calls: tool_name, tool_args = coerce_tool_call(msg.tool_calls[0]) if tool_name: return (str(tool_name), tool_args) except Exception: pass if getattr(state, "next", None): node_or_tool = str(state.next[0]) if node_or_tool in known_tools and node_or_tool not in _GRAPH_NODE_NAMES: return (node_or_tool, {}) return (None, {}) # #endregion AgentChat.ToolResolver.ExtractFromState # #region AgentChat.ToolResolver.FindTool [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,lookup] # @ingroup AgentChat # @BRIEF Find a registered LangChain tool by name. # @POST Returns tool object or None if not found. def find_tool(tool_name: str): from src.agent.tools import get_all_tools return next((tool_obj for tool_obj in get_all_tools() if getattr(tool_obj, "name", None) == tool_name), None) # #endregion AgentChat.ToolResolver.FindTool # #endregion AgentChat.ToolResolver