From ddac7ea3ed95d2d19755d0f33db60e8c44b1cde1 Mon Sep 17 00:00:00 2001 From: Viktor Avelino <64113566+viktoravelino@users.noreply.github.com> Date: Wed, 28 Jan 2026 09:11:37 -0500 Subject: [PATCH] fix: make sure to keep table outputs when changing provider (#11426) * fix: make sure to keep table outputs when changing provider * [autofix.ci] apply automated fixes * fix: enable real-time refresh for Else output option in SmartRouterComponent * [autofix.ci] apply automated fixes --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Himavarsha <40851462+HimavarshaVS@users.noreply.github.com> --- src/lfx/src/lfx/_assets/component_index.json | 7 ++++--- src/lfx/src/lfx/_assets/stable_hash_history.json | 2 +- .../components/llm_operations/llm_conditional_router.py | 3 ++- 3 files changed, 7 insertions(+), 5 deletions(-) diff --git a/src/lfx/src/lfx/_assets/component_index.json b/src/lfx/src/lfx/_assets/component_index.json index 29ec5da292..729486e81d 100644 --- a/src/lfx/src/lfx/_assets/component_index.json +++ b/src/lfx/src/lfx/_assets/component_index.json @@ -85967,7 +85967,7 @@ "icon": "route", "legacy": false, "metadata": { - "code_hash": "9c6736e784f6", + "code_hash": "2b49dc46dae8", "dependencies": { "dependencies": [ { @@ -86021,7 +86021,7 @@ "show": true, "title_case": false, "type": "code", - "value": "from typing import Any\n\nfrom lfx.base.models.unified_models import (\n get_language_model_options,\n get_llm,\n update_model_options_in_build_config,\n)\nfrom lfx.custom import Component\nfrom lfx.io import (\n BoolInput,\n MessageInput,\n MessageTextInput,\n ModelInput,\n MultilineInput,\n Output,\n SecretStrInput,\n TableInput,\n)\nfrom lfx.schema.message import Message\nfrom lfx.schema.table import EditMode\n\n\nclass SmartRouterComponent(Component):\n display_name = \"Smart Router\"\n description = \"Routes an input message using LLM-based categorization.\"\n icon = \"route\"\n name = \"SmartRouter\"\n\n def __init__(self, **kwargs):\n super().__init__(**kwargs)\n self._matched_category = None\n\n inputs = [\n ModelInput(\n name=\"model\",\n display_name=\"Language Model\",\n info=\"Select your model provider\",\n real_time_refresh=True,\n required=True,\n ),\n SecretStrInput(\n name=\"api_key\",\n display_name=\"API Key\",\n info=\"Model Provider API key\",\n real_time_refresh=True,\n advanced=True,\n ),\n MessageTextInput(\n name=\"input_text\",\n display_name=\"Input\",\n info=\"The primary text input for the operation.\",\n required=True,\n ),\n TableInput(\n name=\"routes\",\n display_name=\"Routes\",\n info=(\n \"Define the categories for routing. Each row should have a route/category name \"\n \"and optionally a custom output value.\"\n ),\n table_schema=[\n {\n \"name\": \"route_category\",\n \"display_name\": \"Route Name\",\n \"type\": \"str\",\n \"description\": \"Name for the route (used for both output name and category matching)\",\n \"edit_mode\": EditMode.INLINE,\n },\n {\n \"name\": \"route_description\",\n \"display_name\": \"Route Description\",\n \"type\": \"str\",\n \"description\": \"Description of when this route should be used (helps LLM understand the category)\",\n \"default\": \"\",\n \"edit_mode\": EditMode.POPOVER,\n },\n {\n \"name\": \"output_value\",\n \"display_name\": \"Route Message (Optional)\",\n \"type\": \"str\",\n \"description\": (\n \"Optional message to send when this route is matched.\"\n \"Leave empty to pass through the original input text.\"\n ),\n \"default\": \"\",\n \"edit_mode\": EditMode.POPOVER,\n },\n ],\n value=[\n {\n \"route_category\": \"Positive\",\n \"route_description\": \"Positive feedback, satisfaction, or compliments\",\n \"output_value\": \"\",\n },\n {\n \"route_category\": \"Negative\",\n \"route_description\": \"Complaints, issues, or dissatisfaction\",\n \"output_value\": \"\",\n },\n ],\n real_time_refresh=True,\n required=True,\n ),\n MessageInput(\n name=\"message\",\n display_name=\"Override Output\",\n info=(\n \"Optional override message that will replace both the Input and Output Value \"\n \"for all routes when filled.\"\n ),\n required=False,\n advanced=True,\n ),\n BoolInput(\n name=\"enable_else_output\",\n display_name=\"Include Else Output\",\n info=\"Include an Else output for cases that don't match any route.\",\n value=False,\n advanced=True,\n ),\n MultilineInput(\n name=\"custom_prompt\",\n display_name=\"Additional Instructions\",\n info=(\n \"Additional instructions for LLM-based categorization. \"\n \"These will be added to the base prompt. \"\n \"Use {input_text} for the input text and {routes} for the available categories.\"\n ),\n advanced=True,\n ),\n ]\n\n outputs: list[Output] = []\n\n def update_build_config(self, build_config: dict, field_value: str, field_name: str | None = None):\n \"\"\"Dynamically update build config with user-filtered model options.\"\"\"\n return update_model_options_in_build_config(\n component=self,\n build_config=build_config,\n cache_key_prefix=\"language_model_options\",\n get_options_func=get_language_model_options,\n field_name=field_name,\n field_value=field_value,\n )\n\n def update_outputs(self, frontend_node: dict, field_name: str, field_value: Any) -> dict:\n \"\"\"Create a dynamic output for each category in the categories table.\"\"\"\n if field_name in {\"routes\", \"enable_else_output\"}:\n frontend_node[\"outputs\"] = []\n\n # Get the routes data - either from field_value (if routes field) or from component state\n routes_data = field_value if field_name == \"routes\" else getattr(self, \"routes\", [])\n\n # Add a dynamic output for each category - all using the same method\n for i, row in enumerate(routes_data):\n route_category = row.get(\"route_category\", f\"Category {i + 1}\")\n frontend_node[\"outputs\"].append(\n Output(\n display_name=route_category,\n name=f\"category_{i + 1}_result\",\n method=\"process_case\",\n group_outputs=True,\n )\n )\n # Add default output only if enabled\n if field_name == \"enable_else_output\":\n enable_else = field_value\n else:\n enable_else = getattr(self, \"enable_else_output\", False)\n\n if enable_else:\n frontend_node[\"outputs\"].append(\n Output(display_name=\"Else\", name=\"default_result\", method=\"default_response\", group_outputs=True)\n )\n return frontend_node\n\n def process_case(self) -> Message:\n \"\"\"Process all categories using LLM categorization and return message for matching category.\"\"\"\n # Clear any previous match state\n self._matched_category = None\n\n # Get categories and input text\n categories = getattr(self, \"routes\", [])\n input_text = getattr(self, \"input_text\", \"\")\n\n # Find the matching category using LLM-based categorization\n matched_category = None\n llm = get_llm(model=self.model, user_id=self.user_id, api_key=self.api_key)\n\n if llm and categories:\n # Create prompt for categorization\n category_info = []\n for i, category in enumerate(categories):\n cat_name = category.get(\"route_category\", f\"Category {i + 1}\")\n cat_desc = category.get(\"route_description\", \"\")\n if cat_desc and cat_desc.strip():\n category_info.append(f'\"{cat_name}\": {cat_desc}')\n else:\n category_info.append(f'\"{cat_name}\"')\n\n categories_text = \"\\n\".join([f\"- {info}\" for info in category_info if info])\n\n # Create base prompt\n base_prompt = (\n f\"You are a text classifier. Given the following text and categories, \"\n f\"determine which category best matches the text.\\n\\n\"\n f'Text to classify: \"{input_text}\"\\n\\n'\n f\"Available categories:\\n{categories_text}\\n\\n\"\n f\"Respond with ONLY the exact category name that best matches the text. \"\n f'If none match well, respond with \"NONE\".\\n\\n'\n f\"Category:\"\n )\n\n # Use custom prompt as additional instructions if provided\n custom_prompt = getattr(self, \"custom_prompt\", \"\")\n if custom_prompt and custom_prompt.strip():\n self.status = \"Using custom prompt as additional instructions\"\n # Format custom prompt with variables\n # For the routes variable, create a simpler format for custom prompt usage\n simple_routes = \", \".join(\n [f'\"{cat.get(\"route_category\", f\"Category {i + 1}\")}\"' for i, cat in enumerate(categories)]\n )\n formatted_custom = custom_prompt.format(input_text=input_text, routes=simple_routes)\n # Combine base prompt with custom instructions\n prompt = f\"{base_prompt}\\n\\nAdditional Instructions:\\n{formatted_custom}\"\n else:\n self.status = \"Using default prompt for LLM categorization\"\n prompt = base_prompt\n\n # Log the final prompt being sent to LLM\n self.status = f\"Prompt sent to LLM:\\n{prompt}\"\n\n try:\n # Use the LLM to categorize\n if hasattr(llm, \"invoke\"):\n response = llm.invoke(prompt)\n if hasattr(response, \"content\"):\n categorization = response.content.strip().strip('\"')\n else:\n categorization = str(response).strip().strip('\"')\n else:\n categorization = str(llm(prompt)).strip().strip('\"')\n\n # Log the categorization process\n self.status = f\"LLM response: '{categorization}'\"\n\n # Find matching category based on LLM response\n for i, category in enumerate(categories):\n route_category = category.get(\"route_category\", \"\")\n\n # Log each comparison attempt\n self.status = (\n f\"Comparing '{categorization}' with category {i + 1}: route_category='{route_category}'\"\n )\n\n # Case-insensitive match\n if categorization.lower() == route_category.lower():\n matched_category = i\n self.status = f\"MATCH FOUND! Category {i + 1} matched with '{categorization}'\"\n break\n\n if matched_category is None:\n self.status = (\n f\"No match found for '{categorization}'. Available categories: \"\n f\"{[category.get('route_category', '') for category in categories]}\"\n )\n\n except RuntimeError as e:\n self.status = f\"Error in LLM categorization: {e!s}\"\n else:\n self.status = \"No LLM provided for categorization\"\n\n if matched_category is not None:\n # Store the matched category for other outputs to check\n self._matched_category = matched_category\n\n # Stop all category outputs except the matched one\n for i in range(len(categories)):\n if i != matched_category:\n self.stop(f\"category_{i + 1}_result\")\n\n # Also stop the default output (if it exists)\n enable_else = getattr(self, \"enable_else_output\", False)\n if enable_else:\n self.stop(\"default_result\")\n\n route_category = categories[matched_category].get(\"route_category\", f\"Category {matched_category + 1}\")\n self.status = f\"Categorized as {route_category}\"\n\n # Check if there's an override output (takes precedence over everything)\n override_output = getattr(self, \"message\", None)\n if (\n override_output\n and hasattr(override_output, \"text\")\n and override_output.text\n and str(override_output.text).strip()\n ):\n return Message(text=str(override_output.text))\n if override_output and isinstance(override_output, str) and override_output.strip():\n return Message(text=str(override_output))\n\n # Check if there's a custom output value for this category\n custom_output = categories[matched_category].get(\"output_value\", \"\")\n # Treat None, empty string, or whitespace as blank\n if custom_output and str(custom_output).strip() and str(custom_output).strip().lower() != \"none\":\n # Use custom output value\n return Message(text=str(custom_output))\n # Use input as default output\n return Message(text=input_text)\n # No match found, stop all category outputs\n for i in range(len(categories)):\n self.stop(f\"category_{i + 1}_result\")\n\n # Check if else output is enabled\n enable_else = getattr(self, \"enable_else_output\", False)\n if enable_else:\n # The default_response will handle the else case\n self.stop(\"process_case\")\n return Message(text=\"\")\n # No else output, so no output at all\n self.status = \"No match found and Else output is disabled\"\n return Message(text=\"\")\n\n def default_response(self) -> Message:\n \"\"\"Handle the else case when no conditions match.\"\"\"\n # Check if else output is enabled\n enable_else = getattr(self, \"enable_else_output\", False)\n if not enable_else:\n self.status = \"Else output is disabled\"\n return Message(text=\"\")\n\n # Clear any previous match state if not already set\n if not hasattr(self, \"_matched_category\"):\n self._matched_category = None\n\n categories = getattr(self, \"routes\", [])\n input_text = getattr(self, \"input_text\", \"\")\n\n # Check if a match was already found in process_case\n if hasattr(self, \"_matched_category\") and self._matched_category is not None:\n self.status = (\n f\"Match already found in process_case (Category {self._matched_category + 1}), \"\n \"stopping default_response\"\n )\n self.stop(\"default_result\")\n return Message(text=\"\")\n\n # Check if any category matches using LLM categorization\n has_match = False\n llm = get_llm(model=self.model, user_id=self.user_id, api_key=self.api_key)\n\n if llm and categories:\n try:\n # Create prompt for categorization\n category_info = []\n for i, category in enumerate(categories):\n cat_name = category.get(\"route_category\", f\"Category {i + 1}\")\n cat_desc = category.get(\"route_description\", \"\")\n if cat_desc and cat_desc.strip():\n category_info.append(f'\"{cat_name}\": {cat_desc}')\n else:\n category_info.append(f'\"{cat_name}\"')\n\n categories_text = \"\\n\".join([f\"- {info}\" for info in category_info if info])\n\n # Create base prompt\n base_prompt = (\n \"You are a text classifier. Given the following text and categories, \"\n \"determine which category best matches the text.\\n\\n\"\n f'Text to classify: \"{input_text}\"\\n\\n'\n f\"Available categories:\\n{categories_text}\\n\\n\"\n \"Respond with ONLY the exact category name that best matches the text. \"\n 'If none match well, respond with \"NONE\".\\n\\n'\n \"Category:\"\n )\n\n # Use custom prompt as additional instructions if provided\n custom_prompt = getattr(self, \"custom_prompt\", \"\")\n if custom_prompt and custom_prompt.strip():\n self.status = \"Using custom prompt as additional instructions (default check)\"\n # Format custom prompt with variables\n # For the routes variable, create a simpler format for custom prompt usage\n simple_routes = \", \".join(\n [f'\"{cat.get(\"route_category\", f\"Category {i + 1}\")}\"' for i, cat in enumerate(categories)]\n )\n formatted_custom = custom_prompt.format(input_text=input_text, routes=simple_routes)\n # Combine base prompt with custom instructions\n prompt = f\"{base_prompt}\\n\\nAdditional Instructions:\\n{formatted_custom}\"\n else:\n self.status = \"Using default prompt for LLM categorization (default check)\"\n prompt = base_prompt\n\n # Log the final prompt being sent to LLM for default check\n self.status = f\"Default check - Prompt sent to LLM:\\n{prompt}\"\n\n # Use the LLM to categorize\n if hasattr(llm, \"invoke\"):\n response = llm.invoke(prompt)\n if hasattr(response, \"content\"):\n categorization = response.content.strip().strip('\"')\n else:\n categorization = str(response).strip().strip('\"')\n else:\n categorization = str(llm(prompt)).strip().strip('\"')\n\n # Log the categorization process for default check\n self.status = f\"Default check - LLM response: '{categorization}'\"\n\n # Check if LLM response matches any category\n for i, category in enumerate(categories):\n route_category = category.get(\"route_category\", \"\")\n\n # Log each comparison attempt\n self.status = (\n f\"Default check - Comparing '{categorization}' with category {i + 1}: \"\n f\"route_category='{route_category}'\"\n )\n\n if categorization.lower() == route_category.lower():\n has_match = True\n self.status = f\"Default check - MATCH FOUND! Category {i + 1} matched with '{categorization}'\"\n break\n\n if not has_match:\n self.status = (\n f\"Default check - No match found for '{categorization}'. \"\n f\"Available categories: \"\n f\"{[category.get('route_category', '') for category in categories]}\"\n )\n\n except RuntimeError:\n pass # If there's an error, treat as no match\n\n if has_match:\n # A case matches, stop this output\n self.stop(\"default_result\")\n return Message(text=\"\")\n\n # No case matches, check for override output first, then use input as default\n override_output = getattr(self, \"message\", None)\n if (\n override_output\n and hasattr(override_output, \"text\")\n and override_output.text\n and str(override_output.text).strip()\n ):\n self.status = \"Routed to Else (no match) - using override output\"\n return Message(text=str(override_output.text))\n if override_output and isinstance(override_output, str) and override_output.strip():\n self.status = \"Routed to Else (no match) - using override output\"\n return Message(text=str(override_output))\n self.status = \"Routed to Else (no match) - using input as default\"\n return Message(text=input_text)\n" + "value": "from typing import Any\n\nfrom lfx.base.models.unified_models import (\n get_language_model_options,\n get_llm,\n update_model_options_in_build_config,\n)\nfrom lfx.custom import Component\nfrom lfx.io import (\n BoolInput,\n MessageInput,\n MessageTextInput,\n ModelInput,\n MultilineInput,\n Output,\n SecretStrInput,\n TableInput,\n)\nfrom lfx.schema.message import Message\nfrom lfx.schema.table import EditMode\n\n\nclass SmartRouterComponent(Component):\n display_name = \"Smart Router\"\n description = \"Routes an input message using LLM-based categorization.\"\n icon = \"route\"\n name = \"SmartRouter\"\n\n def __init__(self, **kwargs):\n super().__init__(**kwargs)\n self._matched_category = None\n\n inputs = [\n ModelInput(\n name=\"model\",\n display_name=\"Language Model\",\n info=\"Select your model provider\",\n real_time_refresh=True,\n required=True,\n ),\n SecretStrInput(\n name=\"api_key\",\n display_name=\"API Key\",\n info=\"Model Provider API key\",\n real_time_refresh=True,\n advanced=True,\n ),\n MessageTextInput(\n name=\"input_text\",\n display_name=\"Input\",\n info=\"The primary text input for the operation.\",\n required=True,\n ),\n TableInput(\n name=\"routes\",\n display_name=\"Routes\",\n info=(\n \"Define the categories for routing. Each row should have a route/category name \"\n \"and optionally a custom output value.\"\n ),\n table_schema=[\n {\n \"name\": \"route_category\",\n \"display_name\": \"Route Name\",\n \"type\": \"str\",\n \"description\": \"Name for the route (used for both output name and category matching)\",\n \"edit_mode\": EditMode.INLINE,\n },\n {\n \"name\": \"route_description\",\n \"display_name\": \"Route Description\",\n \"type\": \"str\",\n \"description\": \"Description of when this route should be used (helps LLM understand the category)\",\n \"default\": \"\",\n \"edit_mode\": EditMode.POPOVER,\n },\n {\n \"name\": \"output_value\",\n \"display_name\": \"Route Message (Optional)\",\n \"type\": \"str\",\n \"description\": (\n \"Optional message to send when this route is matched.\"\n \"Leave empty to pass through the original input text.\"\n ),\n \"default\": \"\",\n \"edit_mode\": EditMode.POPOVER,\n },\n ],\n value=[\n {\n \"route_category\": \"Positive\",\n \"route_description\": \"Positive feedback, satisfaction, or compliments\",\n \"output_value\": \"\",\n },\n {\n \"route_category\": \"Negative\",\n \"route_description\": \"Complaints, issues, or dissatisfaction\",\n \"output_value\": \"\",\n },\n ],\n real_time_refresh=True,\n required=True,\n ),\n MessageInput(\n name=\"message\",\n display_name=\"Override Output\",\n info=(\n \"Optional override message that will replace both the Input and Output Value \"\n \"for all routes when filled.\"\n ),\n required=False,\n advanced=True,\n ),\n BoolInput(\n name=\"enable_else_output\",\n display_name=\"Include Else Output\",\n info=\"Include an Else output for cases that don't match any route.\",\n value=False,\n advanced=True,\n real_time_refresh=True,\n ),\n MultilineInput(\n name=\"custom_prompt\",\n display_name=\"Additional Instructions\",\n info=(\n \"Additional instructions for LLM-based categorization. \"\n \"These will be added to the base prompt. \"\n \"Use {input_text} for the input text and {routes} for the available categories.\"\n ),\n advanced=True,\n ),\n ]\n\n outputs: list[Output] = []\n\n def update_build_config(self, build_config: dict, field_value: str, field_name: str | None = None):\n \"\"\"Dynamically update build config with user-filtered model options.\"\"\"\n return update_model_options_in_build_config(\n component=self,\n build_config=build_config,\n cache_key_prefix=\"language_model_options\",\n get_options_func=get_language_model_options,\n field_name=field_name,\n field_value=field_value,\n )\n\n def update_outputs(self, frontend_node: dict, field_name: str, field_value: Any) -> dict:\n \"\"\"Create a dynamic output for each category in the categories table.\"\"\"\n if field_name in {\"routes\", \"enable_else_output\", \"model\"}:\n frontend_node[\"outputs\"] = []\n\n # Get the routes data - either from field_value (if routes field) or from component state\n routes_data = field_value if field_name == \"routes\" else getattr(self, \"routes\", [])\n\n # Add a dynamic output for each category - all using the same method\n for i, row in enumerate(routes_data):\n route_category = row.get(\"route_category\", f\"Category {i + 1}\")\n frontend_node[\"outputs\"].append(\n Output(\n display_name=route_category,\n name=f\"category_{i + 1}_result\",\n method=\"process_case\",\n group_outputs=True,\n )\n )\n # Add default output only if enabled\n if field_name == \"enable_else_output\":\n enable_else = field_value\n else:\n enable_else = getattr(self, \"enable_else_output\", False)\n\n if enable_else:\n frontend_node[\"outputs\"].append(\n Output(display_name=\"Else\", name=\"default_result\", method=\"default_response\", group_outputs=True)\n )\n return frontend_node\n\n def process_case(self) -> Message:\n \"\"\"Process all categories using LLM categorization and return message for matching category.\"\"\"\n # Clear any previous match state\n self._matched_category = None\n\n # Get categories and input text\n categories = getattr(self, \"routes\", [])\n input_text = getattr(self, \"input_text\", \"\")\n\n # Find the matching category using LLM-based categorization\n matched_category = None\n llm = get_llm(model=self.model, user_id=self.user_id, api_key=self.api_key)\n\n if llm and categories:\n # Create prompt for categorization\n category_info = []\n for i, category in enumerate(categories):\n cat_name = category.get(\"route_category\", f\"Category {i + 1}\")\n cat_desc = category.get(\"route_description\", \"\")\n if cat_desc and cat_desc.strip():\n category_info.append(f'\"{cat_name}\": {cat_desc}')\n else:\n category_info.append(f'\"{cat_name}\"')\n\n categories_text = \"\\n\".join([f\"- {info}\" for info in category_info if info])\n\n # Create base prompt\n base_prompt = (\n f\"You are a text classifier. Given the following text and categories, \"\n f\"determine which category best matches the text.\\n\\n\"\n f'Text to classify: \"{input_text}\"\\n\\n'\n f\"Available categories:\\n{categories_text}\\n\\n\"\n f\"Respond with ONLY the exact category name that best matches the text. \"\n f'If none match well, respond with \"NONE\".\\n\\n'\n f\"Category:\"\n )\n\n # Use custom prompt as additional instructions if provided\n custom_prompt = getattr(self, \"custom_prompt\", \"\")\n if custom_prompt and custom_prompt.strip():\n self.status = \"Using custom prompt as additional instructions\"\n # Format custom prompt with variables\n # For the routes variable, create a simpler format for custom prompt usage\n simple_routes = \", \".join(\n [f'\"{cat.get(\"route_category\", f\"Category {i + 1}\")}\"' for i, cat in enumerate(categories)]\n )\n formatted_custom = custom_prompt.format(input_text=input_text, routes=simple_routes)\n # Combine base prompt with custom instructions\n prompt = f\"{base_prompt}\\n\\nAdditional Instructions:\\n{formatted_custom}\"\n else:\n self.status = \"Using default prompt for LLM categorization\"\n prompt = base_prompt\n\n # Log the final prompt being sent to LLM\n self.status = f\"Prompt sent to LLM:\\n{prompt}\"\n\n try:\n # Use the LLM to categorize\n if hasattr(llm, \"invoke\"):\n response = llm.invoke(prompt)\n if hasattr(response, \"content\"):\n categorization = response.content.strip().strip('\"')\n else:\n categorization = str(response).strip().strip('\"')\n else:\n categorization = str(llm(prompt)).strip().strip('\"')\n\n # Log the categorization process\n self.status = f\"LLM response: '{categorization}'\"\n\n # Find matching category based on LLM response\n for i, category in enumerate(categories):\n route_category = category.get(\"route_category\", \"\")\n\n # Log each comparison attempt\n self.status = (\n f\"Comparing '{categorization}' with category {i + 1}: route_category='{route_category}'\"\n )\n\n # Case-insensitive match\n if categorization.lower() == route_category.lower():\n matched_category = i\n self.status = f\"MATCH FOUND! Category {i + 1} matched with '{categorization}'\"\n break\n\n if matched_category is None:\n self.status = (\n f\"No match found for '{categorization}'. Available categories: \"\n f\"{[category.get('route_category', '') for category in categories]}\"\n )\n\n except RuntimeError as e:\n self.status = f\"Error in LLM categorization: {e!s}\"\n else:\n self.status = \"No LLM provided for categorization\"\n\n if matched_category is not None:\n # Store the matched category for other outputs to check\n self._matched_category = matched_category\n\n # Stop all category outputs except the matched one\n for i in range(len(categories)):\n if i != matched_category:\n self.stop(f\"category_{i + 1}_result\")\n\n # Also stop the default output (if it exists)\n enable_else = getattr(self, \"enable_else_output\", False)\n if enable_else:\n self.stop(\"default_result\")\n\n route_category = categories[matched_category].get(\"route_category\", f\"Category {matched_category + 1}\")\n self.status = f\"Categorized as {route_category}\"\n\n # Check if there's an override output (takes precedence over everything)\n override_output = getattr(self, \"message\", None)\n if (\n override_output\n and hasattr(override_output, \"text\")\n and override_output.text\n and str(override_output.text).strip()\n ):\n return Message(text=str(override_output.text))\n if override_output and isinstance(override_output, str) and override_output.strip():\n return Message(text=str(override_output))\n\n # Check if there's a custom output value for this category\n custom_output = categories[matched_category].get(\"output_value\", \"\")\n # Treat None, empty string, or whitespace as blank\n if custom_output and str(custom_output).strip() and str(custom_output).strip().lower() != \"none\":\n # Use custom output value\n return Message(text=str(custom_output))\n # Use input as default output\n return Message(text=input_text)\n # No match found, stop all category outputs\n for i in range(len(categories)):\n self.stop(f\"category_{i + 1}_result\")\n\n # Check if else output is enabled\n enable_else = getattr(self, \"enable_else_output\", False)\n if enable_else:\n # The default_response will handle the else case\n self.stop(\"process_case\")\n return Message(text=\"\")\n # No else output, so no output at all\n self.status = \"No match found and Else output is disabled\"\n return Message(text=\"\")\n\n def default_response(self) -> Message:\n \"\"\"Handle the else case when no conditions match.\"\"\"\n # Check if else output is enabled\n enable_else = getattr(self, \"enable_else_output\", False)\n if not enable_else:\n self.status = \"Else output is disabled\"\n return Message(text=\"\")\n\n # Clear any previous match state if not already set\n if not hasattr(self, \"_matched_category\"):\n self._matched_category = None\n\n categories = getattr(self, \"routes\", [])\n input_text = getattr(self, \"input_text\", \"\")\n\n # Check if a match was already found in process_case\n if hasattr(self, \"_matched_category\") and self._matched_category is not None:\n self.status = (\n f\"Match already found in process_case (Category {self._matched_category + 1}), \"\n \"stopping default_response\"\n )\n self.stop(\"default_result\")\n return Message(text=\"\")\n\n # Check if any category matches using LLM categorization\n has_match = False\n llm = get_llm(model=self.model, user_id=self.user_id, api_key=self.api_key)\n\n if llm and categories:\n try:\n # Create prompt for categorization\n category_info = []\n for i, category in enumerate(categories):\n cat_name = category.get(\"route_category\", f\"Category {i + 1}\")\n cat_desc = category.get(\"route_description\", \"\")\n if cat_desc and cat_desc.strip():\n category_info.append(f'\"{cat_name}\": {cat_desc}')\n else:\n category_info.append(f'\"{cat_name}\"')\n\n categories_text = \"\\n\".join([f\"- {info}\" for info in category_info if info])\n\n # Create base prompt\n base_prompt = (\n \"You are a text classifier. Given the following text and categories, \"\n \"determine which category best matches the text.\\n\\n\"\n f'Text to classify: \"{input_text}\"\\n\\n'\n f\"Available categories:\\n{categories_text}\\n\\n\"\n \"Respond with ONLY the exact category name that best matches the text. \"\n 'If none match well, respond with \"NONE\".\\n\\n'\n \"Category:\"\n )\n\n # Use custom prompt as additional instructions if provided\n custom_prompt = getattr(self, \"custom_prompt\", \"\")\n if custom_prompt and custom_prompt.strip():\n self.status = \"Using custom prompt as additional instructions (default check)\"\n # Format custom prompt with variables\n # For the routes variable, create a simpler format for custom prompt usage\n simple_routes = \", \".join(\n [f'\"{cat.get(\"route_category\", f\"Category {i + 1}\")}\"' for i, cat in enumerate(categories)]\n )\n formatted_custom = custom_prompt.format(input_text=input_text, routes=simple_routes)\n # Combine base prompt with custom instructions\n prompt = f\"{base_prompt}\\n\\nAdditional Instructions:\\n{formatted_custom}\"\n else:\n self.status = \"Using default prompt for LLM categorization (default check)\"\n prompt = base_prompt\n\n # Log the final prompt being sent to LLM for default check\n self.status = f\"Default check - Prompt sent to LLM:\\n{prompt}\"\n\n # Use the LLM to categorize\n if hasattr(llm, \"invoke\"):\n response = llm.invoke(prompt)\n if hasattr(response, \"content\"):\n categorization = response.content.strip().strip('\"')\n else:\n categorization = str(response).strip().strip('\"')\n else:\n categorization = str(llm(prompt)).strip().strip('\"')\n\n # Log the categorization process for default check\n self.status = f\"Default check - LLM response: '{categorization}'\"\n\n # Check if LLM response matches any category\n for i, category in enumerate(categories):\n route_category = category.get(\"route_category\", \"\")\n\n # Log each comparison attempt\n self.status = (\n f\"Default check - Comparing '{categorization}' with category {i + 1}: \"\n f\"route_category='{route_category}'\"\n )\n\n if categorization.lower() == route_category.lower():\n has_match = True\n self.status = f\"Default check - MATCH FOUND! Category {i + 1} matched with '{categorization}'\"\n break\n\n if not has_match:\n self.status = (\n f\"Default check - No match found for '{categorization}'. \"\n f\"Available categories: \"\n f\"{[category.get('route_category', '') for category in categories]}\"\n )\n\n except RuntimeError:\n pass # If there's an error, treat as no match\n\n if has_match:\n # A case matches, stop this output\n self.stop(\"default_result\")\n return Message(text=\"\")\n\n # No case matches, check for override output first, then use input as default\n override_output = getattr(self, \"message\", None)\n if (\n override_output\n and hasattr(override_output, \"text\")\n and override_output.text\n and str(override_output.text).strip()\n ):\n self.status = \"Routed to Else (no match) - using override output\"\n return Message(text=str(override_output.text))\n if override_output and isinstance(override_output, str) and override_output.strip():\n self.status = \"Routed to Else (no match) - using override output\"\n return Message(text=str(override_output))\n self.status = \"Routed to Else (no match) - using input as default\"\n return Message(text=input_text)\n" }, "custom_prompt": { "_input_type": "MultilineInput", @@ -86063,6 +86063,7 @@ "name": "enable_else_output", "override_skip": false, "placeholder": "", + "real_time_refresh": true, "required": false, "show": true, "title_case": false, @@ -115836,6 +115837,6 @@ "num_components": 355, "num_modules": 95 }, - "sha256": "1f88df2adf882e10e315292273e4f38d0b6268b18c88b59cfb83605952fccf98", + "sha256": "76bbf5825736277459f162978335dbd84b47f74e4eb2be8362a068b00060b651", "version": "0.3.0" } \ No newline at end of file diff --git a/src/lfx/src/lfx/_assets/stable_hash_history.json b/src/lfx/src/lfx/_assets/stable_hash_history.json index 4342c3ec74..e9442ea318 100644 --- a/src/lfx/src/lfx/_assets/stable_hash_history.json +++ b/src/lfx/src/lfx/_assets/stable_hash_history.json @@ -1176,7 +1176,7 @@ }, "SmartRouter": { "versions": { - "0.3.0": "9c6736e784f6" + "0.3.0": "2b49dc46dae8" } }, "LLMSelectorComponent": { diff --git a/src/lfx/src/lfx/components/llm_operations/llm_conditional_router.py b/src/lfx/src/lfx/components/llm_operations/llm_conditional_router.py index 738c3139ee..64d605eccb 100644 --- a/src/lfx/src/lfx/components/llm_operations/llm_conditional_router.py +++ b/src/lfx/src/lfx/components/llm_operations/llm_conditional_router.py @@ -117,6 +117,7 @@ class SmartRouterComponent(Component): info="Include an Else output for cases that don't match any route.", value=False, advanced=True, + real_time_refresh=True, ), MultilineInput( name="custom_prompt", @@ -145,7 +146,7 @@ class SmartRouterComponent(Component): def update_outputs(self, frontend_node: dict, field_name: str, field_value: Any) -> dict: """Create a dynamic output for each category in the categories table.""" - if field_name in {"routes", "enable_else_output"}: + if field_name in {"routes", "enable_else_output", "model"}: frontend_node["outputs"] = [] # Get the routes data - either from field_value (if routes field) or from component state