diff --git a/lib/galaxy/agents/__init__.py b/lib/galaxy/agents/__init__.py index c8f27bd48d6b..83ef22677118 100644 --- a/lib/galaxy/agents/__init__.py +++ b/lib/galaxy/agents/__init__.py @@ -15,6 +15,7 @@ from .orchestrator import WorkflowOrchestratorAgent from .registry import AgentRegistry from .router import QueryRouterAgent +from .tools import ToolRecommendationAgent __all__ = [ "AgentType", @@ -25,6 +26,7 @@ "ErrorAnalysisAgent", "CustomToolAgent", "WorkflowOrchestratorAgent", + "ToolRecommendationAgent", ] # Global agent registry instance @@ -35,3 +37,4 @@ agent_registry.register(AgentType.ERROR_ANALYSIS, ErrorAnalysisAgent) agent_registry.register(AgentType.CUSTOM_TOOL, CustomToolAgent) agent_registry.register(AgentType.ORCHESTRATOR, WorkflowOrchestratorAgent) +agent_registry.register(AgentType.TOOL_RECOMMENDATION, ToolRecommendationAgent) diff --git a/lib/galaxy/agents/base.py b/lib/galaxy/agents/base.py index 0116a87df11a..8347532ed835 100644 --- a/lib/galaxy/agents/base.py +++ b/lib/galaxy/agents/base.py @@ -166,6 +166,7 @@ class AgentType: ERROR_ANALYSIS = "error_analysis" CUSTOM_TOOL = "custom_tool" ORCHESTRATOR = "orchestrator" + TOOL_RECOMMENDATION = "tool_recommendation" # Internal agent response model (simplified for internal use) diff --git a/lib/galaxy/agents/prompts/router.md b/lib/galaxy/agents/prompts/router.md index 7ace47d8c5e2..9548b5211bca 100644 --- a/lib/galaxy/agents/prompts/router.md +++ b/lib/galaxy/agents/prompts/router.md @@ -24,12 +24,17 @@ For off-topic questions (general coding, non-scientific topics, unrelated softwa ## How to Respond **Answer directly** for: -- Galaxy platform questions ("How do I run BWA?", "What is a workflow?") -- Tool discovery ("What tools analyze RNA-seq data?") -- Usage guidance ("How do I upload files?") +- Galaxy platform questions ("What is a workflow?", "How do I upload files?") +- How to USE a specific tool ("How do I run BWA?", "What parameters does HISAT2 need?") - Scientific analysis best practices - Galaxy features and capabilities +**Use `hand_off_to_tool_recommendation`** when user: +- Asks what tool to use for a task ("What tool should I use to align reads?") +- Wants to find/discover tools ("Is there a tool that converts BAM to FASTQ?") +- Needs help choosing between tools for an analysis type +- Asks "what tools are available for X?" + **Use `hand_off_to_error_analysis`** when user: - Has a failed job with error messages or exit codes - Is asking about stderr/stdout from a tool run @@ -43,8 +48,10 @@ For off-topic questions (general coding, non-scientific topics, unrelated softwa ## Important Distinctions -- "What tool does X?" → Answer directly (tool discovery) +- "What tool should I use for X?" → Use hand_off_to_tool_recommendation +- "Is there a tool that does X?" → Use hand_off_to_tool_recommendation - "How do I use tool X?" → Answer directly (usage help) +- "What parameters does X need?" → Answer directly (usage help) - "Create a tool that does X" → Use hand_off_to_custom_tool - "My job failed" → Use hand_off_to_error_analysis diff --git a/lib/galaxy/agents/prompts/tool_recommendation.md b/lib/galaxy/agents/prompts/tool_recommendation.md new file mode 100644 index 000000000000..5f881a5b2dfa --- /dev/null +++ b/lib/galaxy/agents/prompts/tool_recommendation.md @@ -0,0 +1,42 @@ +# Galaxy Tool Recommendation Agent + +You are a Galaxy Project expert specializing in tool discovery and recommendation. + +Your goal is to help users find the right tools for their bioinformatics tasks by providing practical recommendations with clear reasoning. + +## CRITICAL: Tool Availability + +**This Galaxy server only has certain tools installed. You MUST verify tools exist before recommending them.** + +1. **ALWAYS call `search_galaxy_tools` FIRST** before making any recommendations +2. **ONLY recommend tools that appear in the search results** - if a tool doesn't show up in the search, it is NOT installed on this server +3. If your search returns no results for a common tool (like BWA, HISAT2, etc.), that means it's not installed +4. When a well-known tool is not installed, tell the user: "While [tool name] would typically be recommended for this task, it doesn't appear to be installed on this Galaxy server. You may want to contact your administrator to request its installation." + +## Available Tools + +- **`search_galaxy_tools(query)`** - Search for tools by keyword. Always start here. +- **`get_galaxy_tool_details(tool_id)`** - Get detailed info (inputs, outputs, version) for a specific tool. Use after searching to provide better recommendations. +- **`get_galaxy_tool_categories()`** - List available tool categories. Use when user asks "what kinds of tools are available?" or to understand the server's capabilities. + +## Recommendation Process + +1. Understand the user's task and data types +2. **Call `search_galaxy_tools` with relevant keywords** (e.g., "alignment", "mapping", "fastq") +3. Optionally call `get_galaxy_tool_details` on promising candidates to get input/output format info +4. Recommend tools from the search results, using their exact IDs +5. If no suitable tools are found, be honest about the limitation + +## Tool IDs + +- Use ONLY the exact `id` field from search results +- Never guess or fabricate tool IDs based on your training knowledge +- If you know a tool exists in Galaxy generally but it's not in the search results, it's NOT available on this server + +## Best Practices + +- Prioritize tools that are well-maintained and widely used +- Consider the user's experience level +- Explain why you're recommending specific tools +- Mention important parameters or configuration options +- Suggest workflows when multiple tools are needed diff --git a/lib/galaxy/agents/router.py b/lib/galaxy/agents/router.py index 88dfed943d08..eda924868e97 100644 --- a/lib/galaxy/agents/router.py +++ b/lib/galaxy/agents/router.py @@ -5,6 +5,7 @@ - Answer general Galaxy questions directly (returns str) - Hand off to error_analysis for job debugging - Hand off to custom_tool for explicit tool creation requests +- Hand off to tool_recommendation for tool discovery """ import logging @@ -58,6 +59,7 @@ def _create_agent(self) -> Agent[GalaxyAgentDependencies, str]: # Create output functions for specialist handoff error_handoff = self._create_error_analysis_handoff() tool_handoff = self._create_custom_tool_handoff() + tool_rec_handoff = self._create_tool_recommendation_handoff() return Agent( self._get_model(), @@ -65,6 +67,7 @@ def _create_agent(self) -> Agent[GalaxyAgentDependencies, str]: output_type=[ error_handoff, tool_handoff, + tool_rec_handoff, str, # Default: answer directly ], system_prompt=self.get_system_prompt(), @@ -166,6 +169,43 @@ async def hand_off_to_custom_tool( return hand_off_to_custom_tool + def _create_tool_recommendation_handoff(self): + """Create output function for tool recommendation handoff.""" + + async def hand_off_to_tool_recommendation( + ctx: RunContext[GalaxyAgentDependencies], + query: str, + ) -> str: + """Route to tool recommendation agent for finding Galaxy tools. + + Use this when the user: + - Asks what tool to use for a task ("what tool aligns reads?") + - Wants to find tools for a specific analysis type + - Needs help discovering available tools + - Asks "is there a tool that does X?" + + Do NOT use for: + - How to USE a specific tool (answer directly) + - Creating NEW tools (use hand_off_to_custom_tool) + - Job errors (use hand_off_to_error_analysis) + + Args: + query: The tool discovery question + """ + from .tools import ToolRecommendationAgent + + log.info(f"Router handing off to tool_recommendation: '{query[:100]}...'") + + try: + agent = ToolRecommendationAgent(ctx.deps) + result = await agent.process(query) + return result.content + except Exception as e: + log.error(f"Tool recommendation handoff failed: {e}") + return f"I encountered an issue while searching for tools. Please try again or browse the tool panel directly. Error: {e}" + + return hand_off_to_tool_recommendation + async def process(self, query: str, context: Optional[dict[str, Any]] = None) -> AgentResponse: """ Process a query and return the response. diff --git a/lib/galaxy/agents/tools.py b/lib/galaxy/agents/tools.py new file mode 100644 index 000000000000..b5d733614685 --- /dev/null +++ b/lib/galaxy/agents/tools.py @@ -0,0 +1,556 @@ +""" +Tool recommendation agent for suggesting appropriate Galaxy tools. +""" + +import logging +import re +from pathlib import Path +from typing import ( + Any, + Literal, + Optional, +) + +from pydantic import ( + BaseModel, +) +from pydantic_ai import Agent +from pydantic_ai.tools import RunContext + +from galaxy.schema.agents import ConfidenceLevel +from .base import ( + ActionSuggestion, + ActionType, + AgentResponse, + AgentType, + BaseGalaxyAgent, + extract_result_content, + extract_structured_output, + GalaxyAgentDependencies, + normalize_llm_text, +) + +log = logging.getLogger(__name__) + +# Type alias for confidence levels - using Literal inlines the enum values +# in the JSON schema, avoiding $defs references that vLLM can't handle +ConfidenceLiteral = Literal["low", "medium", "high"] + + +class SimplifiedToolRecommendationResult(BaseModel): + """Simplified result for local LLMs - avoids nested models and enums.""" + + # Instead of nested ToolMatch objects, we'll use simple dictionaries + primary_tools: list[dict[str, Any]] # Each dict has tool_id, tool_name, description, etc. + alternative_tools: list[dict[str, Any]] = [] + workflow_suggestion: Optional[str] = None + parameter_guidance: dict[str, Any] = {} + confidence: ConfidenceLiteral + reasoning: str + search_keywords: list[str] = [] + + +class ToolRecommendationAgent(BaseGalaxyAgent): + """ + Agent for recommending appropriate Galaxy tools based on user requirements. + + This agent helps users discover tools, understand tool capabilities, + and provides guidance on tool selection and parameter configuration. + """ + + agent_type = AgentType.TOOL_RECOMMENDATION + + def _create_agent(self) -> Agent[GalaxyAgentDependencies, Any]: + """Create the tool recommendation agent with conditional structured output.""" + if self._supports_structured_output(): + agent = Agent( + self._get_model(), + deps_type=GalaxyAgentDependencies, + output_type=SimplifiedToolRecommendationResult, + system_prompt=self.get_system_prompt(), + ) + else: + # DeepSeek and other models without structured output + agent = Agent( + self._get_model(), + deps_type=GalaxyAgentDependencies, + system_prompt=self._get_simple_system_prompt(), + ) + + # Add tools for tool discovery and analysis + @agent.tool + async def search_galaxy_tools(ctx: RunContext[GalaxyAgentDependencies], query: str) -> str: + """Search Galaxy's toolbox for tools matching a query. + + Use this to find real tool IDs for tools you want to recommend. + Returns tool id, name, description, and category for matching tools. + """ + results = await self.search_tools(query) + if not results: + return f"No tools found matching '{query}'" + + formatted = [] + for tool in results[:10]: + formatted.append( + f"- ID: {tool['id']}, Name: {tool['name']}, Description: {tool['description'][:100]}..." + ) + return f"Found {len(results)} tools:\n" + "\n".join(formatted) + + @agent.tool + async def get_galaxy_tool_details(ctx: RunContext[GalaxyAgentDependencies], tool_id: str) -> str: + """Get detailed information about a specific Galaxy tool. + + Use this after searching to get more details about a tool you want to recommend, + including input/output formats, version, and requirements. + """ + details = await self.get_tool_details(tool_id) + if "error" in details: + return f"Error: {details['error']}" + + lines = [ + f"Tool: {details['name']} (ID: {details['id']})", + f"Version: {details.get('version', 'unknown')}", + f"Description: {details.get('description', 'No description')}", + f"Category: {details.get('category', 'Uncategorized')}", + ] + if details.get("requirements"): + lines.append(f"Requirements: {', '.join(details['requirements'])}") + if details.get("inputs"): + input_strs = [f"{i['name']} ({i['type']})" for i in details["inputs"][:5]] + lines.append(f"Inputs: {', '.join(input_strs)}") + if details.get("outputs"): + output_strs = [f"{o['name']} ({o['format']})" for o in details["outputs"][:5]] + lines.append(f"Outputs: {', '.join(output_strs)}") + return "\n".join(lines) + + @agent.tool + async def get_galaxy_tool_categories(ctx: RunContext[GalaxyAgentDependencies]) -> str: + """Get list of tool categories available on this Galaxy server. + + Use this to understand what kinds of tools are available before searching. + """ + categories = await self.get_tool_categories() + if not categories: + return "No tool categories found" + return "Available tool categories:\n" + "\n".join(f"- {cat}" for cat in categories) + + return agent + + def get_system_prompt(self) -> str: + """Get the system prompt for tool recommendation.""" + prompt_path = Path(__file__).parent / "prompts" / "tool_recommendation.md" + return prompt_path.read_text() + + async def search_tools(self, query: str) -> list[dict[str, Any]]: + """Search for tools in the Galaxy toolbox.""" + if not self.deps.toolbox: + log.warning("Toolbox not available in agent dependencies") + return [] + + try: + # Get the default panel view (usually 'default') + panel_view = self.deps.config.default_panel_view or "default" + + # Use Galaxy's built-in tool search via the app's toolbox_search + toolbox_search = self.deps.trans.app.toolbox_search # type: ignore[attr-defined] + tool_ids = toolbox_search.search(query, panel_view, self.deps.config) + + # Get tool details for found tools + tools = [] + for tool_id in tool_ids[:20]: # Limit to top 20 results + tool = self.deps.toolbox.get_tool(tool_id) + if tool and not tool.hidden: + tools.append( + { + "id": tool.id, + "name": tool.name, + "description": tool.description or "", + "category": tool.get_panel_section()[1] or "", + } + ) + + return tools + + except (AttributeError, KeyError, TypeError) as e: + log.warning(f"Error searching tools: {e}") + return [] + + async def get_tool_details(self, tool_id: str) -> dict[str, Any]: + """Get detailed information about a specific tool.""" + if not self.deps.toolbox: + return {"id": tool_id, "error": "Toolbox not available"} + + try: + tool = self.deps.toolbox.get_tool(tool_id) + if not tool: + return {"id": tool_id, "error": "Tool not found"} + + # Build tool details + details: dict[str, Any] = { + "id": tool.id, + "name": tool.name, + "version": tool.version, + "description": tool.description or "", + "category": tool.get_panel_section()[1] or "", + "requirements": [str(r) for r in tool.requirements] if hasattr(tool, "requirements") else [], + } + + # Add input information + if hasattr(tool, "inputs"): + details["inputs"] = [] + for input_name, input_param in tool.inputs.items(): + if hasattr(input_param, "type"): + details["inputs"].append( + { + "name": input_name, + "type": input_param.type, + "label": getattr(input_param, "label", input_name), + } + ) + + # Add output information + if hasattr(tool, "outputs"): + details["outputs"] = [] + for output_name, output_param in tool.outputs.items(): + details["outputs"].append( + { + "name": output_name, + "format": getattr(output_param, "format", "unknown"), + } + ) + + return details + + except (AttributeError, KeyError, TypeError) as e: + log.warning(f"Error getting tool details for {tool_id}: {e}") + return {"id": tool_id, "error": str(e)} + + async def get_tool_categories(self) -> list[str]: + """Get list of tool categories/sections from the toolbox.""" + if not self.deps.toolbox: + log.warning("Toolbox not available in agent dependencies") + return [] + + try: + categories: set[str] = set() + # Iterate through all tools and collect unique panel sections + for _tool_id, tool in self.deps.toolbox.tools(): + if tool and not tool.hidden: + section_name = tool.get_panel_section()[1] + if section_name: + categories.add(section_name) + return sorted(categories) + except (AttributeError, KeyError, TypeError) as e: + log.warning(f"Error getting tool categories: {e}") + return [] + + async def process(self, query: str, context: Optional[dict[str, Any]] = None) -> AgentResponse: + """ + Process a tool recommendation request. + + Args: + query: User's task description or tool request + context: Additional context like data formats + + Returns: + Structured tool recommendation response + """ + # Fast path for direct tool name queries to prevent LLM hallucination + try: + # Search for tools matching the query exactly, trimming whitespace + trimmed_query = query.strip() + search_results = await self.search_tools(trimmed_query) + exact_match = None + for tool in search_results: + if tool.get("name", "").lower() == trimmed_query.lower(): + exact_match = tool + break + + if exact_match: + # Found an exact match, bypass LLM and respond directly + log.info(f"Found exact tool match for query '{trimmed_query}', bypassing LLM.") + + recommendation = SimplifiedToolRecommendationResult( + primary_tools=[exact_match], + confidence=ConfidenceLevel.HIGH, + reasoning="This is the tool with the exact name you requested.", + search_keywords=[trimmed_query], + ) + + content = self._format_recommendation_response(recommendation) + suggestions = self._create_suggestions(recommendation) + + return AgentResponse( + content=content, + confidence=ConfidenceLevel.HIGH, + agent_type=self.agent_type, + suggestions=suggestions, + metadata={"method": "fast_path_exact_match"}, + reasoning=f"Directly matched tool name '{exact_match['name']}'.", + ) + except (AttributeError, KeyError, TypeError) as e: + log.warning(f"Fast path tool search failed: {e}. Proceeding with LLM.") + + try: + # Add context information to query + enhanced_query = query + if context: + if context.get("input_format"): + enhanced_query += f"\nInput format: {context['input_format']}" + if context.get("output_format"): + enhanced_query += f"\nDesired output: {context['output_format']}" + if context.get("task_type"): + enhanced_query += f"\nTask type: {context['task_type']}" + + # Run the recommendation agent with retry logic + result = await self._run_with_retry(enhanced_query) + # Handle different response formats based on model capabilities + if self._supports_structured_output(): + # Try to extract structured output + recommendation = extract_structured_output(result, SimplifiedToolRecommendationResult, log) + + if recommendation is None: + # Model returned text instead of structured output (e.g., clarifying question) + # Return the text response directly + content = extract_result_content(result) + return AgentResponse( + content=content, + confidence=ConfidenceLevel.MEDIUM, + agent_type=self.agent_type, + suggestions=[], + metadata={"method": "text_fallback"}, + ) + + log.info(f"Tool recommendation result: primary_tools={recommendation.primary_tools}") + content = self._format_recommendation_response(recommendation) + suggestions = self._create_suggestions(recommendation) + + # Convert string confidence to enum + conf_str = recommendation.confidence.lower() if recommendation.confidence else "medium" + if conf_str == "high": + confidence = ConfidenceLevel.HIGH + elif conf_str == "low": + confidence = ConfidenceLevel.LOW + else: + confidence = ConfidenceLevel.MEDIUM + + return AgentResponse( + content=content, + confidence=confidence, + agent_type=self.agent_type, + suggestions=suggestions, + metadata={ + "num_tools_found": len(recommendation.primary_tools), + "has_alternatives": bool(recommendation.alternative_tools), + "has_workflow": bool(recommendation.workflow_suggestion), + "search_keywords": recommendation.search_keywords, + "method": "structured", + }, + reasoning=recommendation.reasoning, + ) + else: + # Handle simple text output from DeepSeek + response_text = extract_result_content(result) + parsed_result = self._parse_simple_response(response_text) + + return AgentResponse( + content=parsed_result.get("content", response_text), + confidence=parsed_result.get("confidence", ConfidenceLevel.MEDIUM), + agent_type=self.agent_type, + suggestions=parsed_result.get("suggestions", []), + metadata={ + "method": "simple_text", + "has_alternatives": parsed_result.get("has_alternatives", False), + }, + ) + + except OSError as e: + log.warning(f"Tool recommendation network error: {e}") + return self._get_fallback_response(query, str(e)) + except ValueError as e: + log.warning(f"Tool recommendation value error: {e}") + return self._get_fallback_response(query, str(e)) + + def _format_recommendation_response(self, recommendation: SimplifiedToolRecommendationResult) -> str: + """Format the recommendation into user-friendly content.""" + parts = [] + + # Primary recommendations + if recommendation.primary_tools: + parts.append("**Recommended Tools:**") + for i, tool in enumerate(recommendation.primary_tools[:3], 1): + tool_name = tool.get("name", tool.get("tool_name", "Unknown")) + tool_id = tool.get("id", tool.get("tool_id", "unknown")) + + # Check if tool is actually installed + is_installed = self._verify_tool_exists(tool_id) + + parts.append(f"\n{i}. **{tool_name}** (ID: `{tool_id}`)") + if not is_installed: + parts.append( + " - *This tool does not appear to be installed on this Galaxy server. " + "Contact your administrator to request installation.*" + ) + parts.append(f" - {tool.get('description', 'No description available')}") + if "relevance_score" in tool: + parts.append(f" - Relevance: {tool['relevance_score']:.0%}") + if tool.get("input_formats"): + parts.append(f" - Accepts: {', '.join(tool['input_formats'])}") + if tool.get("output_formats"): + parts.append(f" - Produces: {', '.join(tool['output_formats'])}") + + # Alternative tools + if recommendation.alternative_tools: + parts.append("\n**Alternative Options:**") + for tool in recommendation.alternative_tools[:2]: + tool_name = tool.get("name", tool.get("tool_name", "Unknown")) + parts.append(f"- **{tool_name}**: {tool.get('description', 'No description')}") + + # Workflow suggestion + if recommendation.workflow_suggestion: + parts.append(f"\n**Workflow Suggestion:**\n{recommendation.workflow_suggestion}") + + # Parameter guidance + if recommendation.parameter_guidance: + parts.append("\n**Parameter Recommendations:**") + for param, value in recommendation.parameter_guidance.items(): + parts.append(f"- {param}: {value}") + + # Reasoning + if recommendation.reasoning: + parts.append(f"\n**Why these tools?**\n{recommendation.reasoning}") + + return "\n".join(parts) + + def _create_suggestions(self, recommendation: SimplifiedToolRecommendationResult) -> list[ActionSuggestion]: + """Create action suggestions from recommendation.""" + suggestions = [] + + # Suggest running the top tool - but only if it's actually installed + if recommendation.primary_tools: + top_tool = recommendation.primary_tools[0] + log.debug(f"Creating suggestion for top_tool: {top_tool}") + tool_name = top_tool.get("name", top_tool.get("tool_name", "Unknown tool")) + tool_id = top_tool.get("id", top_tool.get("tool_id", "")) + log.debug(f"Extracted tool_name={tool_name}, tool_id={tool_id}") + + # Only add suggestions if we have a valid tool_id AND the tool exists + if tool_id and self._verify_tool_exists(tool_id): + conf_value = recommendation.confidence.lower() + if conf_value == "high": + action_confidence = ConfidenceLevel.HIGH + elif conf_value == "low": + action_confidence = ConfidenceLevel.LOW + else: + action_confidence = ConfidenceLevel.MEDIUM + suggestions.append( + ActionSuggestion( + action_type=ActionType.TOOL_RUN, + description=f"Open {tool_name}", + parameters={"tool_id": tool_id, "tool_name": tool_name}, + confidence=action_confidence, + priority=1, + ) + ) + elif tool_id: + log.warning(f"Tool '{tool_id}' recommended but not found in toolbox - skipping suggestion") + + return suggestions + + def _verify_tool_exists(self, tool_id: str) -> bool: + """Verify that a tool ID actually exists in the Galaxy toolbox.""" + if not self.deps.toolbox: + log.warning("Toolbox not available for tool verification") + return False + + try: + tool = self.deps.toolbox.get_tool(tool_id) + return tool is not None and not tool.hidden + except (AttributeError, KeyError, TypeError) as e: + log.debug(f"Error verifying tool {tool_id}: {e}") + return False + + def _get_fallback_content(self) -> str: + """Get fallback content for recommendation failures.""" + return "Unable to generate tool recommendations at this time." + + def _get_simple_system_prompt(self) -> str: + """Simple system prompt for models without structured output.""" + return """ + You are a Galaxy tool recommendation expert. Analyze the user's request and recommend tools. + + Respond in this exact format: + TOOL: [primary tool name] + TOOL_ID: [tool identifier] + REASON: [why this tool is recommended] + ALTERNATIVES: [alternative tools, comma-separated] + CONFIDENCE: [high/medium/low] + + Example: + TOOL: BWA-MEM + TOOL_ID: bwa_mem + REASON: Best for mapping paired-end reads to reference genome + ALTERNATIVES: Bowtie2, HISAT2 + CONFIDENCE: high + """ + + def _parse_simple_response(self, response_text: str) -> dict[str, Any]: + """Parse simple text response into structured format.""" + # Normalize text for consistent parsing + normalized_text = normalize_llm_text(response_text) + + # Extract structured information from text + tool = re.search(r"TOOL:\s*([^\n]+)", normalized_text, re.IGNORECASE) + tool_id = re.search(r"TOOL_ID:\s*([^\n]+)", normalized_text, re.IGNORECASE) + reason = re.search(r"REASON:\s*([^\n]+)", normalized_text, re.IGNORECASE) + alternatives = re.search(r"ALTERNATIVES:\s*([^\n]+)", normalized_text, re.IGNORECASE) + confidence_match = re.search(r"CONFIDENCE:\s*(\w+)", normalized_text, re.IGNORECASE) + + # Parse confidence level + confidence_level = ConfidenceLevel.MEDIUM + if confidence_match: + conf_str = confidence_match.group(1).lower() + if conf_str == "high": + confidence_level = ConfidenceLevel.HIGH + elif conf_str == "low": + confidence_level = ConfidenceLevel.LOW + + # Build content + content_parts = [] + if tool and tool.group(1).strip(): + tool_name = tool.group(1).strip() + tool_id_value = tool_id.group(1).strip() if tool_id else tool_name.lower().replace(" ", "_") + content_parts.append(f"**Recommended Tool:** {tool_name}") + content_parts.append(f"Tool ID: `{tool_id_value}`") + + if reason and reason.group(1).strip(): + content_parts.append(f"**Why:** {reason.group(1).strip()}") + + if alternatives and alternatives.group(1).strip(): + alt_list = alternatives.group(1).strip() + content_parts.append(f"**Alternatives:** {alt_list}") + + if not content_parts: + content_parts = [normalized_text] # Fallback to full response + + # Create suggestions + suggestions = [] + if tool and tool.group(1).strip(): + tool_name = tool.group(1).strip() + tool_id_value = tool_id.group(1).strip() if tool_id else tool_name.lower().replace(" ", "_") + suggestions.append( + ActionSuggestion( + action_type=ActionType.TOOL_RUN, + description=f"Run {tool_name}", + parameters={"tool_id": tool_id_value, "tool_name": tool_name}, + confidence=confidence_level, + priority=1, + ) + ) + + return { + "content": "\n\n".join(content_parts), + "confidence": confidence_level, + "has_alternatives": bool(alternatives and alternatives.group(1).strip()), + "suggestions": suggestions, + }