{
  "data": {
    "edges": [
      {
        "animated": false,
        "className": "",
        "data": {
          "sourceHandle": {
            "dataType": "ChatInput",
            "id": "ChatInput-docqa",
            "name": "message",
            "output_types": [
              "Message"
            ]
          },
          "targetHandle": {
            "fieldName": "search_query",
            "id": "KnowledgeRetrieve-docqa",
            "inputTypes": [
              "Message"
            ],
            "type": "str"
          }
        },
        "id": "reactflow__edge-ChatInput-docqa{\u0153dataType\u0153:\u0153ChatInput\u0153,\u0153id\u0153:\u0153ChatInput-docqa\u0153,\u0153name\u0153:\u0153message\u0153,\u0153output_types\u0153:[\u0153Message\u0153]}-KnowledgeRetrieve-docqa{\u0153fieldName\u0153:\u0153search_query\u0153,\u0153id\u0153:\u0153KnowledgeRetrieve-docqa\u0153,\u0153inputTypes\u0153:[\u0153Message\u0153],\u0153type\u0153:\u0153str\u0153}",
        "selected": false,
        "source": "ChatInput-docqa",
        "sourceHandle": "{\u0153dataType\u0153: \u0153ChatInput\u0153, \u0153id\u0153: \u0153ChatInput-docqa\u0153, \u0153name\u0153: \u0153message\u0153, \u0153output_types\u0153: [\u0153Message\u0153]}",
        "target": "KnowledgeRetrieve-docqa",
        "targetHandle": "{\u0153fieldName\u0153: \u0153search_query\u0153, \u0153id\u0153: \u0153KnowledgeRetrieve-docqa\u0153, \u0153inputTypes\u0153: [\u0153Message\u0153], \u0153type\u0153: \u0153str\u0153}"
      },
      {
        "animated": false,
        "className": "",
        "data": {
          "sourceHandle": {
            "dataType": "Knowledge",
            "id": "KnowledgeRetrieve-docqa",
            "name": "retrieve_data",
            "output_types": [
              "Table"
            ]
          },
          "targetHandle": {
            "fieldName": "input_data",
            "id": "Parser-docqa",
            "inputTypes": [
              "DataFrame",
              "Table",
              "Data",
              "JSON"
            ],
            "type": "other"
          }
        },
        "id": "reactflow__edge-KnowledgeRetrieve-docqa{\u0153dataType\u0153:\u0153Knowledge\u0153,\u0153id\u0153:\u0153KnowledgeRetrieve-docqa\u0153,\u0153name\u0153:\u0153retrieve_data\u0153,\u0153output_types\u0153:[\u0153Table\u0153]}-Parser-docqa{\u0153fieldName\u0153:\u0153input_data\u0153,\u0153id\u0153:\u0153Parser-docqa\u0153,\u0153inputTypes\u0153:[\u0153DataFrame\u0153,\u0153Table\u0153,\u0153Data\u0153,\u0153JSON\u0153],\u0153type\u0153:\u0153other\u0153}",
        "selected": false,
        "source": "KnowledgeRetrieve-docqa",
        "sourceHandle": "{\u0153dataType\u0153: \u0153Knowledge\u0153, \u0153id\u0153: \u0153KnowledgeRetrieve-docqa\u0153, \u0153name\u0153: \u0153retrieve_data\u0153, \u0153output_types\u0153: [\u0153Table\u0153]}",
        "target": "Parser-docqa",
        "targetHandle": "{\u0153fieldName\u0153: \u0153input_data\u0153, \u0153id\u0153: \u0153Parser-docqa\u0153, \u0153inputTypes\u0153: [\u0153DataFrame\u0153, \u0153Table\u0153, \u0153Data\u0153, \u0153JSON\u0153], \u0153type\u0153: \u0153other\u0153}"
      },
      {
        "animated": false,
        "className": "",
        "data": {
          "sourceHandle": {
            "dataType": "parser",
            "id": "Parser-docqa",
            "name": "parsed_text",
            "output_types": [
              "Message"
            ]
          },
          "targetHandle": {
            "fieldName": "context",
            "id": "Prompt-docqa",
            "inputTypes": [
              "Message",
              "Text"
            ],
            "type": "str"
          }
        },
        "id": "reactflow__edge-Parser-docqa{\u0153dataType\u0153:\u0153parser\u0153,\u0153id\u0153:\u0153Parser-docqa\u0153,\u0153name\u0153:\u0153parsed_text\u0153,\u0153output_types\u0153:[\u0153Message\u0153]}-Prompt-docqa{\u0153fieldName\u0153:\u0153context\u0153,\u0153id\u0153:\u0153Prompt-docqa\u0153,\u0153inputTypes\u0153:[\u0153Message\u0153,\u0153Text\u0153],\u0153type\u0153:\u0153str\u0153}",
        "selected": false,
        "source": "Parser-docqa",
        "sourceHandle": "{\u0153dataType\u0153: \u0153parser\u0153, \u0153id\u0153: \u0153Parser-docqa\u0153, \u0153name\u0153: \u0153parsed_text\u0153, \u0153output_types\u0153: [\u0153Message\u0153]}",
        "target": "Prompt-docqa",
        "targetHandle": "{\u0153fieldName\u0153: \u0153context\u0153, \u0153id\u0153: \u0153Prompt-docqa\u0153, \u0153inputTypes\u0153: [\u0153Message\u0153, \u0153Text\u0153], \u0153type\u0153: \u0153str\u0153}"
      },
      {
        "animated": false,
        "className": "",
        "data": {
          "sourceHandle": {
            "dataType": "ChatInput",
            "id": "ChatInput-docqa",
            "name": "message",
            "output_types": [
              "Message"
            ]
          },
          "targetHandle": {
            "fieldName": "question",
            "id": "Prompt-docqa",
            "inputTypes": [
              "Message",
              "Text"
            ],
            "type": "str"
          }
        },
        "id": "reactflow__edge-ChatInput-docqa{\u0153dataType\u0153:\u0153ChatInput\u0153,\u0153id\u0153:\u0153ChatInput-docqa\u0153,\u0153name\u0153:\u0153message\u0153,\u0153output_types\u0153:[\u0153Message\u0153]}-Prompt-docqa{\u0153fieldName\u0153:\u0153question\u0153,\u0153id\u0153:\u0153Prompt-docqa\u0153,\u0153inputTypes\u0153:[\u0153Message\u0153,\u0153Text\u0153],\u0153type\u0153:\u0153str\u0153}",
        "selected": false,
        "source": "ChatInput-docqa",
        "sourceHandle": "{\u0153dataType\u0153: \u0153ChatInput\u0153, \u0153id\u0153: \u0153ChatInput-docqa\u0153, \u0153name\u0153: \u0153message\u0153, \u0153output_types\u0153: [\u0153Message\u0153]}",
        "target": "Prompt-docqa",
        "targetHandle": "{\u0153fieldName\u0153: \u0153question\u0153, \u0153id\u0153: \u0153Prompt-docqa\u0153, \u0153inputTypes\u0153: [\u0153Message\u0153, \u0153Text\u0153], \u0153type\u0153: \u0153str\u0153}"
      },
      {
        "animated": false,
        "className": "",
        "data": {
          "sourceHandle": {
            "dataType": "Prompt",
            "id": "Prompt-docqa",
            "name": "prompt",
            "output_types": [
              "Message"
            ]
          },
          "targetHandle": {
            "fieldName": "input_value",
            "id": "Agent-docqa",
            "inputTypes": [
              "Message"
            ],
            "type": "str"
          }
        },
        "id": "reactflow__edge-Prompt-docqa{\u0153dataType\u0153:\u0153Prompt\u0153,\u0153id\u0153:\u0153Prompt-docqa\u0153,\u0153name\u0153:\u0153prompt\u0153,\u0153output_types\u0153:[\u0153Message\u0153]}-Agent-docqa{\u0153fieldName\u0153:\u0153input_value\u0153,\u0153id\u0153:\u0153Agent-docqa\u0153,\u0153inputTypes\u0153:[\u0153Message\u0153],\u0153type\u0153:\u0153str\u0153}",
        "selected": false,
        "source": "Prompt-docqa",
        "sourceHandle": "{\u0153dataType\u0153: \u0153Prompt\u0153, \u0153id\u0153: \u0153Prompt-docqa\u0153, \u0153name\u0153: \u0153prompt\u0153, \u0153output_types\u0153: [\u0153Message\u0153]}",
        "target": "Agent-docqa",
        "targetHandle": "{\u0153fieldName\u0153: \u0153input_value\u0153, \u0153id\u0153: \u0153Agent-docqa\u0153, \u0153inputTypes\u0153: [\u0153Message\u0153], \u0153type\u0153: \u0153str\u0153}"
      },
      {
        "animated": false,
        "className": "",
        "data": {
          "sourceHandle": {
            "dataType": "Agent",
            "id": "Agent-docqa",
            "name": "response",
            "output_types": [
              "Message"
            ]
          },
          "targetHandle": {
            "fieldName": "input_value",
            "id": "ChatOutput-9SFlE",
            "inputTypes": [
              "Data",
              "JSON",
              "DataFrame",
              "Table",
              "Message"
            ],
            "type": "other"
          }
        },
        "id": "xy-edge__Agent-docqa{\u0153dataType\u0153:\u0153Agent\u0153,\u0153id\u0153:\u0153Agent-docqa\u0153,\u0153name\u0153:\u0153response\u0153,\u0153output_types\u0153:[\u0153Message\u0153]}-ChatOutput-9SFlE{\u0153fieldName\u0153:\u0153input_value\u0153,\u0153id\u0153:\u0153ChatOutput-9SFlE\u0153,\u0153inputTypes\u0153:[\u0153Data\u0153,\u0153JSON\u0153,\u0153DataFrame\u0153,\u0153Table\u0153,\u0153Message\u0153],\u0153type\u0153:\u0153other\u0153}",
        "selected": false,
        "source": "Agent-docqa",
        "sourceHandle": "{\u0153dataType\u0153: \u0153Agent\u0153, \u0153id\u0153: \u0153Agent-docqa\u0153, \u0153name\u0153: \u0153response\u0153, \u0153output_types\u0153: [\u0153Message\u0153]}",
        "target": "ChatOutput-9SFlE",
        "targetHandle": "{\u0153fieldName\u0153: \u0153input_value\u0153, \u0153id\u0153: \u0153ChatOutput-9SFlE\u0153, \u0153inputTypes\u0153: [\u0153Data\u0153, \u0153JSON\u0153, \u0153DataFrame\u0153, \u0153Table\u0153, \u0153Message\u0153], \u0153type\u0153: \u0153other\u0153}"
      }
    ],
    "nodes": [
      {
        "data": {
          "description": "Get chat inputs from the Playground.",
          "display_name": "Chat Input",
          "id": "ChatInput-docqa",
          "node": {
            "base_classes": [
              "Message"
            ],
            "beta": false,
            "conditional_paths": [],
            "custom_fields": {},
            "description": "Get chat inputs from the Playground.",
            "display_name": "Chat Input",
            "documentation": "",
            "edited": false,
            "field_order": [
              "input_value",
              "should_store_message",
              "sender",
              "sender_name",
              "session_id",
              "context_id",
              "files"
            ],
            "frozen": false,
            "icon": "MessagesSquare",
            "legacy": false,
            "lf_version": "1.10.0",
            "metadata": {
              "code_hash": "45974e6b1909",
              "component_class_name": "ChatInput",
              "dependencies": {
                "dependencies": [
                  {
                    "name": "lfx",
                    "version": null
                  }
                ],
                "total_dependencies": 1
              },
              "module": "lfx.components.input_output.chat.ChatInput"
            },
            "output_types": [],
            "outputs": [
              {
                "allows_loop": false,
                "cache": true,
                "display_name": "Chat Message",
                "group_outputs": false,
                "method": "message_response",
                "name": "message",
                "selected": "Message",
                "tool_mode": true,
                "types": [
                  "Message"
                ],
                "value": "__UNDEFINED__"
              }
            ],
            "pinned": false,
            "template": {
              "_type": "Component",
              "code": {
                "advanced": true,
                "api_editable": false,
                "dynamic": true,
                "fileTypes": [],
                "file_path": "",
                "info": "",
                "list": false,
                "load_from_db": false,
                "multiline": true,
                "name": "code",
                "password": false,
                "placeholder": "",
                "required": true,
                "show": true,
                "title_case": false,
                "type": "code",
                "value": "from lfx.base.data.utils import IMG_FILE_TYPES, TEXT_FILE_TYPES\nfrom lfx.base.io.chat import ChatComponent\nfrom lfx.inputs.inputs import BoolInput\nfrom lfx.io import (\n    DropdownInput,\n    FileInput,\n    MessageTextInput,\n    MultilineInput,\n    Output,\n)\nfrom lfx.schema.image import get_file_paths\nfrom lfx.schema.message import Message\nfrom lfx.services.deps import get_settings_service\nfrom lfx.utils.constants import (\n    MESSAGE_SENDER_AI,\n    MESSAGE_SENDER_NAME_USER,\n    MESSAGE_SENDER_USER,\n)\nfrom lfx.utils.file_path_security import (\n    component_file_access_scopes,\n    enforce_local_file_access,\n    is_local_file_access_restricted,\n)\n\n\nclass ChatInput(ChatComponent):\n    display_name = \"Chat Input\"\n    description = \"Get chat inputs from the Playground.\"\n    documentation: str = \"https://docs.langflow.org/chat-input-and-output\"\n    icon = \"MessagesSquare\"\n    name = \"ChatInput\"\n    minimized = True\n\n    inputs = [\n        MultilineInput(\n            name=\"input_value\",\n            display_name=\"Input Text\",\n            value=\"\",\n            info=\"Message to be passed as input.\",\n            input_types=[],\n        ),\n        BoolInput(\n            name=\"should_store_message\",\n            display_name=\"Store Messages\",\n            info=\"Store the message in the history.\",\n            value=True,\n            advanced=True,\n        ),\n        DropdownInput(\n            name=\"sender\",\n            display_name=\"Sender Type\",\n            options=[MESSAGE_SENDER_AI, MESSAGE_SENDER_USER],\n            value=MESSAGE_SENDER_USER,\n            info=\"Type of sender.\",\n            advanced=True,\n        ),\n        MessageTextInput(\n            name=\"sender_name\",\n            display_name=\"Sender Name\",\n            info=\"Name of the sender.\",\n            value=MESSAGE_SENDER_NAME_USER,\n            advanced=True,\n        ),\n        MessageTextInput(\n            name=\"session_id\",\n            display_name=\"Session ID\",\n            info=\"The session ID of the chat. If empty, the current session ID parameter will be used.\",\n            advanced=True,\n        ),\n        MessageTextInput(\n            name=\"context_id\",\n            display_name=\"Context ID\",\n            info=\"The context ID of the chat. Adds an extra layer to the local memory.\",\n            value=\"\",\n            advanced=True,\n        ),\n        FileInput(\n            name=\"files\",\n            display_name=\"Files\",\n            file_types=TEXT_FILE_TYPES + IMG_FILE_TYPES,\n            info=\"Files to be sent with the message.\",\n            advanced=True,\n            is_list=True,\n            temp_file=True,\n        ),\n    ]\n    outputs = [\n        Output(display_name=\"Chat Message\", name=\"message\", method=\"message_response\"),\n    ]\n\n    async def message_response(self) -> Message:\n        # Ensure files is a list and filter out empty/None values\n        files = self.files if self.files else []\n        if files and not isinstance(files, list):\n            files = [files]\n        # Filter out None/empty values\n        files = [f for f in files if f is not None and f != \"\"]\n\n        # Build endpoints inject caller-controlled file references directly into\n        # ChatInput. Resolve and validate local references before the Message can\n        # later read them while constructing model attachment content.\n        if files:\n            settings_service = get_settings_service()\n            storage_type = getattr(getattr(settings_service, \"settings\", None), \"storage_type\", \"local\")\n            # The storage factory falls back to local storage for unsupported\n            # values, so only the explicitly configured S3 backend may bypass\n            # filesystem containment.\n            if str(storage_type).lower() != \"s3\" and is_local_file_access_restricted():\n                scope_ids = list(component_file_access_scopes(self))\n                # Public executions replace graph.flow_id with a per-visitor\n                # virtual ID. Only the server-populated source provenance may\n                # restore the already-validated public attachment namespace.\n                source_flow_id = getattr(self.graph, \"source_flow_id\", None)\n                if source_flow_id:\n                    scope_ids.append(source_flow_id)\n                for file_path in get_file_paths(files):\n                    enforce_local_file_access(file_path, scope_ids=scope_ids)\n\n        session_id = self.session_id or self.graph.session_id or \"\"\n        message = await Message.create(\n            text=self.input_value,\n            sender=self.sender,\n            sender_name=self.sender_name,\n            session_id=session_id,\n            context_id=self.context_id,\n            files=files,\n        )\n        if session_id and isinstance(message, Message) and self.should_store_message:\n            stored_message = await self.send_message(\n                message,\n            )\n            self.message.value = stored_message\n            message = stored_message\n\n        self.status = message\n        return message\n"
              },
              "context_id": {
                "_input_type": "MessageTextInput",
                "advanced": true,
                "display_name": "Context ID",
                "dynamic": false,
                "info": "The context ID of the chat. Adds an extra layer to the local memory.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "name": "context_id",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "type": "str",
                "value": ""
              },
              "files": {
                "advanced": true,
                "display_name": "Files",
                "dynamic": false,
                "fileTypes": [
                  "csv",
                  "json",
                  "pdf",
                  "txt",
                  "md",
                  "mdx",
                  "yaml",
                  "yml",
                  "xml",
                  "html",
                  "htm",
                  "docx",
                  "py",
                  "sh",
                  "sql",
                  "js",
                  "ts",
                  "tsx",
                  "jpg",
                  "jpeg",
                  "png",
                  "bmp",
                  "image"
                ],
                "file_path": "",
                "info": "Files to be sent with the message.",
                "list": true,
                "name": "files",
                "placeholder": "",
                "required": false,
                "show": true,
                "temp_file": true,
                "title_case": false,
                "trace_as_metadata": true,
                "type": "file",
                "value": ""
              },
              "input_value": {
                "advanced": false,
                "display_name": "Input Text",
                "dynamic": false,
                "info": "Message to be passed as input.",
                "input_types": [],
                "list": false,
                "load_from_db": false,
                "multiline": true,
                "name": "input_value",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "type": "str",
                "value": "What is this document is about?"
              },
              "sender": {
                "advanced": true,
                "display_name": "Sender Type",
                "dynamic": false,
                "info": "Type of sender.",
                "name": "sender",
                "options": [
                  "Machine",
                  "User"
                ],
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "trace_as_metadata": true,
                "type": "str",
                "value": "User"
              },
              "sender_name": {
                "advanced": true,
                "display_name": "Sender Name",
                "dynamic": false,
                "info": "Name of the sender.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "load_from_db": false,
                "name": "sender_name",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "type": "str",
                "value": "User"
              },
              "session_id": {
                "advanced": true,
                "display_name": "Session ID",
                "dynamic": false,
                "info": "The session ID of the chat. If empty, the current session ID parameter will be used.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "load_from_db": false,
                "name": "session_id",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "type": "str",
                "value": ""
              },
              "should_store_message": {
                "_input_type": "BoolInput",
                "advanced": true,
                "display_name": "Store Messages",
                "dynamic": false,
                "info": "Store the message in the history.",
                "list": false,
                "name": "should_store_message",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "trace_as_metadata": true,
                "type": "bool",
                "value": true
              }
            }
          },
          "selected_output": "message",
          "type": "ChatInput"
        },
        "dragging": false,
        "id": "ChatInput-docqa",
        "measured": {
          "height": 206,
          "width": 320
        },
        "position": {
          "x": 223.46760070052528,
          "y": 486.0770577933449
        },
        "positionAbsolute": {
          "x": 260,
          "y": 520
        },
        "selected": false,
        "type": "genericNode"
      },
      {
        "data": {
          "description": "Create a prompt template with dynamic variables.",
          "display_name": "Prompt Template",
          "id": "Prompt-docqa",
          "node": {
            "base_classes": [
              "Message"
            ],
            "beta": false,
            "conditional_paths": [],
            "custom_fields": {
              "template": [
                "context",
                "question"
              ]
            },
            "description": "Create a prompt template with dynamic variables.",
            "display_name": "Prompt Template",
            "documentation": "",
            "edited": false,
            "error": null,
            "field_order": [
              "template",
              "use_double_brackets",
              "tool_placeholder"
            ],
            "frozen": false,
            "full_path": null,
            "icon": "prompts",
            "is_composition": null,
            "is_input": null,
            "is_output": null,
            "legacy": false,
            "lf_version": "1.10.0",
            "metadata": {
              "code_hash": "e3714ffe5d15",
              "component_class_name": "PromptComponent",
              "dependencies": {
                "dependencies": [
                  {
                    "name": "lfx",
                    "version": null
                  }
                ],
                "total_dependencies": 1
              },
              "module": "lfx.components.models_and_agents.prompt.PromptComponent"
            },
            "name": "",
            "output_types": [],
            "outputs": [
              {
                "allows_loop": false,
                "cache": true,
                "display_name": "Prompt",
                "group_outputs": false,
                "method": "build_prompt",
                "name": "prompt",
                "selected": "Message",
                "tool_mode": true,
                "types": [
                  "Message"
                ],
                "value": "__UNDEFINED__"
              }
            ],
            "pinned": false,
            "template": {
              "_type": "Component",
              "code": {
                "advanced": true,
                "api_editable": false,
                "dynamic": true,
                "fileTypes": [],
                "file_path": "",
                "info": "",
                "list": false,
                "load_from_db": false,
                "multiline": true,
                "name": "code",
                "password": false,
                "placeholder": "",
                "required": true,
                "show": true,
                "title_case": false,
                "type": "code",
                "value": "from typing import Any\n\nfrom lfx.base.prompts.api_utils import process_prompt_template\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.inputs.input_mixin import FieldTypes\nfrom lfx.inputs.inputs import DefaultPromptField\nfrom lfx.io import BoolInput, MessageTextInput, Output, PromptInput\nfrom lfx.log.logger import logger\nfrom lfx.schema.dotdict import dotdict\nfrom lfx.schema.message import Message\nfrom lfx.template.utils import update_template_values\nfrom lfx.utils.mustache_security import validate_mustache_template\n\n\nclass PromptComponent(Component):\n    display_name: str = \"Prompt Template\"\n    description: str = \"Create a prompt template with dynamic variables.\"\n    documentation: str = \"https://docs.langflow.org/components-prompts\"\n    icon = \"prompts\"\n    trace_type = \"prompt\"\n    name = \"Prompt Template\"\n\n    inputs = [\n        PromptInput(name=\"template\", display_name=\"Template\"),\n        BoolInput(\n            name=\"use_double_brackets\",\n            display_name=\"Use Double Brackets\",\n            value=False,\n            advanced=True,\n            info=\"Use {{variable}} syntax instead of {variable}.\",\n            real_time_refresh=True,\n        ),\n        MessageTextInput(\n            name=\"tool_placeholder\",\n            display_name=\"Tool Placeholder\",\n            tool_mode=True,\n            advanced=True,\n            show=False,\n            info=\"A placeholder input for tool mode.\",\n        ),\n    ]\n\n    outputs = [\n        Output(display_name=\"Prompt\", name=\"prompt\", method=\"build_prompt\"),\n    ]\n\n    def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None) -> dotdict:\n        \"\"\"Update the template field type based on the selected mode.\"\"\"\n        if field_name == \"use_double_brackets\":\n            # Change the template field type based on mode\n            is_mustache = field_value is True\n            if is_mustache:\n                build_config[\"template\"][\"type\"] = FieldTypes.MUSTACHE_PROMPT.value\n            else:\n                build_config[\"template\"][\"type\"] = FieldTypes.PROMPT.value\n\n            # Re-process the template to update variables when mode changes\n            template_value = build_config.get(\"template\", {}).get(\"value\", \"\")\n            if template_value:\n                # Ensure custom_fields is properly initialized\n                if \"custom_fields\" not in build_config:\n                    build_config[\"custom_fields\"] = {}\n\n                # Clean up fields from the OLD mode before processing with NEW mode\n                # This ensures we don't keep fields with wrong syntax even if validation fails\n                old_custom_fields = build_config[\"custom_fields\"].get(\"template\", [])\n                for old_field in list(old_custom_fields):\n                    # Remove the field from custom_fields and template\n                    if old_field in old_custom_fields:\n                        old_custom_fields.remove(old_field)\n                    build_config.pop(old_field, None)\n\n                # Try to process template with new mode to add new variables\n                # If validation fails, at least we cleaned up old fields\n                try:\n                    # Validate mustache templates for security\n                    if is_mustache:\n                        validate_mustache_template(template_value)\n\n                    # Re-process template with new mode to add new variables\n                    _ = process_prompt_template(\n                        template=template_value,\n                        name=\"template\",\n                        custom_fields=build_config[\"custom_fields\"],\n                        frontend_node_template=build_config,\n                        is_mustache=is_mustache,\n                    )\n                except ValueError as e:\n                    # If validation fails, we still updated the mode and cleaned old fields\n                    # User will see error when they try to save\n                    logger.debug(\"Template validation failed during mode switch\", exc_info=e)\n        return build_config\n\n    async def build_prompt(self) -> Message:\n        use_double_brackets = self.use_double_brackets if hasattr(self, \"use_double_brackets\") else False\n        template_format = \"mustache\" if use_double_brackets else \"f-string\"\n        prompt = await Message.from_template_and_variables(template_format=template_format, **self._attributes)\n        self.status = prompt.text\n        return prompt\n\n    def _update_template(self, frontend_node: dict):\n        prompt_template = frontend_node[\"template\"][\"template\"][\"value\"]\n        use_double_brackets = frontend_node[\"template\"].get(\"use_double_brackets\", {}).get(\"value\", False)\n        is_mustache = use_double_brackets is True\n\n        try:\n            # Validate mustache templates for security\n            if is_mustache:\n                validate_mustache_template(prompt_template)\n\n            custom_fields = frontend_node[\"custom_fields\"]\n            frontend_node_template = frontend_node[\"template\"]\n            _ = process_prompt_template(\n                template=prompt_template,\n                name=\"template\",\n                custom_fields=custom_fields,\n                frontend_node_template=frontend_node_template,\n                is_mustache=is_mustache,\n            )\n        except ValueError as e:\n            # If validation fails, don't add variables but allow component to be created\n            logger.debug(\"Template validation failed in _update_template\", exc_info=e)\n        return frontend_node\n\n    async def update_frontend_node(self, new_frontend_node: dict, current_frontend_node: dict):\n        \"\"\"This function is called after the code validation is done.\"\"\"\n        frontend_node = await super().update_frontend_node(new_frontend_node, current_frontend_node)\n        template = frontend_node[\"template\"][\"template\"][\"value\"]\n        use_double_brackets = frontend_node[\"template\"].get(\"use_double_brackets\", {}).get(\"value\", False)\n        is_mustache = use_double_brackets is True\n\n        try:\n            # Validate mustache templates for security\n            if is_mustache:\n                validate_mustache_template(template)\n\n            # Kept it duplicated for backwards compatibility\n            _ = process_prompt_template(\n                template=template,\n                name=\"template\",\n                custom_fields=frontend_node[\"custom_fields\"],\n                frontend_node_template=frontend_node[\"template\"],\n                is_mustache=is_mustache,\n            )\n        except ValueError as e:\n            # If validation fails, don't add variables but allow component to be updated\n            logger.debug(\"Template validation failed in update_frontend_node\", exc_info=e)\n        # Now that template is updated, we need to grab any values that were set in the current_frontend_node\n        # and update the frontend_node with those values\n        update_template_values(new_template=frontend_node, previous_template=current_frontend_node[\"template\"])\n        return frontend_node\n\n    def _get_fallback_input(self, **kwargs):\n        return DefaultPromptField(**kwargs)\n"
              },
              "context": {
                "advanced": false,
                "display_name": "Context",
                "dynamic": false,
                "field_type": "str",
                "fileTypes": [],
                "file_path": "",
                "info": "",
                "input_types": [
                  "Message",
                  "Text"
                ],
                "list": false,
                "load_from_db": false,
                "multiline": true,
                "name": "context",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "type": "str",
                "value": ""
              },
              "question": {
                "advanced": false,
                "display_name": "Question",
                "dynamic": false,
                "field_type": "str",
                "fileTypes": [],
                "file_path": "",
                "info": "",
                "input_types": [
                  "Message",
                  "Text"
                ],
                "list": false,
                "load_from_db": false,
                "multiline": true,
                "name": "question",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "type": "str",
                "value": ""
              },
              "template": {
                "advanced": false,
                "display_name": "Template",
                "dynamic": false,
                "info": "",
                "list": false,
                "load_from_db": false,
                "name": "template",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "trace_as_input": true,
                "type": "prompt",
                "value": "You are answering questions using context retrieved from a knowledge base.\n\nUse ONLY the context below. If the answer is not in the context, say you don't know.\n\nContext:\n{context}\n\nQuestion:\n{question}\n\nAnswer:"
              },
              "tool_placeholder": {
                "_input_type": "MessageTextInput",
                "advanced": true,
                "display_name": "Tool Placeholder",
                "dynamic": false,
                "info": "A placeholder input for tool mode.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "load_from_db": false,
                "name": "tool_placeholder",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": true,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "type": "str",
                "value": ""
              },
              "use_double_brackets": {
                "_input_type": "BoolInput",
                "advanced": true,
                "display_name": "Use Double Brackets",
                "dynamic": false,
                "info": "Use {{variable}} syntax instead of {variable}.",
                "list": false,
                "list_add_label": "Add More",
                "name": "use_double_brackets",
                "override_skip": false,
                "placeholder": "",
                "real_time_refresh": true,
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "bool",
                "value": false
              }
            },
            "tool_mode": false
          },
          "selected_output": "prompt",
          "type": "Prompt"
        },
        "dragging": false,
        "id": "Prompt-docqa",
        "measured": {
          "height": 428,
          "width": 320
        },
        "position": {
          "x": 1560,
          "y": 566.1295971978984
        },
        "positionAbsolute": {
          "x": 1560,
          "y": 420
        },
        "selected": false,
        "type": "genericNode"
      },
      {
        "data": {
          "id": "KnowledgeRetrieve-docqa",
          "node": {
            "base_classes": [
              "DataFrame",
              "Table"
            ],
            "beta": false,
            "conditional_paths": [],
            "custom_fields": {},
            "description": "Ingest into or retrieve from a Langflow knowledge base.",
            "display_name": "Knowledge",
            "documentation": "",
            "edited": false,
            "field_order": [
              "knowledge_base",
              "mode",
              "input_df",
              "column_config",
              "chunk_size",
              "api_key",
              "allow_duplicates",
              "metadata_json",
              "search_query",
              "top_k",
              "include_metadata",
              "include_embeddings",
              "metadata_filter"
            ],
            "frozen": false,
            "icon": "database",
            "last_updated": "2026-06-25T19:34:01.266Z",
            "legacy": false,
            "lf_version": "1.10.0",
            "metadata": {
              "code_hash": "a0825fa78c40",
              "component_class_name": "KnowledgeComponent",
              "dependencies": {
                "dependencies": [
                  {
                    "name": "anyio",
                    "version": "4.14.2"
                  },
                  {
                    "name": "pandas",
                    "version": "2.3.3"
                  },
                  {
                    "name": "lfx",
                    "version": null
                  },
                  {
                    "name": "langflow",
                    "version": null
                  },
                  {
                    "name": "numpy",
                    "version": "2.4.6"
                  }
                ],
                "total_dependencies": 5
              },
              "module": "lfx.components.files_and_knowledge.knowledge.KnowledgeComponent"
            },
            "minimized": false,
            "output_types": [],
            "outputs": [
              {
                "allows_loop": false,
                "cache": true,
                "display_name": "Results",
                "group_outputs": false,
                "method": "build_kb_info",
                "name": "dataframe_output",
                "selected": "JSON",
                "tool_mode": true,
                "types": [
                  "JSON"
                ],
                "value": "__UNDEFINED__"
              },
              {
                "allows_loop": false,
                "cache": true,
                "display_name": "Results",
                "group_outputs": false,
                "method": "retrieve_data",
                "name": "retrieve_data",
                "selected": "Table",
                "tool_mode": true,
                "types": [
                  "Table"
                ],
                "value": "__UNDEFINED__"
              }
            ],
            "pinned": false,
            "template": {
              "_type": "Component",
              "allow_duplicates": {
                "_input_type": "BoolInput",
                "advanced": true,
                "display_name": "Allow Duplicates",
                "dynamic": true,
                "info": "Allow duplicate rows in the knowledge base",
                "list": false,
                "list_add_label": "Add More",
                "name": "allow_duplicates",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": false,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "bool",
                "value": false
              },
              "api_key": {
                "_input_type": "SecretStrInput",
                "advanced": true,
                "display_name": "Embedding Provider API Key",
                "dynamic": true,
                "info": "Overrides global provider settings. Leave blank to use your pre-configured API Key.",
                "input_types": [],
                "load_from_db": false,
                "name": "api_key",
                "override_skip": false,
                "password": true,
                "placeholder": "",
                "required": false,
                "show": false,
                "title_case": false,
                "track_in_telemetry": false,
                "type": "str",
                "value": ""
              },
              "chunk_size": {
                "_input_type": "IntInput",
                "advanced": true,
                "display_name": "Chunk Size",
                "dynamic": true,
                "info": "Batch size for processing embeddings",
                "list": false,
                "list_add_label": "Add More",
                "name": "chunk_size",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": false,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "int",
                "value": 1000
              },
              "code": {
                "advanced": true,
                "api_editable": false,
                "dynamic": true,
                "fileTypes": [],
                "file_path": "",
                "info": "",
                "list": false,
                "load_from_db": false,
                "multiline": true,
                "name": "code",
                "password": false,
                "placeholder": "",
                "required": true,
                "show": true,
                "title_case": false,
                "type": "code",
                "value": "\"\"\"Unified Knowledge component \u2014 ingest into or retrieve from a knowledge base.\n\nThis component merges what used to live in ``ingestion.py`` and\n``retrieval.py`` into a single, mode-driven component. A ``TabInput``\n(\"\ud83d\udce5 Ingest\" / \"\ud83d\udd0d Retrieve\") drives which inputs and which output are\nvisible. The merged shape gives users one node per KB instead of two\nparallel ones that always shared the same ``knowledge_base`` picker,\nembedding-model metadata, and backend registry.\n\nBoth legacy classes (``KnowledgeIngestionComponent``,\n``KnowledgeBaseComponent``) remain importable as thin subclasses of\n``KnowledgeComponent`` \u2014 see the sibling ``ingestion.py`` / ``retrieval.py``\nmodules \u2014 so saved flows continue to load unchanged.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport asyncio\nimport hashlib\nimport json\nimport uuid\nfrom collections.abc import Awaitable, Mapping\nfrom contextlib import aclosing\nfrom dataclasses import asdict, dataclass, field\nfrom datetime import datetime, timezone\nfrom typing import TYPE_CHECKING, Any, TypeVar\n\nimport anyio\nimport pandas as pd\n\nfrom lfx.base.knowledge_bases.backends import BackendType, BaseVectorStoreBackend\nfrom lfx.base.knowledge_bases.backends.base import BackendConfigurationError, IngestedDocument\nfrom lfx.base.knowledge_bases.backends.naming import ensure_storage_routing_allowed\nfrom lfx.base.knowledge_bases.ingestion_sources.base import (\n    IngestionItemResult,\n    IngestionItemStatus,\n    IngestionRunStatus,\n    IngestionSummary,\n)\nfrom lfx.base.knowledge_bases.ingestion_sources.flow_component import FlowComponentSource\nfrom lfx.base.knowledge_bases.knowledge_base_utils import get_knowledge_bases\nfrom lfx.base.knowledge_bases.validation import (\n    is_valid_collection_name as is_valid_kb_collection_name,\n)\nfrom lfx.base.models.unified_models import get_embedding_model_options, get_embeddings\nfrom lfx.components.processing.converter import convert_to_dataframe\nfrom lfx.custom import Component\nfrom lfx.helpers.base_model import coalesce_bool\nfrom lfx.io import (\n    BoolInput,\n    DBProviderInput,\n    DropdownInput,\n    HandleInput,\n    IntInput,\n    MessageTextInput,\n    ModelInput,\n    Output,\n    SecretStrInput,\n    StrInput,\n    TabInput,\n    TableInput,\n)\nfrom lfx.log.logger import logger\nfrom lfx.schema.data import Data\nfrom lfx.schema.dataframe import DataFrame\nfrom lfx.schema.dotdict import dotdict\nfrom lfx.schema.table import EditMode\nfrom lfx.services.deps import (\n    session_scope,\n)\nfrom lfx.utils.component_utils import set_current_fields, set_field_display\nfrom lfx.utils.validate_cloud import raise_error_if_astra_cloud_disable_component\n\n\ndef _inputs_for_mode(default_mode: str) -> list:\n    \"\"\"Return a fresh copy of the canonical inputs list with show flags set for the given mode.\n\n    Used by the legacy subclasses so a saved flow keyed on\n    ``KnowledgeIngestion`` / ``KnowledgeBase`` lands on a node template\n    whose default visibility already matches the pinned mode \u2014 no\n    \"wait for the user to click the mode tab\" UX gap on load.\n    \"\"\"\n    always_visible = set(KnowledgeComponent.default_keys) | set(KnowledgeComponent.mode_config[default_mode])\n    inputs_copy = []\n    for inp in KnowledgeComponent.inputs:\n        clone = inp.model_copy(deep=True) if hasattr(inp, \"model_copy\") else inp\n        if clone.name == \"mode\":\n            clone.value = default_mode\n        else:\n            clone.show = clone.name in always_visible\n        inputs_copy.append(clone)\n    return inputs_copy\n\n\nif TYPE_CHECKING:\n    from pathlib import Path\n\n# Mode constants. Plain-text labels (no emoji) for consistency with the rest of\n# Langflow's TabInput palette \u2014 see ``MemoryComponent`` for the same convention.\nMODE_INGEST = \"Ingest\"\nMODE_RETRIEVE = \"Retrieve\"\n\n# How often a flow ingestion checks whether its job was cancelled from the\n# Knowledge page while it waits on the embedding provider.\nJOB_CANCEL_POLL_SECONDS = 1.0\n\n\nclass KnowledgeIngestionCancelledError(Exception):\n    \"\"\"The ingestion's job was cancelled from the Knowledge page.\"\"\"\n\n\n_T = TypeVar(\"_T\")\n\n\nasync def _finish_despite_cancellation(awaitable: Awaitable[_T]) -> _T:\n    \"\"\"Run ``awaitable`` to completion even if the caller is cancelled meanwhile.\n\n    ``anyio.CancelScope(shield=True)`` holds off AnyIO cancellation, but a\n    repeated native ``Task.cancel()`` still interrupts it, and a stopped flow\n    build can cancel its task more than once. The work runs in its own task,\n    and the caller waits through every cancellation, re-raising once the work\n    has landed.\n    \"\"\"\n    task = asyncio.ensure_future(awaitable)\n    cancelled = False\n    with anyio.CancelScope(shield=True):\n        while not task.done():\n            try:\n                await asyncio.wait({task})\n            except asyncio.CancelledError:\n                cancelled = True\n    if cancelled:\n        error = None if task.cancelled() else task.exception()\n        if error is not None:\n            logger.warning(f\"Work finished during cancellation failed: {error!r}\")\n        raise asyncio.CancelledError from error\n    return task.result()\n\n\ndef _is_retrieve_mode(value: Any) -> bool:\n    \"\"\"Lenient mode check: treats any label containing 'Retrieve' as retrieve mode.\n\n    Older saved flows may carry the emoji-prefixed labels (\"\ud83d\udce5 Ingest\" /\n    \"\ud83d\udd0d Retrieve\") this component used to ship with; substring matching\n    keeps those loading without forcing a flow rewrite.\n    \"\"\"\n    return isinstance(value, str) and \"Retrieve\" in value\n\n\n# Boolean flags of a ``column_config`` row. Toggled cells hold real booleans,\n# but a typed cell (or a flow saved while the cell rendered as plain text)\n# keeps the raw string, e.g. ``\"true\"`` \u2014 read them with ``coalesce_bool``.\n_COLUMN_FLAGS = (\"vectorize\", \"identifier\")\n\n\ndef _normalize_column_config(config_list: list[dict[str, Any]]) -> list[dict[str, Any]]:\n    \"\"\"Return copies of the rows with every flag as a real boolean.\"\"\"\n    return [{**row, **{key: coalesce_bool(row.get(key)) for key in _COLUMN_FLAGS}} for row in config_list]\n\n\n# Error message used by both the ingest and retrieve paths when the user is\n# running against an Astra cloud environment that disables these flows.\nastra_error_msg = \"Knowledge ingestion and retrieval are not supported in Astra cloud environment.\"\n\n_DEFAULT_OPENSEARCH_CONFIG = {\n    \"url_variable\": \"OPENSEARCH_URL\",\n    \"username_variable\": \"OPENSEARCH_USERNAME\",\n    \"password_variable\": \"OPENSEARCH_PASSWORD\",  # pragma: allowlist secret\n    \"index_name\": \"\",\n    \"vector_field\": \"vector_field\",\n    \"text_field\": \"text\",\n}\n\n\nclass KnowledgeComponent(Component):\n    \"\"\"One component for both writing into and reading from a Langflow knowledge base.\n\n    A ``TabInput`` switches between ingestion and retrieval. The\n    ``update_build_config`` / ``update_outputs`` hooks hide the inputs\n    and the output that don't apply to the current mode so the canvas\n    node stays focused.\n    \"\"\"\n\n    display_name = \"Knowledge\"\n    description = \"Ingest into or retrieve from a Langflow knowledge base.\"\n    icon = \"database\"\n    name = \"Knowledge\"\n\n    # ------ Mode \u2192 visible-fields wiring ---------------------------------\n    # ``default_keys`` are inputs always visible regardless of mode.\n    # ``mode_config`` lists the inputs unique to each mode; everything outside\n    # this set is hidden when its mode is not selected.\n    default_keys: list[str] = [\"mode\", \"knowledge_base\"]\n    mode_config: dict[str, list[str]] = {\n        MODE_INGEST: [\n            \"input_df\",\n            \"column_config\",\n            \"chunk_size\",\n            \"api_key\",\n            \"allow_duplicates\",\n            \"metadata_json\",\n        ],\n        MODE_RETRIEVE: [\n            \"search_query\",\n            \"top_k\",\n            \"include_metadata\",\n            \"include_embeddings\",\n            \"metadata_filter\",\n        ],\n    }\n\n    def __init__(self, *args, **kwargs) -> None:\n        super().__init__(*args, **kwargs)\n        self._cached_kb_path: Path | None = None\n        self._knowledge_output_result: Data | DataFrame | None = None\n\n    def _pre_run_setup(self) -> None:\n        self._knowledge_output_result = None\n\n    async def _get_output_result(self, output: Output):\n        # Both legacy handles dispatch to the selected mode. Terminal nodes\n        # evaluate both handles, but must execute the storage operation once.\n        if output.method not in {\"build_kb_info\", \"retrieve_data\"}:\n            return await super()._get_output_result(output)\n        if self._knowledge_output_result is None:\n            raw_output = output.model_copy(update={\"options\": None})\n            self._knowledge_output_result = await super()._get_output_result(raw_output)\n        output.value = output.apply_options(self._knowledge_output_result)\n        return output.value\n\n    @staticmethod\n    def _embedding_provider_from_selection(selection: Any) -> str | None:\n        \"\"\"Read a provider name from a raw or persisted model selection.\"\"\"\n        if isinstance(selection, list):\n            selection = selection[0] if selection else None\n        if not isinstance(selection, Mapping):\n            return None\n        provider = selection.get(\"provider\")\n        if not isinstance(provider, str) or not provider.strip():\n            return None\n        return provider.strip()\n\n    async def _additional_model_provider_policy_ids(self, purpose, parameters=None) -> tuple[str, ...]:\n        \"\"\"Resolve the embedding provider before dialog hooks or runtime secrets.\n\n        New-KB configuration carries its ModelInput inside a dropdown dialog,\n        while existing KBs persist the provider in the KB row. Neither is a\n        top-level ModelInput, so both must be resolved explicitly before the\n        generic component pipeline hydrates ``api_key``.\n        \"\"\"\n        from lfx.base.models.provider_registry import resolve_provider_id\n        from lfx.services.model_provider_policy import ModelProviderPolicyError\n\n        effective_parameters = parameters if isinstance(parameters, Mapping) else getattr(self, \"_parameters\", None)\n        if not isinstance(effective_parameters, Mapping):\n            effective_parameters = {}\n        knowledge_value = effective_parameters.get(\"knowledge_base\", getattr(self, \"knowledge_base\", None))\n\n        if isinstance(knowledge_value, Mapping):\n            if \"02_embedding_model\" not in knowledge_value:\n                return ()\n            provider = self._embedding_provider_from_selection(knowledge_value.get(\"02_embedding_model\"))\n        elif isinstance(knowledge_value, str) and knowledge_value.strip():\n            metadata = await self._get_kb_metadata(knowledge_value.strip())\n            provider = self._embedding_provider_from_selection(metadata.get(\"model_selection\"))\n            if provider is None:\n                legacy_provider = metadata.get(\"embedding_provider\")\n                provider = legacy_provider.strip() if isinstance(legacy_provider, str) else None\n                if provider == \"Unknown\":\n                    provider = None\n        else:\n            return ()\n\n        if provider is None:\n            unknown_provider = \"unknown\"\n            raise ModelProviderPolicyError(unknown_provider, purpose)\n        return (resolve_provider_id(provider),)\n\n    @dataclass\n    class NewKnowledgeBaseInput:\n        functionality: str = \"create\"\n        fields: dict[str, dict] = field(\n            default_factory=lambda: {\n                \"data\": {\n                    \"node\": {\n                        \"name\": \"create_knowledge_base\",\n                        \"description\": \"Create new knowledge in Langflow.\",\n                        \"display_name\": \"Create new Knowledge Base\",\n                        \"field_order\": [\n                            \"01_new_kb_name\",\n                            \"02_embedding_model\",\n                            \"03_knowledge_backend\",\n                        ],\n                        \"template\": {\n                            \"01_new_kb_name\": StrInput(\n                                name=\"new_kb_name\",\n                                display_name=\"Knowledge Name\",\n                                info=\"Name of the new knowledge to create.\",\n                                required=True,\n                            ),\n                            \"02_embedding_model\": ModelInput(\n                                name=\"embedding_model\",\n                                display_name=\"Choose Embedding Model\",\n                                info=(\n                                    \"Select the embedding model to use for this knowledge base. \"\n                                    \"Langflow uses the configured credentials for that model provider.\"\n                                ),\n                                required=True,\n                                model_type=\"embedding\",\n                            ),\n                            \"03_knowledge_backend\": DBProviderInput(\n                                name=\"knowledge_backend\",\n                                display_name=\"DB Provider\",\n                                info=(\n                                    \"Select where this knowledge base stores vectors. \"\n                                    \"OpenSearch uses the global DB Providers settings.\"\n                                ),\n                                required=True,\n                            ),\n                        },\n                    },\n                }\n            }\n        )\n\n    # ------ Inputs --------------------------------------------------------\n    # ``knowledge_base`` is kept at position 0 so legacy tests that check\n    # ``component.inputs[0].dialog_inputs`` continue to work.\n    inputs = [\n        DropdownInput(\n            name=\"knowledge_base\",\n            display_name=\"Knowledge\",\n            info=\"Select the knowledge to load data from.\",\n            required=True,\n            options=[],\n            refresh_button=True,\n            real_time_refresh=True,\n            dialog_inputs=asdict(NewKnowledgeBaseInput()),\n        ),\n        TabInput(\n            name=\"mode\",\n            display_name=\"Mode\",\n            options=[MODE_INGEST, MODE_RETRIEVE],\n            value=MODE_INGEST,\n            info=\"Switch between writing new data into the knowledge base and querying it.\",\n            real_time_refresh=True,\n            tool_mode=True,\n        ),\n        # --- Ingest-only inputs (default-shown; hidden when mode == Retrieve) -\n        HandleInput(\n            name=\"input_df\",\n            display_name=\"Input\",\n            info=(\n                \"Table with all original columns (already chunked / processed). \"\n                \"Accepts Message, Data, or DataFrame. If Message or Data is provided, \"\n                \"it is converted to a DataFrame automatically.\"\n            ),\n            input_types=[\"Message\", \"Data\", \"JSON\", \"DataFrame\", \"Table\"],\n            required=True,\n            dynamic=True,\n            show=True,\n        ),\n        TableInput(\n            name=\"column_config\",\n            display_name=\"Column Configuration\",\n            info=\"Configure column behavior for the knowledge base.\",\n            required=True,\n            table_schema=[\n                {\n                    \"name\": \"column_name\",\n                    \"display_name\": \"Column Name\",\n                    \"type\": \"str\",\n                    \"description\": \"Name of the column in the source DataFrame\",\n                    \"edit_mode\": EditMode.INLINE,\n                },\n                {\n                    \"name\": \"vectorize\",\n                    \"display_name\": \"Vectorize\",\n                    \"type\": \"boolean\",\n                    \"description\": \"Create embeddings for this column\",\n                    \"default\": False,\n                    \"edit_mode\": EditMode.INLINE,\n                },\n                {\n                    \"name\": \"identifier\",\n                    \"display_name\": \"Identifier\",\n                    \"type\": \"boolean\",\n                    \"description\": \"Use this column as unique identifier\",\n                    \"default\": False,\n                    \"edit_mode\": EditMode.INLINE,\n                },\n            ],\n            value=[\n                {\n                    \"column_name\": \"text\",\n                    \"vectorize\": True,\n                    \"identifier\": True,\n                },\n            ],\n            dynamic=True,\n            show=True,\n        ),\n        IntInput(\n            name=\"chunk_size\",\n            display_name=\"Chunk Size\",\n            info=\"Batch size for processing embeddings\",\n            advanced=True,\n            value=1000,\n            dynamic=True,\n            show=True,\n        ),\n        SecretStrInput(\n            name=\"api_key\",\n            display_name=\"Embedding Provider API Key\",\n            info=\"Overrides global provider settings. Leave blank to use your pre-configured API Key.\",\n            advanced=True,\n            required=False,\n            dynamic=True,\n            show=True,\n        ),\n        BoolInput(\n            name=\"allow_duplicates\",\n            display_name=\"Allow Duplicates\",\n            info=\"Allow duplicate rows in the knowledge base\",\n            advanced=True,\n            value=False,\n            dynamic=True,\n            show=True,\n        ),\n        StrInput(\n            name=\"metadata_json\",\n            display_name=\"Metadata\",\n            info=(\n                \"Optional JSON object of user metadata applied to every chunk produced by this \"\n                'run (e.g. {\"tag\": \"invoice\", \"year\": \"2026\"}). Same shape as the upload modal '\n                \"Metadata section so chunks browser filters + Knowledge retrieval metadata_filter \"\n                \"work uniformly across upload, folder, and flow-driven ingestion. Malformed JSON is \"\n                \"ignored with a warning rather than failing the run.\"\n            ),\n            advanced=True,\n            required=False,\n            dynamic=True,\n            show=True,\n        ),\n        # --- Retrieve-only inputs (default-hidden; shown when mode == Retrieve) -\n        MessageTextInput(\n            name=\"search_query\",\n            display_name=\"Search Query\",\n            info=\"Optional search query to filter knowledge base data.\",\n            tool_mode=True,\n            dynamic=True,\n            show=False,\n        ),\n        IntInput(\n            name=\"top_k\",\n            display_name=\"Top K Results\",\n            info=\"Number of top results to return from the knowledge base.\",\n            value=5,\n            advanced=True,\n            required=False,\n            dynamic=True,\n            show=False,\n        ),\n        BoolInput(\n            name=\"include_metadata\",\n            display_name=\"Include Metadata\",\n            info=\"Whether to include all metadata in the output. If false, only content is returned.\",\n            value=True,\n            advanced=False,\n            dynamic=True,\n            show=False,\n        ),\n        BoolInput(\n            name=\"include_embeddings\",\n            display_name=\"Include Embeddings\",\n            info=\"Whether to include embeddings in the output. Only applicable if 'Include Metadata' is enabled.\",\n            value=False,\n            advanced=True,\n            dynamic=True,\n            show=False,\n        ),\n        MessageTextInput(\n            name=\"metadata_filter\",\n            display_name=\"Metadata Filter\",\n            info=(\n                \"Optional JSON object of user-metadata key/value pairs. Only chunks \"\n                'whose source_metadata matches every key are returned (e.g. {\"tag\": \"invoice\"} '\n                'or {\"tag\": [\"invoice\", \"audit\"]} for OR-of-values). Backends without '\n                \"native filtering apply the match client-side after retrieval.\"\n            ),\n            advanced=True,\n            dynamic=True,\n            show=False,\n        ),\n    ]\n\n    # ------ Outputs -------------------------------------------------------\n    # Both outputs are declared at the class level so the runtime can\n    # dispatch to either method depending on which one the saved flow has\n    # wired up. ``update_outputs`` filters the canvas-visible output per\n    # selected mode; see ``TypeConverterComponent`` and ``MemoryComponent``\n    # for the same pattern.\n    #\n    # Output names match the legacy ``KnowledgeIngestionComponent``\n    # (``dataframe_output``) and ``KnowledgeBaseComponent``\n    # (``retrieve_data``) so saved flow edges keyed on those names resolve\n    # cleanly against the merged component.\n    outputs = [\n        Output(\n            display_name=\"Results\",\n            name=\"dataframe_output\",\n            method=\"build_kb_info\",\n            types=[\"JSON\"],\n            selected=\"JSON\",\n        ),\n        Output(\n            display_name=\"Results\",\n            name=\"retrieve_data\",\n            method=\"retrieve_data\",\n            info=\"Returns the data from the selected knowledge base.\",\n            types=[\"Table\"],\n            selected=\"Table\",\n        ),\n    ]\n\n    # ------ Mode-driven UI updates ---------------------------------------\n    async def update_frontend_node(self, new_frontend_node: dict, current_frontend_node: dict):\n        \"\"\"Sync the visible output with the current ``mode`` value on canvas load.\n\n        Saved flows hit this path: ``update_outputs`` is normally only triggered\n        by ``real_time_refresh`` field edits, so without this re-sync the\n        canvas could land with both outputs visible.\n        \"\"\"\n        await super().update_frontend_node(new_frontend_node, current_frontend_node)\n        template = new_frontend_node.get(\"template\", {})\n        mode_value = template.get(\"mode\", {}).get(\"value\", MODE_INGEST)\n        self.update_outputs(new_frontend_node, \"mode\", mode_value)\n        # The rebuilt template carries class-default (Ingest) ``show`` flags,\n        # so a saved Retrieve node returned ``search_query`` hidden and the canvas\n        # silently dropped the edge feeding it.\n        self._apply_mode_visibility(template)\n        return new_frontend_node\n\n    def _apply_mode_visibility(self, build_config, field_name: str | None = None, field_value: Any = None):\n        \"\"\"Show only the inputs that belong to the effective mode.\"\"\"\n        current_mode = build_config.get(\"mode\", {}).get(\"value\") if isinstance(build_config, dict) else None\n        if field_name == \"mode\":\n            current_mode = field_value\n        # Map legacy/emoji-prefixed labels onto the current canonical values so\n        # flows saved before the label change still toggle visibility correctly.\n        if _is_retrieve_mode(current_mode):\n            current_mode = MODE_RETRIEVE\n        elif current_mode not in self.mode_config:\n            current_mode = MODE_INGEST\n        return set_current_fields(\n            build_config=build_config if isinstance(build_config, dotdict) else dotdict(build_config),\n            action_fields=self.mode_config,\n            selected_action=current_mode,\n            default_fields=self.default_keys,\n            func=set_field_display,\n        )\n\n    def update_outputs(self, frontend_node: dict, field_name: str, field_value: Any) -> dict:\n        \"\"\"Filter visible outputs to match the selected mode.\n\n        Triggered by the ``mode`` ``TabInput`` ``real_time_refresh`` flag.\n        The class-level ``outputs`` list always carries both entries so the\n        runtime can resolve either ``build_kb_info`` or ``retrieve_data``\n        regardless of canvas visibility \u2014 we just hide the unused one here.\n        \"\"\"\n        if field_name != \"mode\":\n            return frontend_node\n        if _is_retrieve_mode(field_value):\n            frontend_node[\"outputs\"] = [\n                Output(\n                    display_name=\"Results\",\n                    name=\"retrieve_data\",\n                    method=\"retrieve_data\",\n                    info=\"Returns the data from the selected knowledge base.\",\n                    types=[\"Table\"],\n                    selected=\"Table\",\n                )\n            ]\n        else:\n            frontend_node[\"outputs\"] = [\n                Output(\n                    display_name=\"Results\",\n                    name=\"dataframe_output\",\n                    method=\"build_kb_info\",\n                    types=[\"JSON\"],\n                    selected=\"JSON\",\n                )\n            ]\n        return frontend_node\n\n    async def update_build_config(\n        self,\n        build_config,\n        field_value: Any,\n        field_name: str | None = None,\n    ):\n        \"\"\"Refresh KB options, drive the create-KB dialog, and hide off-mode fields.\"\"\"\n        # Astra-cloud gate covers both ingest and retrieve paths.\n        raise_error_if_astra_cloud_disable_component(astra_error_msg)\n\n        # Always populate the create-KB dialog's embedding-model options so the\n        # ModelInput renders correctly regardless of which input triggered the\n        # refresh.\n        try:\n            dialog_template = (\n                build_config[\"knowledge_base\"]\n                .get(\"dialog_inputs\", {})\n                .get(\"fields\", {})\n                .get(\"data\", {})\n                .get(\"node\", {})\n                .get(\"template\", {})\n            )\n            if \"02_embedding_model\" in dialog_template:\n                embedding_options = get_embedding_model_options(user_id=self.user_id)\n                dialog_template[\"02_embedding_model\"][\"options\"] = embedding_options\n        except Exception:  # noqa: BLE001\n            self.log(\"Failed to populate embedding model options in dialog\")\n\n        # KB-picker refresh + create-new-KB flow (lifted from the legacy ingestion\n        # component verbatim; relied on by both the canvas refresh button and the\n        # dialog-submit path).\n        if field_name == \"knowledge_base\":\n            # Lazy import keeps lfx importable without langflow installed.\n            from langflow.services.database.models.user.crud import get_user_by_id\n\n            async with session_scope() as db:\n                if not self.user_id:\n                    msg = \"User ID is required for fetching knowledge base list.\"\n                    raise ValueError(msg)\n                current_user = await get_user_by_id(db, self.user_id)\n                if not current_user:\n                    msg = f\"User with ID {self.user_id} not found.\"\n                    raise ValueError(msg)\n            if isinstance(field_value, dict) and \"01_new_kb_name\" in field_value:\n                model_selection = field_value[\"02_embedding_model\"]\n                if isinstance(model_selection, dict):\n                    model_selection = [model_selection]\n\n                backend_type, backend_config = self._normalize_backend_selection(\n                    field_value.get(\"03_knowledge_backend\")\n                )\n                ensure_storage_routing_allowed(backend_config, is_superuser=bool(current_user.is_superuser))\n                new_kb_name = field_value[\"01_new_kb_name\"]\n                from langflow.api.utils.kb_helpers import local_chroma_rejection_reason, validate_kb_name\n\n                validate_kb_name(new_kb_name)\n                rejection = local_chroma_rejection_reason(backend_type, backend_config)\n                if rejection:\n                    raise ValueError(rejection)\n\n                embed_model = get_embeddings(\n                    model=model_selection,\n                    user_id=self.user_id,\n                )\n\n                try:\n                    await asyncio.wait_for(\n                        asyncio.to_thread(embed_model.embed_query, \"test\"),\n                        timeout=10,\n                    )\n                except TimeoutError as e:\n                    msg = \"Embedding validation timed out. Please verify network connectivity and key.\"\n                    raise ValueError(msg) from e\n                except Exception as e:\n                    msg = f\"Embedding validation failed: {e!s}\"\n                    raise ValueError(msg) from e\n\n                build_config[\"knowledge_base\"][\"value\"] = new_kb_name\n                await self._create_knowledge_base_record(\n                    user_id=self.user_id,\n                    name=field_value[\"01_new_kb_name\"],\n                    model_selection=model_selection,\n                    backend_type=backend_type,\n                    backend_config=backend_config,\n                )\n\n            build_config[\"knowledge_base\"][\"options\"] = await get_knowledge_bases(user_id=self.user_id)\n            if build_config[\"knowledge_base\"][\"value\"] not in build_config[\"knowledge_base\"][\"options\"]:\n                build_config[\"knowledge_base\"][\"value\"] = None\n\n        # Honor the current mode regardless of which field triggered the refresh.\n        return self._apply_mode_visibility(build_config, field_name, field_value)\n\n    # =====================================================================\n    #                       INGESTION CODE PATH\n    # =====================================================================\n    @staticmethod\n    def _scalar_notna(value) -> bool:\n        \"\"\"Check if a value is not NA, safely handling arrays and sequences.\n\n        ``pd.notna`` returns an array when given an array-like input, which\n        cannot be used directly in a boolean context.  This helper collapses\n        the result to a single scalar ``bool``.\n        \"\"\"\n        result = pd.notna(value)\n        if hasattr(result, \"__iter__\") and not isinstance(result, str):\n            import numpy as np\n\n            arr = np.asarray(result)\n            return arr.size > 0 and arr.all()\n        return bool(result)\n\n    def _validate_column_config(self, df_source: pd.DataFrame) -> list[dict[str, Any]]:\n        \"\"\"Validate column configuration using Structured Output patterns.\"\"\"\n        if not self.column_config:\n            msg = \"Column configuration cannot be empty\"\n            raise ValueError(msg)\n\n        config_list = self.column_config if isinstance(self.column_config, list) else []\n\n        df_columns = set(df_source.columns)\n        for config in config_list:\n            col_name = config.get(\"column_name\")\n            if col_name not in df_columns:\n                msg = f\"Column '{col_name}' not found in DataFrame. Available columns: {sorted(df_columns)}\"\n                raise ValueError(msg)\n\n        return config_list\n\n    @staticmethod\n    def _extract_source_types_from_df(df_source: pd.DataFrame) -> set[str]:\n        \"\"\"Pull file extensions out of common path/name columns on the source DataFrame.\n\n        The direct-upload ingestion path stores extensions in\n        the KB row's ``source_types`` so the KB list can render the\n        correct file-type icon. When ingestion happens via a connected\n        ``input_df`` (e.g. File \u2192 Knowledge) the same field stayed empty and\n        the icon defaulted to a blank tile. We look at the well-known columns\n        the File / S3 / cloud-storage components produce and collect any\n        plausible extension so the icon is consistent across both flows.\n        \"\"\"\n        candidate_columns = (\"file_path\", \"file_name\", \"filename\", \"source\", \"path\", \"mimetype\")\n        extensions: set[str] = set()\n        for col in candidate_columns:\n            if col not in df_source.columns:\n                continue\n            for value in df_source[col].dropna():\n                ext = KnowledgeComponent._extension_from_value(value)\n                # Drop anything that doesn't look like an extension (URL\n                # query strings, version segments, etc.) \u2014 the icon palette\n                # keys off short alphanumeric tokens like \"pdf\"/\"docx\".\n                if ext:\n                    extensions.add(ext)\n        return extensions\n\n    @staticmethod\n    def _extension_from_value(value: Any) -> str | None:\n        \"\"\"Return a normalized file-extension token from a path / filename / MIME string.\n\n        Accepts values like ``\"report.PDF\"``, ``\"/docs/notes.txt\"``, or\n        ``\"application/pdf\"`` and returns ``\"pdf\"`` / ``\"txt\"``. Returns\n        ``None`` if no plausible extension can be derived.\n        \"\"\"\n        extension_length_limit = 10\n        if value is None:\n            return None\n        text = str(value).strip()\n        if not text:\n            return None\n        # MIME types like ``application/pdf`` carry the canonical extension\n        # in the subtype slot \u2014 preserve them so File / S3 messages keyed\n        # only on ``mimetype`` still resolve to an icon.\n        if \"/\" in text and \".\" not in text.rsplit(\"/\", 1)[-1]:\n            subtype = text.rsplit(\"/\", 1)[-1].strip().lower()\n            return subtype if subtype and len(subtype) <= extension_length_limit and subtype.isalnum() else None\n        if \".\" not in text:\n            return None\n        ext = text.rsplit(\".\", 1)[-1].strip().lower()\n        if ext and len(ext) <= extension_length_limit and ext.isalnum():\n            return ext\n        return None\n\n    @classmethod\n    def _extract_source_types_from_mapping(cls, mapping: Any) -> set[str]:\n        \"\"\"Pull extensions from a Message/Data-style ``data`` dict or a plain dict.\"\"\"\n        if not isinstance(mapping, dict):\n            return set()\n        candidate_keys = (\"file_path\", \"file_name\", \"filename\", \"source\", \"path\", \"mimetype\")\n        extensions: set[str] = set()\n        for key in candidate_keys:\n            ext = cls._extension_from_value(mapping.get(key))\n            if ext:\n                extensions.add(ext)\n        return extensions\n\n    @classmethod\n    def _extract_source_types_from_input(cls, input_value: Any) -> set[str]:\n        \"\"\"Pull file extensions out of the raw component input.\n\n        ``convert_to_dataframe`` strips Message/Data fields down to ``text``\n        when projecting onto a DataFrame, so file metadata attached to a\n        File-component \"Raw Content\" output never reaches\n        ``_extract_source_types_from_df``. Looking at the raw input first\n        keeps the KB icon consistent with direct upload.\n        \"\"\"\n        if input_value is None:\n            return set()\n        if isinstance(input_value, list):\n            extensions: set[str] = set()\n            for item in input_value:\n                extensions |= cls._extract_source_types_from_input(item)\n            return extensions\n        if isinstance(input_value, pd.DataFrame):\n            return cls._extract_source_types_from_df(input_value)\n        # Message / Data / JSON all expose a ``data`` dict via the lfx schema.\n        mapping = getattr(input_value, \"data\", None)\n        if mapping is None and isinstance(input_value, dict):\n            mapping = input_value\n        return cls._extract_source_types_from_mapping(mapping)\n\n    async def _refresh_kb_stats(\n        self,\n        *,\n        kb_record_id: Any,\n        backend: BaseVectorStoreBackend,\n        extensions: set[str],\n    ) -> None:\n        \"\"\"Recount the KB from its vector store and write the totals onto the row.\n\n        Backend-agnostic: counts come from ``count`` / ``iter_documents`` /\n        ``storage_size_bytes``, so a Chroma, OpenSearch, or pgVector KB all report\n        real numbers. ``extensions`` are merged into ``source_types`` so the KB\n        list renders the right file-type icon regardless of which ingestion route\n        produced the chunks.\n\n        Best-effort: the chunks are already written and committed by the time this\n        runs, so a stats refresh failure must not fail the ingestion.\n        \"\"\"\n        if kb_record_id is None:\n            return\n        try:\n            from langflow.api.utils import knowledge_base_service\n        except ImportError:\n            return\n\n        try:\n            chunks = await backend.count()\n            characters = 0\n            words = 0\n            async with aclosing(backend.iter_documents()) as batches:\n                async for batch in batches:\n                    for document in batch:\n                        characters += len(document.content)\n                        words += len(document.content.split())\n\n            record = await knowledge_base_service.get_by_id(kb_record_id)\n            existing = set(record.source_types or []) if record is not None else set()\n\n            await knowledge_base_service.update_stats(\n                kb_record_id,\n                chunks=chunks,\n                words=words,\n                characters=characters,\n                size_bytes=await backend.storage_size_bytes(),\n                source_types=sorted(existing | extensions),\n            )\n        except Exception as e:  # noqa: BLE001 \u2014 stats are cosmetic; chunks are already written\n            self.log(f\"Warning: Could not refresh knowledge base stats: {e}\")\n\n    @staticmethod\n    def _default_backend_selection() -> tuple[str, dict[str, Any]]:\n        \"\"\"Deployment default for a new KB with no explicit backend chosen.\n\n        Environment-driven: pgVector when ``PGVECTOR_CONNECTION_STRING`` is set,\n        else OpenSearch when ``OPENSEARCH_URL`` is set, else SQLite. An\n        unspecified selection (headless flows, multi-replica) therefore lands on\n        the deployment's shared store.\n        \"\"\"\n        from lfx.base.knowledge_bases.backends.postgres import resolve_default_kb_backend\n\n        backend_type = resolve_default_kb_backend()\n        if backend_type == BackendType.OPENSEARCH.value:\n            return backend_type, dict(_DEFAULT_OPENSEARCH_CONFIG)\n        return backend_type, {}\n\n    @classmethod\n    def _normalize_backend_selection(cls, value: Any) -> tuple[str, dict[str, Any]]:\n        \"\"\"Normalize a DBProviderInput value into backend type/config.\"\"\"\n        if not value:\n            return cls._default_backend_selection()\n\n        backend_config: dict[str, Any]\n        if isinstance(value, str):\n            backend_type, backend_config = value, {}\n        elif isinstance(value, dict):\n            backend_type = str(value.get(\"backend_type\") or value.get(\"id\") or BackendType.SQLITE.value)\n            backend_config = value.get(\"backend_config\") or value.get(\"config\") or {}\n        else:\n            msg = \"Invalid vector-store backend selection.\"\n            raise TypeError(msg)\n        if not isinstance(backend_config, dict):\n            msg = \"Vector-store backend configuration must be an object.\"\n            raise TypeError(msg)\n        if backend_type == BackendType.OPENSEARCH.value:\n            return backend_type, {**_DEFAULT_OPENSEARCH_CONFIG, **backend_config}\n        if backend_type in {BackendType.SQLITE.value, BackendType.POSTGRES.value}:\n            return backend_type, backend_config\n        msg = \"Chroma is retired. Choose SQLite, pgVector, or OpenSearch for a new knowledge base.\"\n        raise ValueError(msg)\n\n    async def _create_knowledge_base_record(\n        self,\n        *,\n        user_id: Any,\n        name: str,\n        model_selection: list[dict[str, Any]],\n        backend_type: str,\n        backend_config: dict[str, Any],\n    ) -> None:\n        \"\"\"Persist the component-created KB in the DB when Langflow is available.\"\"\"\n        from langflow.api.utils import knowledge_base_service\n\n        await knowledge_base_service.create_record(\n            user_id=user_id,\n            name=name,\n            model_selection=model_selection,\n            column_config=_normalize_column_config(self.column_config) if isinstance(self.column_config, list) else [],\n            backend_type=backend_type,\n            backend_config=backend_config,\n        )\n\n    def _build_column_metadata(self, config_list: list[dict[str, Any]], df_source: pd.DataFrame) -> dict[str, Any]:\n        \"\"\"Build detailed column metadata.\"\"\"\n        metadata: dict[str, Any] = {\n            \"total_columns\": len(df_source.columns),\n            \"mapped_columns\": len(config_list),\n            \"unmapped_columns\": len(df_source.columns) - len(config_list),\n            \"columns\": [],\n            \"summary\": {\"vectorized_columns\": [], \"identifier_columns\": []},\n        }\n\n        for config in config_list:\n            col_name = config.get(\"column_name\")\n            vectorize = coalesce_bool(config.get(\"vectorize\"))\n            identifier = coalesce_bool(config.get(\"identifier\"))\n\n            metadata[\"columns\"].append(\n                {\n                    \"name\": col_name,\n                    \"vectorize\": vectorize,\n                    \"identifier\": identifier,\n                }\n            )\n\n            if vectorize:\n                metadata[\"summary\"][\"vectorized_columns\"].append(col_name)\n            if identifier:\n                metadata[\"summary\"][\"identifier_columns\"].append(col_name)\n\n        return metadata\n\n    async def _create_vector_store(\n        self,\n        df_source: pd.DataFrame,\n        config_list: list[dict[str, Any]],\n        embedding_function,\n        *,\n        job_id: uuid.UUID | None = None,\n    ) -> tuple[BaseVectorStoreBackend, set[int]]:\n        \"\"\"Create the vector store and return the input row positions actually stored.\"\"\"\n        from langflow.api.utils.kb_helpers import backend_for_name\n\n        owner_id = self._user_uuid\n        if owner_id is None:\n            msg = \"User ID is required for knowledge base storage.\"\n            raise ValueError(msg)\n        backend = await backend_for_name(owner_id, self.knowledge_base, embedding_function=embedding_function)\n        try:\n            return await self._ingest_into_backend(backend, df_source, config_list, embedding_function, job_id=job_id)\n        except BaseException:\n            await _finish_despite_cancellation(backend.teardown())\n            raise\n\n    async def _ingest_into_backend(\n        self,\n        backend: BaseVectorStoreBackend,\n        df_source: pd.DataFrame,\n        config_list: list[dict[str, Any]],\n        embedding_function,\n        *,\n        job_id: uuid.UUID | None = None,\n    ) -> tuple[BaseVectorStoreBackend, set[int]]:\n        await backend.ensure_ready()\n\n        from langflow.services.knowledge_base_storage.runtime import operation, resolve_record\n\n        record = await resolve_record(self._user_uuid, self.knowledge_base)\n        async with operation(record, shared=True):\n            existing_ids = set()\n            if not self.allow_duplicates:\n                async with aclosing(backend.iter_documents()) as batches:\n                    async for batch in batches:\n                        for document in batch:\n                            doc_id = document.metadata.get(\"_id\")\n                            if doc_id:\n                                existing_ids.add(doc_id)\n\n        # Keep row positions through both duplicate checks so run history can\n        # distinguish stored rows from rows skipped before or after embedding.\n        data_objects = await self._convert_df_to_data_objects(df_source, config_list)\n\n        user_metadata_tag = self._resolve_user_metadata_tag()\n\n        documents = []\n        row_indices = []\n        for row_index, data_obj in enumerate(data_objects):\n            if not self.allow_duplicates and data_obj.data[\"_id\"] in existing_ids:\n                continue\n            doc = data_obj.to_lc_document()\n            if user_metadata_tag:\n                doc.metadata[\"source_metadata\"] = user_metadata_tag\n            documents.append(doc)\n            row_indices.append(row_index)\n        if not documents:\n            return backend, set()\n\n        # No storage lease may surround the provider call. The write lease\n        # below rechecks routing and IDs after concurrent operations finish.\n        vectors = await self._embed_unless_job_cancelled(\n            embedding_function, [doc.page_content for doc in documents], job_id\n        )\n        if len(vectors) != len(documents):\n            msg = \"Embedding provider returned an incorrect number of vectors\"\n            raise BackendConfigurationError(msg)\n        embedded = [\n            (row_index, IngestedDocument(doc.page_content, doc.metadata, vector, id=doc.id))\n            for row_index, doc, vector in zip(row_indices, documents, vectors, strict=True)\n        ]\n        async with operation(record):\n            # A cancel that landed after the embeddings returned must still\n            # keep this run's chunks out of the KB.\n            await self._raise_if_job_cancelled(job_id)\n            if not self.allow_duplicates:\n                current_ids = set()\n                async with aclosing(backend.iter_documents()) as batches:\n                    async for batch in batches:\n                        current_ids.update(doc.metadata[\"_id\"] for doc in batch if doc.metadata.get(\"_id\"))\n                unique = []\n                for row_index, doc in embedded:\n                    doc_id = doc.metadata.get(\"_id\")\n                    if doc_id and doc_id in current_ids:\n                        continue\n                    if doc_id:\n                        current_ids.add(doc_id)\n                    unique.append((row_index, doc))\n                embedded = unique\n\n            if embedded:\n                await backend.add_embedded_documents([doc for _, doc in embedded])\n                self.log(f\"Added {len(embedded)} documents to vector store '{self.knowledge_base}'\")\n\n        return backend, {row_index for row_index, _ in embedded}\n\n    async def _embed_unless_job_cancelled(\n        self, embedding_function, texts: list[str], job_id: uuid.UUID | None\n    ) -> list[list[float]]:\n        \"\"\"Embed ``texts``, giving up as soon as the run's job is cancelled.\n\n        The Knowledge page cancels a run by marking its job. The embedding call\n        is the long, write-free part of a flow ingestion, so it is the safe\n        place to stop.\n        \"\"\"\n        await self._raise_if_job_cancelled(job_id)\n        if job_id is None:\n            return await embedding_function.aembed_documents(texts)\n\n        # A task group would wrap provider errors in an ExceptionGroup, so a\n        # plain watcher task cancels this scope instead.\n        with anyio.CancelScope() as scope:\n            watcher = asyncio.ensure_future(self._cancel_when_job_cancelled(job_id, scope))\n            try:\n                return await embedding_function.aembed_documents(texts)\n            finally:\n                # The watcher may be mid-poll; let its DB session close.\n                watcher.cancel()\n                with anyio.CancelScope(shield=True):\n                    await asyncio.wait({watcher})\n        raise self._cancelled_error()\n\n    async def _cancel_when_job_cancelled(self, job_id: uuid.UUID, scope: anyio.CancelScope) -> None:\n        # The cancel is a job-row write from another request, so it is polled.\n        while True:\n            await asyncio.sleep(JOB_CANCEL_POLL_SECONDS)\n            if await self._job_cancelled(job_id):\n                scope.cancel()\n                return\n\n    async def _raise_if_job_cancelled(self, job_id: uuid.UUID | None) -> None:\n        if await self._job_cancelled(job_id):\n            raise self._cancelled_error()\n\n    def _cancelled_error(self) -> KnowledgeIngestionCancelledError:\n        return KnowledgeIngestionCancelledError(f\"Ingestion into '{self.knowledge_base}' was cancelled.\")\n\n    async def _job_cancelled(self, job_id: uuid.UUID | None) -> bool:\n        if job_id is None:\n            return False\n        try:\n            from langflow.services.database.models.jobs.model import JobStatus\n            from langflow.services.deps import get_job_service\n        except ImportError:\n            return False\n        try:\n            job = await get_job_service().get_job_by_job_id(job_id)\n        except Exception as exc:  # noqa: BLE001 \u2014 a failed poll must not abort the ingestion\n            self.log(f\"Could not check whether ingestion job {job_id} was cancelled: {exc}\")\n            return False\n        return job is not None and job.status == JobStatus.CANCELLED\n\n    async def _convert_df_to_data_objects(\n        self,\n        df_source: pd.DataFrame,\n        config_list: list[dict[str, Any]],\n        existing_ids: set[str] | None = None,\n    ) -> list[Data]:\n        \"\"\"Convert DataFrame to Data objects for vector store.\"\"\"\n        data_objects: list[Data] = []\n\n        if existing_ids is None:\n            existing_ids = set()\n\n        content_cols = []\n        identifier_cols = []\n\n        for config in config_list:\n            col_name = config.get(\"column_name\")\n\n            if coalesce_bool(config.get(\"vectorize\")):\n                content_cols.append(col_name)\n            if coalesce_bool(config.get(\"identifier\")):\n                identifier_cols.append(col_name)\n\n        for _, row in df_source.iterrows():\n            identifier_parts = [str(row[col]) for col in content_cols if col in row and self._scalar_notna(row[col])]\n\n            page_content = \" \".join(identifier_parts)\n\n            data_dict = {\n                \"text\": page_content,\n            }\n\n            if identifier_cols:\n                identifier_parts = [\n                    str(row[col]) for col in identifier_cols if col in row and self._scalar_notna(row[col])\n                ]\n                page_content = \" \".join(identifier_parts)\n\n            for col in df_source.columns:\n                if col not in content_cols and col in row and self._scalar_notna(row[col]):\n                    value = row[col]\n                    data_dict[col] = str(value)\n\n            page_content_hash = hashlib.sha256(page_content.encode()).hexdigest()\n            data_dict[\"_id\"] = page_content_hash\n\n            if not self.allow_duplicates and page_content_hash in existing_ids:\n                self.log(f\"Skipping duplicate row with hash {page_content_hash}\")\n                continue\n\n            data_obj = Data(data=data_dict)\n            data_objects.append(data_obj)\n\n        return data_objects\n\n    def is_valid_collection_name(self, name: str) -> bool:\n        \"\"\"Return whether ``name`` satisfies the shared collection-name contract.\"\"\"\n        return is_valid_kb_collection_name(name)\n\n    def _resolve_store_path(\n        self,\n        kb_user: str,\n        backend_type: str,\n        backend_config: dict[str, Any] | None,\n    ) -> Path | None:\n        \"\"\"Return this KB's on-disk directory, or ``None`` when it needs no disk.\n\n        Only local Chroma stores anything on this box; every other backend\n        resolves to ``None`` and never touches the filesystem.\n        \"\"\"\n        # Lazy import keeps lfx importable without langflow's DB services.\n        from langflow.api.utils.kb_helpers import resolve_local_store_path\n\n        return resolve_local_store_path(\n            self.knowledge_base,\n            kb_user,\n            backend_type=backend_type,\n            backend_config=backend_config,\n        )\n\n    async def _kb_username(self) -> str:\n        \"\"\"Resolve the owning user's username (used to root a local-Chroma path).\"\"\"\n        cached = getattr(self, \"_cached_kb_username\", None)\n        if cached is not None:\n            return cached\n\n        # Lazy import to keep ``lfx`` importable standalone \u2014 langflow's\n        # user/DB models are not always available at module load time.\n        from langflow.services.database.models.user.crud import get_user_by_id\n\n        async with session_scope() as db:\n            if not self.user_id:\n                msg = \"User ID is required for fetching knowledge base path.\"\n                raise ValueError(msg)\n            current_user = await get_user_by_id(db, self.user_id)\n            if not current_user:\n                msg = f\"User with ID {self.user_id} not found.\"\n                raise ValueError(msg)\n            self._cached_kb_username = current_user.username\n\n        return self._cached_kb_username\n\n    async def _kb_path(self) -> Path | None:\n        \"\"\"Resolve the KB's local directory from its configured backend.\"\"\"\n        cached_path = getattr(self, \"_cached_kb_path\", None)\n        if cached_path is not None:\n            return cached_path\n\n        backend_type, backend_config = await self._resolve_backend_config()\n        kb_user = await self._kb_username()\n        self._cached_kb_path = self._resolve_store_path(kb_user, backend_type, backend_config)\n        return self._cached_kb_path\n\n    def _resolve_user_metadata_tag(self) -> str:\n        \"\"\"Return the JSON-encoded user metadata tag for chunk writes.\"\"\"\n        raw = getattr(self, \"metadata_json\", None)\n        if not raw:\n            return \"\"\n        text = raw.strip() if isinstance(raw, str) else raw\n        if not text:\n            return \"\"\n        try:\n            decoded = json.loads(text)\n        except (TypeError, json.JSONDecodeError) as exc:\n            self.log(f\"KnowledgeComponent: metadata_json is not valid JSON ({exc}); skipping metadata stamp.\")\n            return \"\"\n        if not isinstance(decoded, dict):\n            self.log(\"KnowledgeComponent: metadata_json must decode to a JSON object; skipping metadata stamp.\")\n            return \"\"\n        return json.dumps(decoded, sort_keys=True)\n\n    async def build_kb_info(self) -> Data:\n        \"\"\"Main ingestion routine \u2192 returns a dict with KB metadata.\n\n        The annotation is intentionally narrowed to ``Data`` even though the\n        cross-mode fallback below may return a ``DataFrame`` from\n        ``retrieve_data``. The frontend builds React-Flow handle IDs from\n        this output's type list; widening it to ``Data | DataFrame`` makes\n        the API advertise ``[\"JSON\", \"Table\"]`` for the ingest output and\n        breaks every saved-edge sourceHandle that was generated against\n        a single-type handle (BUG-02). Python doesn't enforce return\n        annotations at runtime, so the rare fallback path keeps working.\n        \"\"\"\n        if _is_retrieve_mode(getattr(self, \"mode\", MODE_INGEST)):\n            return await self.retrieve_data()\n        raise_error_if_astra_cloud_disable_component(astra_error_msg)\n\n        run_id: uuid.UUID | None = None\n        run_job_id: uuid.UUID | None = None\n        run_summary: IngestionSummary | None = None\n        run_status: IngestionRunStatus = IngestionRunStatus.SUCCEEDED\n        run_error: str | None = None\n        kb_record_id: uuid.UUID | None = None\n        previous_kb_status = \"ready\"\n        previous_failure_reason: str | None = None\n        try:\n            input_value = self.input_df[0] if isinstance(self.input_df, list) else self.input_df\n            df_source: DataFrame = convert_to_dataframe(input_value, auto_parse=False)\n\n            config_list = self._validate_column_config(df_source)\n            column_metadata = self._build_column_metadata(config_list, df_source)\n\n            # Embedding config comes from the ``knowledge_base`` row \u2014 the sole\n            # authority. No sidecar is read, so ingestion resolves the same model\n            # on any replica.\n            stored_metadata = await self._get_kb_metadata()\n            # A cancelled run restores this status. Only an earlier failure is\n            # kept; \"empty\" is an API projection of a ready KB with no chunks.\n            if stored_metadata.get(\"status\") == \"failed\":\n                previous_kb_status = \"failed\"\n                previous_failure_reason = stored_metadata.get(\"failure_reason\")\n            model_selection = stored_metadata.get(\"model_selection\")\n            if model_selection:\n                model_selection = [model_selection] if isinstance(model_selection, dict) else list(model_selection)\n            else:\n                embedding_model_name = stored_metadata.get(\"embedding_model\")\n                embedding_provider = stored_metadata.get(\"embedding_provider\", \"Unknown\")\n                if embedding_model_name:\n                    try:\n                        all_options = get_embedding_model_options(user_id=self.user_id)\n                        match = next(\n                            (o for o in all_options if o.get(\"name\") == embedding_model_name),\n                            None,\n                        )\n                        if match:\n                            model_selection = [match]\n                        else:\n                            self.log(\n                                f\"Embedding model '{embedding_model_name}' (provider: {embedding_provider}) \"\n                                \"from the knowledge base record is no longer available in the model registry. \"\n                                \"Please re-create this knowledge base with a supported embedding model.\"\n                            )\n                            msg = (\n                                f\"Embedding model '{embedding_model_name}' is no longer recognized. \"\n                                \"The knowledge base was created with an older format and the model \"\n                                \"is not available in the current registry. \"\n                                \"Please re-create the knowledge base with a supported embedding model.\"\n                            )\n                            raise ValueError(msg)\n                    except ValueError:\n                        raise\n                    except Exception:  # noqa: BLE001\n                        self.log(\n                            f\"Failed to look up embedding model '{embedding_model_name}' in registry. \"\n                            \"Please re-create this knowledge base with a supported embedding model.\"\n                        )\n                        msg = (\n                            f\"Could not look up embedding model '{embedding_model_name}' \"\n                            f\"(provider: {embedding_provider}). \"\n                            \"Please re-create the knowledge base with a supported embedding model.\"\n                        )\n                        raise ValueError(msg)  # noqa: B904\n\n            if not model_selection:\n                msg = \"No embedding model configuration found. Please create the knowledge base first.\"\n                raise ValueError(msg)\n\n            # ``api_key=None`` falls through to the user's variables table and\n            # then the environment inside ``get_embeddings`` \u2014 the same\n            # resolution the server-side ingestion path uses. The component input\n            # only overrides it.\n            embedding_function = get_embeddings(\n                model=model_selection,\n                user_id=self.user_id,\n                api_key=self.api_key if getattr(self, \"api_key\", None) else None,\n                chunk_size=self.chunk_size,\n            )\n\n            kb_path = await self._kb_path()\n\n            run_id, run_job_id, run_summary, kb_record_id = await self._begin_ingestion_run(kb_path)\n            if kb_record_id is not None:\n                await self._record_kb_status(kb_record_id, \"ingesting\")\n\n            backend, stored_rows = await self._create_vector_store(\n                df_source, config_list, embedding_function=embedding_function, job_id=run_job_id\n            )\n\n            try:\n                # Stamp the KB with the file extensions we just ingested so\n                # the Knowledge Bases list renders the correct icon for\n                # flow-driven ingestion (input_df), matching direct upload.\n                # We look at the raw input first because ``convert_to_dataframe``\n                # drops Message/Data metadata fields (file_path, mimetype, \u2026)\n                # when projecting onto the DataFrame.\n                source_types = self._extract_source_types_from_input(input_value)\n                source_types |= self._extract_source_types_from_df(df_source)\n                # Storage routing returns a guarded proxy with the backend's\n                # methods, so a concrete-class check would skip the refresh.\n                await self._refresh_kb_stats(\n                    kb_record_id=kb_record_id,\n                    backend=backend,\n                    extensions=source_types,\n                )\n            finally:\n                await _finish_despite_cancellation(backend.teardown())\n\n            meta: dict[str, Any] = {\n                \"kb_id\": str(kb_record_id) if kb_record_id is not None else None,\n                \"kb_name\": self.knowledge_base,\n                \"rows\": len(df_source),\n                \"chunks_created\": len(stored_rows),\n                \"skipped\": len(df_source) - len(stored_rows),\n                \"column_metadata\": column_metadata,\n                \"path\": str(kb_path),\n                \"config_columns\": len(config_list),\n                \"timestamp\": datetime.now(tz=timezone.utc).isoformat(),\n            }\n\n            if run_summary is not None:\n                for row_index in range(len(df_source)):\n                    stored = row_index in stored_rows\n                    run_summary.record_item(\n                        IngestionItemResult(\n                            item_id=f\"{self.knowledge_base}:{row_index}\",\n                            display_name=f\"{self.knowledge_base} (row {row_index + 1})\",\n                            status=IngestionItemStatus.SUCCEEDED if stored else IngestionItemStatus.SKIPPED,\n                            chunks_created=int(stored),\n                        ),\n                    )\n\n            if kb_record_id is not None:\n                # Stats were already refreshed straight from the backend above.\n                await self._record_kb_status(kb_record_id, \"ready\")\n\n            self.status = (\n                f\"\u2705 KB **{self.knowledge_base}** saved \u00b7 {len(stored_rows)} chunks \"\n                f\"\u00b7 {len(df_source) - len(stored_rows)} rows skipped.\"\n            )\n\n            return Data(data=meta)\n\n        except asyncio.CancelledError:\n            run_status = IngestionRunStatus.CANCELLED\n            run_error = \"Flow ingestion was cancelled.\"\n            raise\n        except KnowledgeIngestionCancelledError as e:\n            run_status = IngestionRunStatus.CANCELLED\n            run_error = str(e)\n            raise\n        except (OSError, ValueError, RuntimeError, KeyError) as e:\n            run_status = IngestionRunStatus.FAILED\n            run_error = str(e) or e.__class__.__name__\n            msg = f\"Error during KB ingestion: {e}\"\n            raise RuntimeError(msg) from e\n        except Exception as e:\n            run_status = IngestionRunStatus.FAILED\n            run_error = str(e) or e.__class__.__name__\n            raise\n        finally:\n\n            async def settle_run() -> None:\n                if run_id is not None and run_summary is not None:\n                    await self._finalize_ingestion_run(\n                        run_id=run_id,\n                        job_id=run_job_id,\n                        summary=run_summary,\n                        status=run_status,\n                        error_message=run_error,\n                    )\n                if kb_record_id is not None:\n                    if run_status is IngestionRunStatus.FAILED:\n                        await self._record_kb_status(kb_record_id, \"failed\", failure_reason=run_error)\n                    elif run_status is IngestionRunStatus.CANCELLED:\n                        await self._record_kb_status(\n                            kb_record_id, previous_kb_status, failure_reason=previous_failure_reason\n                        )\n\n            # A stopped build can cancel this task more than once, and the\n            # writes that make the run terminal must survive every cancel.\n            await _finish_despite_cancellation(settle_run())\n\n    async def _begin_ingestion_run(\n        self,\n        kb_path: Path,\n    ) -> tuple[uuid.UUID | None, uuid.UUID | None, IngestionSummary | None, uuid.UUID | None]:\n        \"\"\"Create a parent ``Job`` and seed an ingestion-run row.\"\"\"\n        if not self.user_id:\n            self.log(\"No user_id on component; skipping ingestion-run tracking.\")\n            return None, None, None, None\n\n        try:\n            user_uuid = uuid.UUID(str(self.user_id))\n        except (ValueError, TypeError) as exc:\n            self.log(f\"Could not coerce user_id={self.user_id!r} to UUID; skipping run tracking ({exc}).\")\n            return None, None, None, None\n\n        try:\n            from langflow.api.utils import ingestion_run_service, knowledge_base_service\n            from langflow.services.database.models.jobs.model import JobStatus, JobType\n            from langflow.services.deps import get_job_service\n        except ImportError as exc:\n            self.log(f\"Run-history wiring unavailable; ingestion will not be recorded ({exc}).\")\n            return None, None, None, None\n\n        job_id: uuid.UUID | None = None\n        run_id: uuid.UUID | None = None\n        summary: IngestionSummary | None = None\n        kb_record_id: uuid.UUID | None = None\n        try:\n            kb_record = await knowledge_base_service.get_by_user_and_name(user_uuid, self.knowledge_base)\n            kb_record_id = kb_record.id if kb_record is not None else None\n\n            user_metadata = self._parse_user_metadata_dict()\n\n            job_id = uuid.uuid4()\n            job_service = get_job_service()\n            raw_flow_id = getattr(self, \"flow_id\", None)\n            flow_id_uuid = uuid.UUID(str(raw_flow_id)) if raw_flow_id else job_id\n            source = FlowComponentSource(\n                user_id=user_uuid,\n                source_config={\n                    \"knowledge_base\": self.knowledge_base,\n                    \"kb_path\": str(kb_path),\n                    \"flow_id\": str(flow_id_uuid),\n                },\n            )\n            summary = IngestionSummary(\n                kb_name=self.knowledge_base,\n                source_type=source.source_type.value,\n                user_id=user_uuid,\n                job_id=job_id,\n                source_config=source.describe().get(\"config\") or {},\n                user_metadata=user_metadata,\n            )\n            # Inserted running and already heartbeated, so an orphan sweep in\n            # another worker never sees this live run without a heartbeat.\n            await job_service.create_job(\n                job_id=job_id,\n                flow_id=flow_id_uuid,\n                job_type=JobType.INGESTION,\n                asset_id=kb_record_id,\n                asset_type=\"knowledge_base\",\n                user_id=user_uuid,\n                status=JobStatus.IN_PROGRESS,\n                heartbeat=True,\n            )\n\n            run_id = await ingestion_run_service.create_run(\n                kb_name=self.knowledge_base,\n                source=source,\n                job_id=job_id,\n                user_id=user_uuid,\n                kb_id=kb_record_id,\n                user_metadata=user_metadata,\n            )\n            if run_id is None:\n                return None, job_id, None, kb_record_id\n            await ingestion_run_service.mark_running(run_id)\n\n            self.log(f\"Started ingestion run job_id={job_id} kb_name={self.knowledge_base} kb_id={kb_record_id}\")\n            # Keeps the job heartbeated until _finalize_ingestion_run stops it.\n            await job_service.start_keep_alive(job_id)\n        except asyncio.CancelledError:\n            # The caller has not received the run ID yet, so it cannot finalize\n            # a cancellation that arrives while tracking is being initialized.\n            if job_id is not None and summary is not None:\n                await _finish_despite_cancellation(\n                    self._finalize_ingestion_run(\n                        run_id=run_id or job_id,\n                        job_id=job_id,\n                        summary=summary,\n                        status=IngestionRunStatus.CANCELLED,\n                        error_message=\"Flow ingestion was cancelled.\",\n                    )\n                )\n            raise\n        except Exception as exc:  # noqa: BLE001 \u2014 telemetry must never abort ingestion\n            self.log(f\"Could not begin ingestion-run tracking: {exc}\")\n            return None, None, None, kb_record_id\n        else:\n            return run_id, job_id, summary, kb_record_id\n\n    async def _finalize_ingestion_run(\n        self,\n        *,\n        run_id: uuid.UUID,\n        job_id: uuid.UUID | None,\n        summary: IngestionSummary,\n        status: IngestionRunStatus,\n        error_message: str | None,\n    ) -> None:\n        \"\"\"Persist the final summary and transition the parent Job.\"\"\"\n        try:\n            from langflow.api.utils import ingestion_run_service\n            from langflow.services.database.models.jobs.model import JobStatus\n            from langflow.services.deps import get_job_service\n        except ImportError as exc:\n            self.log(f\"Run-history wiring unavailable; ingestion-run finalize skipped ({exc}).\")\n            return\n\n        try:\n            if job_id is not None:\n                # The heartbeat also rewrites job_metadata, so it stops first.\n                await get_job_service().stop_keep_alive(job_id)\n            await ingestion_run_service.finalize_run(\n                run_id,\n                summary=summary,\n                status=status,\n                error_message=error_message,\n            )\n            if job_id is not None:\n                terminal_status = {\n                    IngestionRunStatus.FAILED: JobStatus.FAILED,\n                    IngestionRunStatus.CANCELLED: JobStatus.CANCELLED,\n                }.get(status, JobStatus.COMPLETED)\n                await get_job_service().update_job_status(job_id, terminal_status, finished_timestamp=True)\n        except Exception as exc:  # noqa: BLE001 \u2014 telemetry must never re-raise\n            self.log(f\"Could not finalize ingestion-run tracking: {exc}\")\n\n    async def _record_kb_status(\n        self,\n        kb_record_id: uuid.UUID,\n        status_value: str,\n        *,\n        failure_reason: str | None = None,\n    ) -> None:\n        \"\"\"Mirror Path A's KB-row status transitions.\"\"\"\n        try:\n            from langflow.api.utils import knowledge_base_service\n            from langflow.services.database.models.knowledge_base.model import KnowledgeBaseStatus\n        except ImportError:\n            return\n        try:\n            await knowledge_base_service.update_status(\n                kb_record_id,\n                status=KnowledgeBaseStatus(status_value),\n                failure_reason=failure_reason,\n            )\n        except Exception as exc:  # noqa: BLE001\n            self.log(f\"Could not update KB status to {status_value}: {exc}\")\n\n    def _parse_user_metadata_dict(self) -> dict[str, Any]:\n        \"\"\"Decode ``metadata_json`` to a dict, or ``{}`` on any error.\"\"\"\n        raw = getattr(self, \"metadata_json\", None)\n        if not raw:\n            return {}\n        text = raw.strip() if isinstance(raw, str) else raw\n        if not text:\n            return {}\n        try:\n            decoded = json.loads(text)\n        except (TypeError, json.JSONDecodeError):\n            return {}\n        return decoded if isinstance(decoded, dict) else {}\n\n    # =====================================================================\n    #                       RETRIEVAL CODE PATH\n    # =====================================================================\n    @property\n    def _user_uuid(self) -> uuid.UUID | None:\n        \"\"\"Return self.user_id as a UUID, converting from str if necessary.\"\"\"\n        if not self.user_id:\n            return None\n        return self.user_id if isinstance(self.user_id, uuid.UUID) else uuid.UUID(self.user_id)\n\n    async def _get_kb_metadata(self, knowledge_base: str | None = None) -> dict:\n        \"\"\"Load this knowledge base's embedding config from its database row.\n\n        The row is the sole authority. There is no on-disk sidecar to fall back\n        to, which is what lets retrieval work identically on a replica whose\n        filesystem never held the KB directory.\n        \"\"\"\n        raise_error_if_astra_cloud_disable_component(astra_error_msg)\n        try:\n            from langflow.api.utils import knowledge_base_service\n        except ImportError:\n            return {}\n\n        user_uuid = self._user_uuid\n        if user_uuid is None:\n            return {}\n        record = await knowledge_base_service.get_by_user_and_name(\n            user_uuid,\n            knowledge_base or self.knowledge_base,\n        )\n        if record is None:\n            return {}\n        return knowledge_base_service.record_to_metadata_dict(record)\n\n    async def _backend_from_record(self) -> tuple[str, dict[str, Any]] | None:\n        \"\"\"Read ``(backend_type, backend_config)`` off the KB's database row.\n\n        Returns ``None`` when there is no row to read \u2014 a legacy KB predating the\n        record, or bare lfx with no langflow installed. A lookup that *fails* is a\n        different situation and is allowed to propagate: guessing a backend after a\n        database error is how vectors end up in a store nothing queries.\n        \"\"\"\n        try:\n            from langflow.api.utils import knowledge_base_service\n        except ImportError:\n            return None\n\n        user_uuid = self._user_uuid\n        if user_uuid is None:\n            return None\n        record = await knowledge_base_service.get_by_user_and_name(user_uuid, self.knowledge_base)\n        if record is None:\n            return None\n        return record.backend_type or BackendType.CHROMA.value, record.backend_config or {}\n\n    async def _resolve_backend_config(self) -> tuple[str, dict[str, Any]]:\n        \"\"\"Resolve this KB's backend from its database row.\n\n        Ingestion and retrieval both land here, so they cannot disagree about\n        where this KB's vectors live. A backend that cannot be resolved raises\n        rather than defaulting to local storage: guessing is how a pod ends up\n        writing vectors to a local directory while queries follow the real config\n        to a remote store and silently find nothing.\n        \"\"\"\n        resolved = await self._backend_from_record()\n        if resolved is None:\n            msg = (\n                f\"Cannot determine the vector-store backend for knowledge base \"\n                f\"'{self.knowledge_base}': it has no database record. Refusing to fall \"\n                f\"back to local storage, which would write to a different store than \"\n                f\"queries read from. If this knowledge base predates database-backed \"\n                f\"metadata, adopt it with 'langflow reconcile-kb-from-disk'.\"\n            )\n            raise ValueError(msg)\n        return resolved\n\n    def _resolve_model_selection(self, metadata: dict[str, Any]) -> list[dict[str, Any]]:\n        \"\"\"Resolve the ``get_embeddings``-compatible model selection from metadata.\"\"\"\n        model_selection = metadata.get(\"model_selection\")\n        if model_selection:\n            selection_list = [model_selection] if isinstance(model_selection, dict) else list(model_selection)\n            return [self._hydrate_model_metadata(entry) for entry in selection_list]\n\n        embedding_model_name = metadata.get(\"embedding_model\")\n        embedding_provider = metadata.get(\"embedding_provider\", \"Unknown\")\n        if not embedding_model_name:\n            msg = (\n                f\"Knowledge base '{self.knowledge_base}' has no embedding model recorded; \"\n                \"re-create it with a supported embedding model.\"\n            )\n            raise ValueError(msg)\n\n        match = self._find_catalog_entry(embedding_model_name)\n        if match is None:\n            msg = (\n                f\"Embedding model '{embedding_model_name}' (provider '{embedding_provider}') \"\n                \"recorded for this knowledge base is no longer available in the model registry. \"\n                \"Please re-create the knowledge base with a supported embedding model.\"\n            )\n            raise ValueError(msg)\n        return [match]\n\n    def _hydrate_model_metadata(self, entry: dict[str, Any]) -> dict[str, Any]:\n        \"\"\"Fill in ``metadata.embedding_class`` / ``param_mapping`` if missing.\"\"\"\n        entry_metadata = entry.get(\"metadata\") or {}\n        has_class = bool(entry_metadata.get(\"embedding_class\"))\n        has_mapping = bool(entry_metadata.get(\"param_mapping\"))\n        if has_class and has_mapping:\n            return entry\n\n        model_name = entry.get(\"name\")\n        if not model_name:\n            return entry\n\n        catalog_entry = self._find_catalog_entry(model_name)\n        if catalog_entry is None:\n            return entry\n\n        catalog_metadata = catalog_entry.get(\"metadata\") or {}\n        merged_metadata = {**catalog_metadata, **entry_metadata}\n        return {**entry, \"metadata\": merged_metadata}\n\n    def _find_catalog_entry(self, model_name: str) -> dict[str, Any] | None:\n        \"\"\"Look up an embedding model by name in the unified-models catalog.\"\"\"\n        options = get_embedding_model_options(user_id=self.user_id)\n        return next((o for o in options if o.get(\"name\") == model_name), None)\n\n    async def retrieve_data(self) -> DataFrame:\n        \"\"\"Retrieve data from the selected knowledge base.\n\n        Annotation narrowed to ``DataFrame`` to keep this output's handle\n        type list at ``[\"Table\"]`` only; widening to a union surfaces\n        ``[\"Table\", \"JSON\"]`` on the API and breaks every starter-project\n        edge whose sourceHandle was stored against a single-type handle\n        (BUG-02). The cross-mode defensive fallback may still return a\n        ``Data`` at runtime \u2014 Python ignores return annotations, so\n        nothing breaks.\n        \"\"\"\n        if not _is_retrieve_mode(getattr(self, \"mode\", MODE_INGEST)):\n            return await self.build_kb_info()\n        raise_error_if_astra_cloud_disable_component(astra_error_msg)\n\n        if not self.user_id:\n            msg = \"User ID is required for fetching Knowledge Base data.\"\n            raise ValueError(msg)\n\n        # Saved flows can load in standalone lfx, but knowledge retrieval\n        # requires the application's user database and storage services.\n        try:\n            from langflow.services.database.models.user.crud import get_user_by_id\n        except ModuleNotFoundError as exc:\n            if exc.name != \"langflow\":\n                raise\n            msg = \"Knowledge Base retrieval requires Langflow's application services. Run this flow in Langflow.\"\n            raise RuntimeError(msg) from None\n\n        async with session_scope() as db:\n            current_user = await get_user_by_id(db, self.user_id)\n            if not current_user:\n                msg = f\"User with ID {self.user_id} not found.\"\n                raise ValueError(msg)\n\n        metadata = await self._get_kb_metadata()\n        if not metadata:\n            msg = f\"Metadata not found for knowledge base: {self.knowledge_base}. Ensure it has been indexed.\"\n            raise ValueError(msg)\n\n        model_selection = self._resolve_model_selection(metadata)\n        chunk_size = metadata.get(\"chunk_size\")\n\n        # Resolve where this KB lives before instantiating embeddings: path\n        # containment is a cheap local check, and a request that will be refused\n        # shouldn't first pay for a credential lookup and a provider client.\n        backend_type, _backend_config = await self._resolve_backend_config()\n\n        # ``api_key=None`` is not \"no credential\": ``get_embeddings`` resolves the\n        # provider's key from the user's variables table and then the environment,\n        # exactly as the server-side ingestion path does. The component input just\n        # takes precedence when supplied.\n        embedding_function = get_embeddings(\n            model=model_selection,\n            user_id=self.user_id,\n            api_key=self.api_key if getattr(self, \"api_key\", None) else None,\n            chunk_size=chunk_size,\n        )\n\n        from langflow.api.utils.kb_helpers import backend_for_name\n\n        owner_id = self._user_uuid\n        if owner_id is None:\n            msg = \"User ID is required for knowledge base storage.\"\n            raise ValueError(msg)\n        backend = await backend_for_name(owner_id, self.knowledge_base, embedding_function=embedding_function)\n        try:\n            user_metadata_filter = _parse_metadata_filter(getattr(self, \"metadata_filter\", None))\n            use_scores = bool(self.search_query)\n            exact_filter = backend_type == BackendType.SQLITE.value\n            search_k = self.top_k if exact_filter or not user_metadata_filter else self.top_k * 4\n            search_options = {\"source_filter\": user_metadata_filter} if exact_filter else {}\n            results = await backend.similarity_search(\n                query=self.search_query or \"\",\n                k=search_k,\n                with_scores=use_scores,\n                **search_options,\n            )\n            if user_metadata_filter and not exact_filter:\n                results = [\n                    (doc, score) for doc, score in results if _chunk_matches_filter(doc.metadata, user_metadata_filter)\n                ][: self.top_k]\n\n            embeddings_by_key: dict[tuple[str, str], list[float]] = {}\n            if self.include_embeddings and results:\n                # Join each retrieved chunk to its stored embedding. We key on\n                # ``_id`` when present and fall back to page content otherwise,\n                # so KBs populated by direct file upload \u2014 whose chunks carry no\n                # ``_id`` \u2014 still resolve. See ``_embedding_match_key``.\n                wanted_keys = {_embedding_match_key(doc.page_content, doc.metadata) for doc, _score in results}\n                async with aclosing(backend.iter_documents(include_embeddings=True)) as batches:\n                    async for batch in batches:\n                        for entry in batch:\n                            if entry.embedding is None:\n                                continue\n                            key = _embedding_match_key(entry.content, entry.metadata)\n                            if key in wanted_keys:\n                                embeddings_by_key[key] = entry.embedding\n                        if wanted_keys <= embeddings_by_key.keys():\n                            break\n\n            data_list: list[Data] = []\n            for doc, score in results:\n                kwargs: dict[str, Any] = {\"content\": doc.page_content}\n                if use_scores:\n                    kwargs[\"_score\"] = backend.normalize_score(score)\n                if self.include_metadata:\n                    kwargs.update(doc.metadata)\n                if self.include_embeddings:\n                    kwargs[\"_embeddings\"] = embeddings_by_key.get(_embedding_match_key(doc.page_content, doc.metadata))\n                data_list.append(Data(**kwargs))\n\n            return DataFrame(data=data_list)\n        finally:\n            await backend.teardown()\n\n\ndef _embedding_match_key(content: str, metadata: dict[str, Any] | None) -> tuple[str, str]:\n    \"\"\"Build a stable key for aligning a retrieved chunk with its stored embedding.\n\n    Embeddings are gathered separately (via ``iter_documents``) and then joined\n    back onto the search results. Component-driven ingestion stamps a\n    content-hash ``_id`` on every chunk, but direct file-upload ingestion\n    (``KBIngestionHelper.perform_ingestion``) does not. Keying the join purely on\n    ``_id`` therefore left every upload-populated KB with ``_embeddings: None``.\n\n    We prefer ``_id`` when present (so legitimately distinct chunks that happen to\n    share text aren't collapsed) and fall back to the chunk's page content\n    otherwise. The content fallback is exact for the embedding use case: two\n    chunks with identical text necessarily share the same embedding vector. The\n    leading ``\"id\"`` / ``\"content\"`` tag namespaces the two key spaces so a mixed\n    KB (some chunks with ``_id``, some without) never cross-matches.\n    \"\"\"\n    doc_id = metadata.get(\"_id\") if metadata else None\n    if doc_id:\n        return (\"id\", str(doc_id))\n    return (\"content\", content or \"\")\n\n\ndef _parse_metadata_filter(raw: str | None) -> dict[str, list[str]]:\n    \"\"\"Decode the ``metadata_filter`` input into a {key: [values]} map.\n\n    Empty or malformed input maps to an empty filter so retrieval falls back\n    to the unfiltered path. JSON errors are swallowed here rather than raised:\n    surfacing component-config errors at the canvas node would break a flow\n    run for what is meant to be an optional refinement.\n    \"\"\"\n    if not raw:\n        return {}\n    text = raw.strip() if isinstance(raw, str) else raw\n    if not text:\n        return {}\n    try:\n        decoded = json.loads(text)\n    except (TypeError, json.JSONDecodeError):\n        logger.warning(\"KnowledgeComponent: metadata_filter is not valid JSON; ignoring filter.\")\n        return {}\n    if not isinstance(decoded, dict):\n        logger.warning(\"KnowledgeComponent: metadata_filter must be a JSON object; ignoring filter.\")\n        return {}\n    result: dict[str, list[str]] = {}\n    for key, value in decoded.items():\n        if not isinstance(key, str):\n            continue\n        if isinstance(value, list):\n            result[key] = [str(entry) for entry in value]\n        else:\n            result[key] = [str(value)]\n    return result\n\n\ndef _chunk_matches_filter(metadata: dict[str, Any] | None, filt: dict[str, list[str]]) -> bool:\n    \"\"\"AND across keys, OR within key values, mirroring the chunks endpoint.\"\"\"\n    if not filt:\n        return True\n    if not metadata:\n        return False\n    raw = metadata.get(\"source_metadata\")\n    if not raw:\n        return False\n    try:\n        stored = json.loads(raw) if isinstance(raw, str) else raw\n    except json.JSONDecodeError:\n        return False\n    if not isinstance(stored, dict):\n        return False\n    for key, expected_values in filt.items():\n        actual = stored.get(key)\n        if actual is None:\n            return False\n        actual_set = {str(entry) for entry in actual} if isinstance(actual, list) else {str(actual)}\n        if not actual_set & set(expected_values):\n            return False\n    return True\n"
              },
              "column_config": {
                "_input_type": "TableInput",
                "advanced": false,
                "display_name": "Column Configuration",
                "dynamic": true,
                "info": "Configure column behavior for the knowledge base.",
                "input_types": [
                  "DataFrame",
                  "Table"
                ],
                "is_list": true,
                "list_add_label": "Add More",
                "name": "column_config",
                "override_skip": false,
                "placeholder": "",
                "required": true,
                "show": false,
                "table_icon": "Table",
                "table_schema": [
                  {
                    "description": "Name of the column in the source DataFrame",
                    "display_name": "Column Name",
                    "edit_mode": "inline",
                    "name": "column_name",
                    "type": "str"
                  },
                  {
                    "default": false,
                    "description": "Create embeddings for this column",
                    "display_name": "Vectorize",
                    "edit_mode": "inline",
                    "name": "vectorize",
                    "type": "boolean"
                  },
                  {
                    "default": false,
                    "description": "Use this column as unique identifier",
                    "display_name": "Identifier",
                    "edit_mode": "inline",
                    "name": "identifier",
                    "type": "boolean"
                  }
                ],
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "trigger_icon": "Table",
                "trigger_text": "Open table",
                "type": "table",
                "value": [
                  {
                    "column_name": "text",
                    "identifier": true,
                    "vectorize": true
                  }
                ]
              },
              "include_embeddings": {
                "_input_type": "BoolInput",
                "advanced": true,
                "display_name": "Include Embeddings",
                "dynamic": false,
                "info": "Whether to include embeddings in the output. Only applicable if 'Include Metadata' is enabled.",
                "list": false,
                "list_add_label": "Add More",
                "name": "include_embeddings",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "type": "bool",
                "value": false
              },
              "include_metadata": {
                "_input_type": "BoolInput",
                "advanced": false,
                "display_name": "Include Metadata",
                "dynamic": false,
                "info": "Whether to include all metadata in the output. If false, only content is returned.",
                "list": false,
                "list_add_label": "Add More",
                "name": "include_metadata",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "type": "bool",
                "value": true
              },
              "input_df": {
                "_input_type": "HandleInput",
                "advanced": false,
                "display_name": "Input",
                "dynamic": true,
                "info": "Table with all original columns (already chunked / processed). Accepts Message, Data, or DataFrame. If Message or Data is provided, it is converted to a DataFrame automatically.",
                "input_types": [
                  "Message",
                  "Data",
                  "JSON",
                  "DataFrame",
                  "Table"
                ],
                "list": false,
                "list_add_label": "Add More",
                "name": "input_df",
                "override_skip": false,
                "placeholder": "",
                "required": true,
                "show": false,
                "title_case": false,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "other",
                "value": ""
              },
              "knowledge_base": {
                "_input_type": "DropdownInput",
                "advanced": false,
                "combobox": false,
                "dialog_inputs": {},
                "display_name": "Knowledge",
                "dynamic": false,
                "info": "Select the knowledge to load data from.",
                "name": "knowledge_base",
                "options": [],
                "options_metadata": [],
                "placeholder": "",
                "real_time_refresh": true,
                "refresh_button": true,
                "required": true,
                "show": true,
                "title_case": false,
                "toggle": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "type": "str",
                "value": ""
              },
              "metadata_filter": {
                "_input_type": "MessageTextInput",
                "advanced": true,
                "display_name": "Metadata Filter",
                "dynamic": false,
                "info": "Optional JSON object of user-metadata key/value pairs. Only chunks whose source_metadata matches every key are returned (e.g. {\"tag\": \"invoice\"} or {\"tag\": [\"invoice\", \"audit\"]} for OR-of-values). Backends without native filtering apply the match client-side after retrieval.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "name": "metadata_filter",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "str",
                "value": ""
              },
              "metadata_json": {
                "_input_type": "StrInput",
                "advanced": true,
                "display_name": "Metadata",
                "dynamic": true,
                "info": "Optional JSON object of user metadata applied to every chunk produced by this run (e.g. {\"tag\": \"invoice\", \"year\": \"2026\"}). Same shape as the upload modal Metadata section so chunks browser filters + Knowledge retrieval metadata_filter work uniformly across upload, folder, and flow-driven ingestion. Malformed JSON is ignored with a warning rather than failing the run.",
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "name": "metadata_json",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": false,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "str",
                "value": ""
              },
              "mode": {
                "_input_type": "TabInput",
                "advanced": false,
                "display_name": "Mode",
                "dynamic": false,
                "info": "Switch between writing new data into the knowledge base and querying it.",
                "name": "mode",
                "options": [
                  "Ingest",
                  "Retrieve"
                ],
                "override_skip": false,
                "placeholder": "",
                "real_time_refresh": true,
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": true,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "tab",
                "value": "Retrieve"
              },
              "search_query": {
                "_input_type": "MessageTextInput",
                "advanced": false,
                "display_name": "Search Query",
                "dynamic": false,
                "info": "Optional search query to filter knowledge base data.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "name": "search_query",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": true,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "type": "str",
                "value": ""
              },
              "top_k": {
                "_input_type": "IntInput",
                "advanced": true,
                "display_name": "Top K Results",
                "dynamic": false,
                "info": "Number of top results to return from the knowledge base.",
                "list": false,
                "list_add_label": "Add More",
                "name": "top_k",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "type": "int",
                "value": 5
              }
            },
            "tool_mode": false
          },
          "selected_output": "retrieve_data",
          "showNode": true,
          "type": "Knowledge"
        },
        "dragging": false,
        "id": "KnowledgeRetrieve-docqa",
        "measured": {
          "height": 426,
          "width": 320
        },
        "position": {
          "x": 647.8108581436077,
          "y": 125.9719789842382
        },
        "positionAbsolute": {
          "x": 700,
          "y": 520
        },
        "selected": false,
        "type": "genericNode"
      },
      {
        "data": {
          "id": "Parser-docqa",
          "node": {
            "base_classes": [
              "Message"
            ],
            "beta": false,
            "category": "processing",
            "conditional_paths": [],
            "custom_fields": {},
            "description": "Extracts text using a template.",
            "display_name": "Parser",
            "documentation": "",
            "edited": false,
            "field_order": [
              "input_data",
              "mode",
              "pattern",
              "sep"
            ],
            "frozen": false,
            "icon": "braces",
            "key": "parser",
            "legacy": false,
            "lf_version": "1.10.0",
            "metadata": {
              "code_hash": "7ab5fd0e0a10",
              "component_class_name": "ParserComponent",
              "dependencies": {
                "dependencies": [
                  {
                    "name": "lfx",
                    "version": null
                  }
                ],
                "total_dependencies": 1
              },
              "module": "lfx.components.processing.parser.ParserComponent"
            },
            "minimized": false,
            "output_types": [],
            "outputs": [
              {
                "allows_loop": false,
                "cache": true,
                "display_name": "Parsed Text",
                "group_outputs": false,
                "method": "parse_combined_text",
                "name": "parsed_text",
                "selected": "Message",
                "tool_mode": true,
                "types": [
                  "Message"
                ],
                "value": "__UNDEFINED__"
              }
            ],
            "pinned": false,
            "score": 2.220446049250313e-16,
            "template": {
              "_type": "Component",
              "code": {
                "advanced": true,
                "api_editable": false,
                "dynamic": true,
                "fileTypes": [],
                "file_path": "",
                "info": "",
                "list": false,
                "load_from_db": false,
                "multiline": true,
                "name": "code",
                "password": false,
                "placeholder": "",
                "required": true,
                "show": true,
                "title_case": false,
                "type": "code",
                "value": "from lfx.custom.custom_component.component import Component\nfrom lfx.helpers.data import safe_convert\nfrom lfx.inputs.inputs import BoolInput, HandleInput, MessageTextInput, MultilineInput, TabInput\nfrom lfx.schema.data import Data\nfrom lfx.schema.dataframe import DataFrame\nfrom lfx.schema.message import Message\nfrom lfx.template.field.base import Output\n\n\nclass _DefaultDict(dict):\n    def __init__(self, data_item: Data):\n        super().__init__(data_item.data)\n        self.default_value = data_item.default_value\n\n    def __missing__(self, key):\n        return self.default_value if self.default_value is not None else \"\"\n\n\nclass ParserComponent(Component):\n    display_name = \"Parser\"\n    description = \"Extracts text using a template.\"\n    documentation: str = \"https://docs.langflow.org/parser\"\n    icon = \"braces\"\n\n    inputs = [\n        HandleInput(\n            name=\"input_data\",\n            display_name=\"JSON or Table\",\n            input_types=[\"DataFrame\", \"Table\", \"Data\", \"JSON\"],\n            info=\"Accepts a DataFrame, a Data object, or a list of Data objects.\",\n            required=True,\n        ),\n        TabInput(\n            name=\"mode\",\n            display_name=\"Mode\",\n            options=[\"Parser\", \"Stringify\"],\n            value=\"Parser\",\n            info=\"Convert into raw string instead of using a template.\",\n            real_time_refresh=True,\n        ),\n        MultilineInput(\n            name=\"pattern\",\n            display_name=\"Template\",\n            info=(\n                \"Use variables within curly brackets to extract column values for DataFrames \"\n                \"or key values for Data.\"\n                \"For example: `Name: {Name}, Age: {Age}, Country: {Country}`\"\n            ),\n            value=\"Text: {text}\",  # Example default\n            dynamic=True,\n            show=True,\n            required=True,\n        ),\n        MessageTextInput(\n            name=\"sep\",\n            display_name=\"Separator\",\n            advanced=True,\n            value=\"\\n\",\n            info=\"String used to separate rows/items.\",\n        ),\n    ]\n\n    outputs = [\n        Output(\n            display_name=\"Parsed Text\",\n            name=\"parsed_text\",\n            info=\"Formatted text output.\",\n            method=\"parse_combined_text\",\n        ),\n    ]\n\n    def update_build_config(self, build_config, field_value, field_name=None):\n        \"\"\"Dynamically hide/show `template` and enforce requirement based on `stringify`.\"\"\"\n        if field_name == \"mode\":\n            build_config[\"pattern\"][\"show\"] = self.mode == \"Parser\"\n            build_config[\"pattern\"][\"required\"] = self.mode == \"Parser\"\n            if field_value:\n                clean_data = BoolInput(\n                    name=\"clean_data\",\n                    display_name=\"Clean Data\",\n                    info=(\n                        \"Enable to clean the data by removing empty rows and lines \"\n                        \"in each cell of the DataFrame/ Data object.\"\n                    ),\n                    value=True,\n                    advanced=True,\n                    required=False,\n                )\n                build_config[\"clean_data\"] = clean_data.to_dict()\n            else:\n                build_config.pop(\"clean_data\", None)\n\n        return build_config\n\n    def _clean_args(self) -> tuple[DataFrame | None, Data | list[Data] | None]:\n        \"\"\"Prepare arguments based on input type.\"\"\"\n        input_data = self.input_data\n\n        match input_data:\n            case list() if all(isinstance(item, Data) for item in input_data):\n                return None, input_data\n            case DataFrame():\n                return input_data, None\n            case Data():\n                return None, input_data\n            case dict() if \"data\" in input_data:\n                try:\n                    if \"columns\" in input_data:  # Likely a DataFrame\n                        return DataFrame.from_dict(input_data), None\n                    # Likely a Data object\n                    return None, Data(**input_data)\n                except (TypeError, ValueError, KeyError) as e:\n                    msg = f\"Invalid structured input provided: {e!s}\"\n                    raise ValueError(msg) from e\n            case _:\n                msg = f\"Unsupported input type: {type(input_data)}. Expected DataFrame, Data, or list[Data].\"\n                raise ValueError(msg)\n\n    def parse_combined_text(self) -> Message:\n        \"\"\"Parse all rows/items into a single text or convert input to string if `stringify` is enabled.\"\"\"\n        # Early return for stringify option\n        if self.mode == \"Stringify\":\n            return self.convert_to_string()\n\n        df, data = self._clean_args()\n\n        lines = []\n        if df is not None:\n            for _, row in df.iterrows():\n                formatted_text = self.pattern.format(**row.to_dict())\n                lines.append(formatted_text)\n        elif data is not None:\n            data_items = data if isinstance(data, list) else [data]\n            for data_item in data_items:\n                formatted_text = self.pattern.format_map(_DefaultDict(data_item))\n                lines.append(formatted_text)\n\n        combined_text = self.sep.join(lines)\n        self.status = combined_text\n        return Message(text=combined_text)\n\n    def convert_to_string(self) -> Message:\n        \"\"\"Convert input data to string with proper error handling.\"\"\"\n        clean_data = getattr(self, \"clean_data\", False)\n        if isinstance(self.input_data, list):\n            result = \"\\n\".join(safe_convert(item, clean_data=clean_data) for item in self.input_data)\n        else:\n            result = safe_convert(self.input_data, clean_data=clean_data)\n        self.log(f\"Converted to string with length: {len(result)}\")\n\n        message = Message(text=result)\n        self.status = message\n        return message\n"
              },
              "input_data": {
                "_input_type": "HandleInput",
                "advanced": false,
                "display_name": "JSON or Table",
                "dynamic": false,
                "info": "Accepts a DataFrame, a Data object, or a list of Data objects.",
                "input_types": [
                  "DataFrame",
                  "Table",
                  "Data",
                  "JSON"
                ],
                "list": false,
                "list_add_label": "Add More",
                "name": "input_data",
                "placeholder": "",
                "required": true,
                "show": true,
                "title_case": false,
                "trace_as_metadata": true,
                "type": "other",
                "value": ""
              },
              "mode": {
                "_input_type": "TabInput",
                "advanced": false,
                "display_name": "Mode",
                "dynamic": false,
                "info": "Convert into raw string instead of using a template.",
                "name": "mode",
                "options": [
                  "Parser",
                  "Stringify"
                ],
                "placeholder": "",
                "real_time_refresh": true,
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "type": "tab",
                "value": "Parser"
              },
              "pattern": {
                "_input_type": "MultilineInput",
                "advanced": false,
                "copy_field": false,
                "display_name": "Template",
                "dynamic": true,
                "info": "Use variables within curly brackets to extract column values for DataFrames or key values for Data.For example: `Name: {Name}, Age: {Age}, Country: {Country}`",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "multiline": true,
                "name": "pattern",
                "placeholder": "",
                "required": true,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "type": "str",
                "value": "Text: {content}"
              },
              "sep": {
                "_input_type": "MessageTextInput",
                "advanced": true,
                "display_name": "Separator",
                "dynamic": false,
                "info": "String used to separate rows/items.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "name": "sep",
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "type": "str",
                "value": "\n"
              }
            },
            "tool_mode": false
          },
          "selected_output": "parsed_text",
          "showNode": true,
          "type": "parser"
        },
        "dragging": false,
        "id": "Parser-docqa",
        "measured": {
          "height": 330,
          "width": 320
        },
        "position": {
          "x": 1137.3905429071804,
          "y": 303.415061295972
        },
        "positionAbsolute": {
          "x": 1140,
          "y": 520
        },
        "selected": false,
        "type": "genericNode"
      },
      {
        "data": {
          "_connectionMode": false,
          "id": "Agent-docqa",
          "node": {
            "base_classes": [
              "Data",
              "JSON",
              "Message"
            ],
            "beta": false,
            "conditional_paths": [],
            "custom_fields": {},
            "description": "Define the agent's instructions, then enter a task to complete using tools.",
            "display_name": "Agent",
            "documentation": "https://docs.langflow.org/agents",
            "edited": false,
            "field_order": [
              "model",
              "api_key",
              "base_url_ibm_watsonx",
              "project_id",
              "system_prompt",
              "context_id",
              "n_messages",
              "max_tokens",
              "format_instructions",
              "output_schema",
              "tools",
              "input_value",
              "handle_parsing_errors",
              "max_iterations",
              "stream",
              "add_current_date_tool",
              "add_calculator_tool"
            ],
            "frozen": false,
            "icon": "bot",
            "last_updated": "2026-06-25T19:40:06.492Z",
            "legacy": false,
            "lf_version": "1.10.0",
            "metadata": {
              "code_hash": "9fcfe75813e7",
              "component_class_name": "AgentComponent",
              "dependencies": {
                "dependencies": [
                  {
                    "name": "langchain",
                    "version": "1.3.14"
                  },
                  {
                    "name": "langgraph",
                    "version": "1.2.19"
                  },
                  {
                    "name": "lfx",
                    "version": null
                  },
                  {
                    "name": "langchain_core",
                    "version": "1.5.1"
                  }
                ],
                "total_dependencies": 4
              },
              "module": "lfx.components.models_and_agents.agent.AgentComponent"
            },
            "minimized": false,
            "output_types": [],
            "outputs": [
              {
                "allows_loop": false,
                "cache": true,
                "display_name": "Response",
                "group_outputs": false,
                "method": "message_response",
                "name": "response",
                "selected": "Message",
                "tool_mode": true,
                "types": [
                  "Message"
                ],
                "value": "__UNDEFINED__"
              },
              {
                "allows_loop": false,
                "cache": true,
                "display_name": "Structured Response",
                "group_outputs": false,
                "method": "json_response",
                "name": "structured_response",
                "selected": "JSON",
                "tool_mode": true,
                "types": [
                  "Data",
                  "JSON"
                ],
                "value": "__UNDEFINED__"
              }
            ],
            "pinned": false,
            "template": {
              "_type": "Component",
              "add_calculator_tool": {
                "_input_type": "BoolInput",
                "advanced": true,
                "display_name": "Calculator",
                "dynamic": false,
                "info": "If true, adds a zero-config arithmetic calculator tool to the agent (safe: only +, -, *, /, ** operators via AST).",
                "input_types": [],
                "list": false,
                "list_add_label": "Add More",
                "name": "add_calculator_tool",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "bool",
                "value": true
              },
              "add_current_date_tool": {
                "_input_type": "BoolInput",
                "advanced": true,
                "display_name": "Current Date",
                "dynamic": false,
                "info": "If true, will add a tool to the agent that returns the current date.",
                "input_types": [],
                "list": false,
                "list_add_label": "Add More",
                "name": "add_current_date_tool",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "bool",
                "value": false
              },
              "api_key": {
                "_input_type": "SecretStrInput",
                "advanced": true,
                "display_name": "API Key",
                "dynamic": false,
                "info": "Overrides global provider settings. Leave blank to use your pre-configured API Key.",
                "input_types": [],
                "load_from_db": false,
                "name": "api_key",
                "override_skip": false,
                "password": true,
                "placeholder": "",
                "real_time_refresh": true,
                "required": false,
                "show": true,
                "title_case": false,
                "track_in_telemetry": false,
                "type": "str",
                "value": ""
              },
              "base_url_ibm_watsonx": {
                "_input_type": "DropdownInput",
                "advanced": false,
                "combobox": true,
                "dialog_inputs": {},
                "display_name": "watsonx API Endpoint",
                "dynamic": false,
                "external_options": {},
                "info": "The base URL of the API (IBM watsonx.ai only)",
                "input_types": [],
                "name": "base_url_ibm_watsonx",
                "options": [
                  "https://us-south.ml.cloud.ibm.com",
                  "https://eu-de.ml.cloud.ibm.com",
                  "https://eu-gb.ml.cloud.ibm.com",
                  "https://au-syd.ml.cloud.ibm.com",
                  "https://jp-tok.ml.cloud.ibm.com",
                  "https://ca-tor.ml.cloud.ibm.com"
                ],
                "options_metadata": [],
                "override_skip": false,
                "placeholder": "",
                "real_time_refresh": true,
                "required": false,
                "show": false,
                "title_case": false,
                "toggle": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "str",
                "value": "https://us-south.ml.cloud.ibm.com"
              },
              "code": {
                "advanced": true,
                "api_editable": false,
                "dynamic": true,
                "fileTypes": [],
                "file_path": "",
                "info": "",
                "list": false,
                "load_from_db": false,
                "multiline": true,
                "name": "code",
                "password": false,
                "placeholder": "",
                "required": true,
                "show": true,
                "title_case": false,
                "type": "code",
                "value": "from __future__ import annotations\n\nimport uuid\nfrom collections.abc import Mapping\nfrom contextlib import contextmanager\nfrom datetime import datetime, timezone\nfrom typing import TYPE_CHECKING, Any, cast\n\nfrom langchain.agents import create_agent\nfrom langchain.agents.middleware import (\n    HumanInTheLoopMiddleware,\n    ModelCallLimitMiddleware,\n    ToolRetryMiddleware,\n)\nfrom langgraph.types import Command\n\nfrom lfx.components.models_and_agents.agent_helpers.graph_event_adapter import (\n    adapt_graph_events_to_executor_shape,\n)\nfrom lfx.components.models_and_agents.agent_helpers.messages_input_builder import (\n    build_initial_messages,\n)\nfrom lfx.components.models_and_agents.agent_helpers.placeholder_corrective_middleware import (\n    WatsonXPlaceholderMiddleware,\n)\nfrom lfx.components.models_and_agents.agent_helpers.single_tool_call_middleware import (\n    SingleToolCallMiddleware,\n)\nfrom lfx.components.models_and_agents.agent_helpers.tool_approval import ToolApprovalMixin\nfrom lfx.components.models_and_agents.agent_helpers.tool_call_id_middleware import ToolCallIDMiddleware\nfrom lfx.components.models_and_agents.memory import MemoryComponent, _safe_graph_user_id, aget_agent_chat_history\n\nif TYPE_CHECKING:\n    from collections.abc import Awaitable, Callable\n\n    from langchain_core.tools import Tool\n\n    from lfx.schema.log import OnTokenFunctionType, SendMessageFunctionType\n\nfrom lfx.base.agents.agent import LCToolsAgentComponent\nfrom lfx.base.agents.callback import AgentAsyncHandler\nfrom lfx.base.agents.default_system_prompt import DEFAULT_SYSTEM_PROMPT_TEMPLATE\nfrom lfx.base.agents.events import AgentPausedError, ExceptionWithMessageError, process_agent_events\nfrom lfx.base.agents.token_callback import TokenUsageCallbackHandler\nfrom lfx.base.agents.utils import get_chat_output_sender_name\nfrom lfx.base.constants import STREAM_INFO_TEXT\nfrom lfx.base.models.unified_models import (\n    get_language_model_options,\n    get_llm,\n    handle_model_input_update,\n)\nfrom lfx.base.models.watsonx_constants import IBM_WATSONX_URLS\nfrom lfx.components.agentics.helpers.model_config import validate_model_selection\nfrom lfx.components.helpers import CalculatorComponent, CurrentDateComponent\nfrom lfx.components.langchain_utilities.ibm_granite_handler import is_watsonx_model\nfrom lfx.components.langchain_utilities.tool_calling import ToolCallingAgentComponent\nfrom lfx.custom.custom_component.component import get_component_toolkit\nfrom lfx.field_typing.range_spec import RangeSpec\nfrom lfx.inputs.inputs import BoolInput, DropdownInput, ModelInput, StrInput\nfrom lfx.io import IntInput, MessageTextInput, MultilineInput, Output, SecretStrInput, TableInput\nfrom lfx.log.logger import logger\nfrom lfx.memory import delete_message\nfrom lfx.schema.data import Data\nfrom lfx.schema.dotdict import dotdict\nfrom lfx.schema.message import Message\nfrom lfx.schema.table import EditMode\nfrom lfx.utils.constants import MESSAGE_SENDER_AI\n\n\ndef set_advanced_true(component_input):\n    component_input.advanced = True\n    return component_input\n\n\ndef _agent_base_inputs():\n    \"\"\"Return base inputs tailored to AgentComponent's create_agent path.\n\n    `get_base_inputs()` returns a shared list \u2014 replace, don't mutate. We drop\n    inputs that are no-ops here and override info text on the inputs whose\n    semantics shifted under create_agent.\n\n    `verbose` is dropped because the create_agent event stream already surfaces\n    every agent step via the \"Agent Steps\" content blocks; the legacy boolean\n    has nothing to toggle. Saved flows that still carry a `verbose` value just\n    ignore it on load (the schema no longer declares it).\n    \"\"\"\n    drop = {\"verbose\"}\n    overrides = {\n        \"handle_parsing_errors\": BoolInput(\n            name=\"handle_parsing_errors\",\n            display_name=\"Handle Parse Errors\",\n            value=True,\n            advanced=True,\n            info=(\n                \"Adds tool-execution retry as a safety net. `create_agent` already \"\n                \"feeds tool-call validation errors back to the LLM automatically; \"\n                \"this flag layers `ToolRetryMiddleware` on top so transient tool \"\n                \"runtime failures are retried (max 2 retries).\"\n            ),\n        ),\n        \"max_iterations\": IntInput(\n            name=\"max_iterations\",\n            display_name=\"Max Iterations\",\n            value=15,\n            advanced=True,\n            range_spec=RangeSpec(min=1, max=128000, step=1, step_type=\"int\"),\n            info=(\n                \"Maximum number of model calls the agent can make before stopping \"\n                \"(maps to `ModelCallLimitMiddleware.run_limit` on the create_agent \"\n                \"path). Must be at least 1 \u2014 it is a safety cap, never 'unlimited'.\"\n            ),\n        ),\n    }\n    return [overrides.get(inp.name, inp) for inp in LCToolsAgentComponent.get_base_inputs() if inp.name not in drop]\n\n\ndef _extract_text_content(value) -> str:\n    \"\"\"Pull a string payload from a Message-like, AIMessage-like, or string value.\"\"\"\n    if isinstance(value, str):\n        return value\n    text = getattr(value, \"text\", None)\n    if isinstance(text, str):\n        return text\n    content = getattr(value, \"content\", None)\n    if isinstance(content, str):\n        return content\n    return str(value) if value is not None else \"\"\n\n\n@contextmanager\ndef _suppress_send_message(component: Any):\n    \"\"\"Temporarily replace component.send_message with a no-op for the duration of the block.\n\n    Used during the structured-output prompt fallback: run_agent streams the agent's\n    final answer through self.send_message (correct for message_response), but in\n    json_response the orchestrator parses that text into structured Data which the\n    downstream Chat Output emits \u2014 leaving the original emission in place produces a\n    duplicate message in the playground. The original method is always restored on exit,\n    even when the wrapped call raises.\n    \"\"\"\n    original = component.send_message\n\n    async def _noop(message, *_args, **_kwargs):\n        return message\n\n    component.send_message = _noop\n    try:\n        yield\n    finally:\n        component.send_message = original\n\n\nclass AgentComponent(ToolApprovalMixin, ToolCallingAgentComponent):\n    display_name: str = \"Agent\"\n    description: str = \"Define the agent's instructions, then enter a task to complete using tools.\"\n    documentation: str = \"https://docs.langflow.org/agents\"\n    icon = \"bot\"\n    beta = False\n    name = \"Agent\"\n\n    memory_inputs = [set_advanced_true(component_input) for component_input in MemoryComponent().inputs]\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            # Agents require tool calling \u2014 the filter is honored by\n            # ``handle_model_input_update`` so models that can't run with\n            # tools never reach the picker (and any saved selection that\n            # no longer satisfies the constraint is auto-replaced).\n            filters={\"tool_calling\": True},\n        ),\n        SecretStrInput(\n            name=\"api_key\",\n            display_name=\"API Key\",\n            info=\"Overrides global provider settings. Leave blank to use your pre-configured API Key.\",\n            real_time_refresh=True,\n            advanced=True,\n        ),\n        DropdownInput(\n            name=\"base_url_ibm_watsonx\",\n            display_name=\"watsonx API Endpoint\",\n            info=\"The base URL of the API (IBM watsonx.ai only)\",\n            options=IBM_WATSONX_URLS,\n            value=IBM_WATSONX_URLS[0],\n            combobox=True,\n            show=False,\n            real_time_refresh=True,\n        ),\n        StrInput(\n            name=\"project_id\",\n            display_name=\"watsonx Project ID\",\n            info=\"The project ID associated with the foundation model (IBM watsonx.ai only)\",\n            show=False,\n            required=False,\n        ),\n        MultilineInput(\n            name=\"system_prompt\",\n            display_name=\"Agent Instructions\",\n            info=(\n                \"System Prompt: Initial instructions and context provided to guide the agent's behavior. \"\n                \"Supports dynamic placeholders: {current_date}, {model_name}, {optional_user_context}.\"\n            ),\n            value=DEFAULT_SYSTEM_PROMPT_TEMPLATE,\n            advanced=False,\n        ),\n        MessageTextInput(\n            name=\"context_id\",\n            display_name=\"Context ID\",\n            info=\"The context ID of the chat. Adds an extra layer to the local memory.\",\n            value=\"\",\n            advanced=True,\n        ),\n        IntInput(\n            name=\"n_messages\",\n            display_name=\"Number of Chat History Messages\",\n            value=100,\n            info=\"Number of chat history messages to retrieve.\",\n            advanced=True,\n            show=True,\n        ),\n        IntInput(\n            name=\"max_tokens\",\n            display_name=\"Max Tokens\",\n            info=\"Maximum number of tokens to generate. Set to 0 for no explicit limit. Field name varies by provider.\",\n            advanced=True,\n            range_spec=RangeSpec(min=0, max=128000, step=1, step_type=\"int\"),\n        ),\n        MultilineInput(\n            name=\"format_instructions\",\n            display_name=\"Output Format Instructions\",\n            info=\"Generic Template for structured output formatting. Valid only with Structured response.\",\n            value=(\n                \"You are an AI that extracts structured JSON objects from unstructured text. \"\n                \"Use a predefined schema with expected types (str, int, float, bool, dict). \"\n                \"Extract ALL relevant instances that match the schema - if multiple patterns exist, capture them all. \"\n                \"Fill missing or ambiguous values with defaults: null for missing values. \"\n                \"Remove exact duplicates but keep variations that have different field values. \"\n                \"Always return valid JSON in the expected format, never throw errors. \"\n                \"If multiple objects can be extracted, return them all in the structured format.\"\n            ),\n            advanced=True,\n        ),\n        TableInput(\n            name=\"output_schema\",\n            display_name=\"Output Schema\",\n            info=(\n                \"Schema Validation: Define the structure and data types for structured output. \"\n                \"No validation if no output schema.\"\n            ),\n            advanced=True,\n            required=False,\n            value=[],\n            table_schema=[\n                {\n                    \"name\": \"name\",\n                    \"display_name\": \"Name\",\n                    \"type\": \"str\",\n                    \"description\": \"Specify the name of the output field.\",\n                    \"default\": \"field\",\n                    \"edit_mode\": EditMode.INLINE,\n                },\n                {\n                    \"name\": \"description\",\n                    \"display_name\": \"Description\",\n                    \"type\": \"str\",\n                    \"description\": \"Describe the purpose of the output field.\",\n                    \"default\": \"description of field\",\n                    \"edit_mode\": EditMode.POPOVER,\n                },\n                {\n                    \"name\": \"type\",\n                    \"display_name\": \"Type\",\n                    \"type\": \"str\",\n                    \"edit_mode\": EditMode.INLINE,\n                    \"description\": (\"Indicate the data type of the output field (e.g., str, int, float, bool, dict).\"),\n                    \"options\": [\"str\", \"int\", \"float\", \"bool\", \"dict\"],\n                    \"default\": \"str\",\n                },\n                {\n                    \"name\": \"multiple\",\n                    \"display_name\": \"As List\",\n                    \"type\": \"boolean\",\n                    \"description\": \"Set to True if this output field should be a list of the specified type.\",\n                    \"default\": \"False\",\n                    \"edit_mode\": EditMode.INLINE,\n                },\n            ],\n        ),\n        *_agent_base_inputs(),\n        # removed memory inputs from agent component\n        # *memory_inputs,\n        BoolInput(\n            name=\"stream\",\n            display_name=\"Stream\",\n            info=STREAM_INFO_TEXT,\n            value=True,\n            advanced=True,\n        ),\n        BoolInput(\n            name=\"add_current_date_tool\",\n            display_name=\"Current Date\",\n            advanced=True,\n            info=\"If true, will add a tool to the agent that returns the current date.\",\n            value=True,\n        ),\n        BoolInput(\n            name=\"add_calculator_tool\",\n            display_name=\"Calculator\",\n            advanced=True,\n            info=(\n                \"If true, adds a zero-config arithmetic calculator tool to the agent \"\n                \"(safe: only +, -, *, /, ** operators via AST).\"\n            ),\n            value=True,\n        ),\n    ]\n    outputs = [\n        Output(name=\"response\", display_name=\"Response\", method=\"message_response\"),\n        Output(\n            name=\"structured_response\",\n            display_name=\"Structured Response\",\n            method=\"json_response\",\n            types=[\"Data\"],\n        ),\n    ]\n\n    async def _additional_model_provider_policy_ids(self, purpose, parameters=None) -> tuple[str, ...]:\n        \"\"\"Gate the provider selector retained by legacy Agent/ALTK flows.\"\"\"\n        _ = purpose\n        from lfx.base.models.provider_registry import resolve_provider_id\n\n        effective_parameters = parameters if isinstance(parameters, Mapping) else getattr(self, \"_parameters\", None)\n        if not isinstance(effective_parameters, Mapping):\n            effective_parameters = {}\n        selected_model = effective_parameters.get(\"model\", getattr(self, \"model\", None))\n        if selected_model:\n            # The shared Component hook already handles a selected ModelInput;\n            # connected model objects are gated by their upstream vertex.\n            return ()\n        legacy_provider = effective_parameters.get(\"agent_llm\", getattr(self, \"agent_llm\", None))\n        if not isinstance(legacy_provider, str) or not legacy_provider.strip() or legacy_provider == \"Custom\":\n            return ()\n        return (resolve_provider_id(legacy_provider),)\n\n    async def _filter_legacy_provider_options(self, build_config: Mapping[str, Any]) -> None:\n        \"\"\"Filter ALTK/legacy Agent provider choices through the active scope.\"\"\"\n        from lfx.services.model_provider_policy import ModelProviderPolicyPurpose, aresolve_model_provider_policy\n\n        provider_field = build_config.get(\"agent_llm\")\n        if not isinstance(provider_field, dict):\n            return\n        options = provider_field.get(\"options\")\n        if not isinstance(options, list):\n            return\n        candidates = [option for option in options if isinstance(option, str) and option != \"Custom\"]\n        if not candidates:\n            return\n        snapshot = await aresolve_model_provider_policy(\n            user_id=self.user_id,\n            providers=candidates,\n            purpose=ModelProviderPolicyPurpose.CONFIGURE,\n        )\n        allowed = set(snapshot.filter(candidates))\n        keep_indexes = [index for index, option in enumerate(options) if option == \"Custom\" or option in allowed]\n        provider_field[\"options\"] = [options[index] for index in keep_indexes]\n        metadata = provider_field.get(\"options_metadata\")\n        if isinstance(metadata, list) and len(metadata) == len(options):\n            provider_field[\"options_metadata\"] = [metadata[index] for index in keep_indexes]\n\n    def _resolve_selected_model(self):\n        \"\"\"Resolve the selected model, including legacy agent_llm/model_name inputs.\"\"\"\n        try:\n            from langchain_core.language_models import BaseLanguageModel\n\n            if isinstance(self.model, BaseLanguageModel):\n                return self.model\n        except ImportError:\n            pass\n\n        if isinstance(self.model, list) and self.model:\n            return self.model\n\n        legacy_provider = getattr(self, \"agent_llm\", None)\n        legacy_model_name = getattr(self, \"model_name\", None)\n        if not legacy_provider or not legacy_model_name:\n            return self.model\n\n        options = get_language_model_options(user_id=self.user_id)\n        for option in options:\n            if option.get(\"provider\") == legacy_provider and option.get(\"name\") == legacy_model_name:\n                return [option]\n\n        return [\n            {\n                \"name\": legacy_model_name,\n                \"provider\": legacy_provider,\n                \"metadata\": {},\n            }\n        ]\n\n    def _get_max_tokens_value(self):\n        \"\"\"Return the user-supplied max_tokens or None when unset/zero.\"\"\"\n        val = getattr(self, \"max_tokens\", None)\n        if val in {\"\", 0}:\n            return None\n        return val\n\n    def _get_llm(self):\n        \"\"\"Override parent to include max_tokens from the Agent's input field.\n\n        Streaming is mandatory for AgentComponent: ``runnable.astream_events(v2)`` only\n        emits ``on_chat_model_stream`` chunks when the underlying chat model is\n        instantiated with ``streaming=True``. Unlike the LanguageModel component (where\n        ``stream`` is a user-facing toggle), the Agent has no opt-out \u2014 the toggle is\n        kept in the UI for backwards compatibility but is intentionally ignored here.\n        Without ``stream=True``, the chat model accumulates the whole response and\n        only emits ``on_chat_model_end``, silently disabling the Playground's live-\n        typing view and breaking the streaming contract on the /events surface.\n        \"\"\"\n        return get_llm(\n            model=self.model,\n            user_id=self.user_id,\n            api_key=getattr(self, \"api_key\", None),\n            stream=True,\n            max_tokens=self._get_max_tokens_value(),\n            watsonx_url=getattr(self, \"base_url_ibm_watsonx\", None),\n            watsonx_project_id=getattr(self, \"project_id\", None),\n            overrides=getattr(self, \"_model_overrides\", None),\n        )\n\n    async def get_agent_requirements(self):\n        \"\"\"Get the agent requirements for the agent.\"\"\"\n        from langchain_core.tools import StructuredTool\n\n        from lfx.services.model_provider_policy import ModelProviderPolicyPurpose\n\n        await self.arequire_model_provider_policy(ModelProviderPolicyPurpose.USE)\n\n        selected_model = self._resolve_selected_model()\n        try:\n            from langchain_core.language_models import BaseLanguageModel\n\n            is_connected_model = isinstance(selected_model, BaseLanguageModel)\n        except ImportError:\n            is_connected_model = False\n\n        if not is_connected_model:\n            validate_model_selection(selected_model)\n\n        # Ensure _get_llm() uses the resolved model (e.g. from legacy agent_llm/model_name)\n        self.model = selected_model\n        llm_model = self._get_llm()\n        if llm_model is None:\n            msg = \"No language model selected. Please choose a model to proceed.\"\n            raise ValueError(msg)\n\n        # Get memory data\n        self.chat_history = await self.get_memory_data()\n        await logger.adebug(f\"Retrieved {len(self.chat_history)} chat history messages\")\n        if isinstance(self.chat_history, Message):\n            self.chat_history = [self.chat_history]\n\n        # Add current date tool if enabled\n        if self.add_current_date_tool:\n            if not isinstance(self.tools, list):  # type: ignore[has-type]\n                self.tools = []\n            current_date_tool = (await CurrentDateComponent(**self.get_base_args()).to_toolkit()).pop(0)\n\n            if not isinstance(current_date_tool, StructuredTool):\n                msg = \"CurrentDateComponent must be converted to a StructuredTool\"\n                raise TypeError(msg)\n            # Skip if an externally-connected tool already provides the same name.\n            # Duplicate tool names are rejected by Anthropic/Gemini with HTTP 400.\n            if not any(getattr(t, \"name\", None) == current_date_tool.name for t in self.tools):\n                self.tools.append(current_date_tool)\n\n        # Add calculator tool if enabled (zero-config arithmetic)\n        if getattr(self, \"add_calculator_tool\", False):\n            if not isinstance(self.tools, list):  # type: ignore[has-type]\n                self.tools = []\n            calculator_tool = (await CalculatorComponent(**self.get_base_args()).to_toolkit()).pop(0)\n\n            if not isinstance(calculator_tool, StructuredTool):\n                msg = \"CalculatorComponent must be converted to a StructuredTool\"\n                raise TypeError(msg)\n            # Skip if an externally-connected tool already provides the same name.\n            # Duplicate tool names are rejected by Anthropic/Gemini with HTTP 400.\n            if not any(getattr(t, \"name\", None) == calculator_tool.name for t in self.tools):\n                self.tools.append(calculator_tool)\n\n        # Set shared callbacks for tracing the tools used by the agent\n        self.set_tools_callbacks(self.tools, self._get_shared_callbacks())\n\n        return llm_model, self.chat_history, self.tools\n\n    def _get_resolved_model_name(self) -> str:\n        \"\"\"Best-effort human-readable model name for {model_name} injection.\"\"\"\n        try:\n            from langchain_core.language_models import BaseLanguageModel\n\n            if isinstance(self.model, BaseLanguageModel):\n                return type(self.model).__name__\n        except ImportError:\n            pass\n\n        if isinstance(self.model, list) and self.model:\n            first = self.model[0]\n            if isinstance(first, dict):\n                name = first.get(\"name\")\n                if isinstance(name, str) and name:\n                    return name\n\n        legacy_model_name = getattr(self, \"model_name\", None)\n        if isinstance(legacy_model_name, str) and legacy_model_name:\n            return legacy_model_name\n        return \"\"\n\n    def _inject_dynamic_prompt_values(self, prompt: Any | None) -> str | None:\n        \"\"\"Replace known env placeholders in the system prompt.\n\n        Handles {current_date}, {model_name}, and {optional_user_context} (the\n        last one ships with the structured DEFAULT_SYSTEM_PROMPT_TEMPLATE and\n        is currently unused at the AgentComponent layer, so it resolves to \"\").\n        Uses str.replace (not str.format) so user prompts containing literal\n        braces such as JSON examples ({\"key\": 1}) never break the agent.\n\n        `system_prompt` is a connectable MultilineInput, so the value can arrive\n        as a Message (e.g. a Prompt node wired in). Normalize it to text first \u2014\n        a raw Message has no `.replace` and used to crash the agent build.\n        \"\"\"\n        if prompt is None:\n            return None\n        prompt = _extract_text_content(prompt)\n        if not prompt:\n            return prompt\n        replacements = {\n            # Day precision keeps the system prompt byte-identical across turns so\n            # providers can serve the chat history from their prompt cache.\n            \"{current_date}\": datetime.now(tz=timezone.utc).date().isoformat(),\n            \"{model_name}\": self._get_resolved_model_name(),\n            \"{optional_user_context}\": \"\",\n        }\n        for placeholder, value in replacements.items():\n            prompt = prompt.replace(placeholder, value)\n        return prompt\n\n    def create_agent_runnable(self, *, allow_interrupts: bool = True):\n        \"\"\"Build the LangGraph `CompiledStateGraph` via `langchain.agents.create_agent`.\n\n        Replaces the legacy `AgentExecutor` runnable inherited from\n        `ToolCallingAgentComponent`. Other agent components (tool_calling, csv, json,\n        openapi, sql*, vector_store_router) keep the legacy path \u2014 only AgentComponent\n        runs on the new graph API.\n\n        `max_iterations` and `handle_parsing_errors` (legacy AgentExecutor knobs) are\n        translated to LangGraph middleware. Without that translation those user inputs\n        would silently become no-ops on the new API.\n\n        Provider notes:\n        - WatsonX/Granite work natively with create_agent \u2014 `ChatWatsonx.bind_tools`\n          handles tool_choice correctly. The legacy `create_granite_agent` path was\n          dropped because it hardcoded `tool_choice='required'`, which the WatsonX\n          API now rejects.\n        - Ollama and other small/local models often emit malformed tool args. The\n          ToolRetryMiddleware (default `retry_on=(Exception,)`, `on_failure='continue'`)\n          catches Pydantic ValidationErrors from bad args and feeds the error back\n          to the LLM as a retry signal, so the agent recovers gracefully.\n        \"\"\"\n        llm = self._get_llm()\n        tools = self.tools or []\n\n        # Eager bind_tools validation. `create_agent(...)` is lazy \u2014 without this,\n        # an LLM that doesn't support tool calling fails on the first user message\n        # instead of when the user wires up the component, which is a much worse UX.\n        # Gated on a non-empty tools list so a no-tool Agent on a plain chat model\n        # (which legitimately has no `bind_tools`) isn't shut out at flow-build time.\n        # Providers signal \"no tool calling\" inconsistently \u2014 `NotImplementedError`\n        # (langchain default), `AttributeError` (no `bind_tools` attr), or `TypeError`\n        # (signature mismatch). Treat all three as the same UX failure.\n        if tools:\n            try:\n                llm.bind_tools(tools)\n            except (NotImplementedError, AttributeError, TypeError) as exc:\n                # Include the underlying error so a broken tool schema or a\n                # provider implementation bug is not silently disguised as a\n                # \"model can't call tools\" UX error.\n                msg = (\n                    f\"{self.display_name} does not support tool calling, \"\n                    \"or one of the connected tools failed to bind. \"\n                    \"Please connect a tool-calling capable language model and \"\n                    f\"verify your tools are well-formed. Underlying error: {exc!s}\"\n                )\n                raise NotImplementedError(msg) from exc\n\n        middleware = self._build_middleware(llm, allow_interrupts=allow_interrupts)\n        checkpointer = self._build_agent_checkpointer() if allow_interrupts else None\n        return create_agent(\n            model=llm,\n            tools=tools,\n            system_prompt=self.system_prompt or \"\",\n            middleware=middleware or None,\n            checkpointer=checkpointer,\n        )\n\n    def _compute_recursion_limit(self) -> int:\n        \"\"\"Derive the LangGraph recursion_limit from the user-set max_iterations.\n\n        Mirrors the clamp in `_build_middleware` (max(1, max_iterations)) so a\n        saved 0 or negative value cannot under-cap the graph below one full\n        iteration.\n\n        A tool-calling cycle visits the model-call limiter's before_model and\n        after_model nodes, the model, and tools. When any tool requires approval,\n        HumanInTheLoopMiddleware adds another after_model node to every cycle,\n        even if the model repeatedly calls an ungated tool. Leave transition\n        room so the model-call limiter can end the run first.\n        \"\"\"\n        raw = getattr(self, \"max_iterations\", None)\n        run_limit = max(1, int(raw)) if raw is not None else 15\n        steps_per_cycle = 4 + bool(self._gated_interrupt_on())\n        return run_limit * steps_per_cycle + 10\n\n    def _build_middleware(self, llm: Any, *, allow_interrupts: bool = True) -> list:\n        # `llm` is passed in (rather than re-fetched via `self._get_llm()`)\n        # because some providers do credential resolution / client instantiation\n        # lazily on each call. The caller \u2014 `create_agent_runnable` \u2014 already\n        # resolved it once for `bind_tools`, so reuse that instance here.\n        middleware: list = []\n        # LangChain accepts missing IDs on AIMessage.tool_calls, but LangGraph's\n        # invalid-call path and ToolRetryMiddleware both require a string when\n        # they construct an error ToolMessage. Normalize at the model boundary\n        # so either recovery path can return the error to the model instead of\n        # crashing the flow.\n        if self.tools:\n            middleware.append(ToolCallIDMiddleware())\n        max_iterations = getattr(self, \"max_iterations\", None)\n        if max_iterations is not None:\n            # `max_iterations` is a safety cap, not an \"unlimited\" toggle. A saved\n            # 0 or negative value (falsy) must NOT silently drop the limiter and\n            # allow an unbounded model/tool loop \u2014 clamp it to a real minimum.\n            run_limit = max(1, int(max_iterations))\n            middleware.append(ModelCallLimitMiddleware(run_limit=run_limit))\n        # ToolRetryMiddleware only matters when there ARE tools to retry. Attaching\n        # it on a no-tools agent inflates the compiled graph and adds per-invocation\n        # middleware overhead for nothing, which is a measurable contributor to\n        # trivial-prompt latency (QA UI-003).\n        if getattr(self, \"handle_parsing_errors\", False) and self.tools:\n            middleware.append(ToolRetryMiddleware(max_retries=2))\n        # WatsonX models have two known platform quirks; both still reproduce on\n        # the current API, so we keep the protections from the legacy\n        # `create_granite_agent` path.\n        # 1. Multi-tool-call assistant turns are rejected (\"This model only\n        #    supports single tool-calls at once!\"). Clamp to one per turn.\n        # 2. Tool args occasionally come back as literal placeholder strings\n        #    (e.g. `<result-from-search>`). Re-invoke once with a corrective\n        #    SystemMessage.\n        # Order: SingleToolCallMiddleware first (outermost) so the clamp is\n        # applied to the final response, including any corrective re-invoke\n        # produced by WatsonXPlaceholderMiddleware.\n        if is_watsonx_model(llm):\n            middleware.append(SingleToolCallMiddleware())\n            middleware.append(WatsonXPlaceholderMiddleware())\n        # Human-in-the-loop: attach only when a tool is gated AND interrupts are allowed\n        # (the structured-output path disables them), keeping ungated flows unchanged.\n        interrupt_on = self._gated_interrupt_on() if allow_interrupts else {}\n        if interrupt_on:\n            middleware.append(HumanInTheLoopMiddleware(interrupt_on=interrupt_on))\n        return middleware\n\n    async def run_agent(self, agent) -> Message:\n        \"\"\"Run the LangGraph `CompiledStateGraph` and return the final agent Message.\n\n        Overrides the legacy `LCAgentComponent.run_agent` (which builds an\n        `{\"input\": str, \"chat_history\": [...]}` dict for `AgentExecutor`). The graph\n        wants `{\"messages\": [BaseMessage, ...]}`. The event stream is wrapped with\n        `adapt_graph_events_to_executor_shape` so the legacy `process_agent_events`\n        (in `lfx.base.agents.events`) can be reused unchanged.\n        \"\"\"\n        messages = build_initial_messages(\n            input_value=self.input_value,\n            chat_history=getattr(self, \"chat_history\", None),\n        )\n        input_dict = {\"messages\": messages}\n\n        agent_message = self._build_initial_agent_message()\n        token_usage_handler = TokenUsageCallbackHandler()\n\n        # Stream tokens to the event manager when running inside the Langflow runtime.\n        # This is what powers the live-typing view in the chat UI.\n        on_token_callback: OnTokenFunctionType | None = None\n        if getattr(self, \"_event_manager\", None):\n            on_token_callback = cast(\"OnTokenFunctionType\", self._event_manager.on_token)\n\n        # Align LangGraph's `recursion_limit` with `max_iterations` so the\n        # middleware cap (ModelCallLimitMiddleware) is what bounds the loop \u2014\n        # not LangGraph's default 25-step guard, which fires at ~12 model+tool\n        # iterations and raises a raw GraphRecursionError (QA UI-009/UI-010).\n        # Include the extra after_model node when tool approval is configured.\n        recursion_limit = self._compute_recursion_limit()\n\n        agent_config: dict[str, Any] = {\n            \"callbacks\": [\n                AgentAsyncHandler(self.log),\n                token_usage_handler,\n                *self._get_shared_callbacks(),\n            ],\n            \"recursion_limit\": recursion_limit,\n        }\n        # The durable checkpointer keys on the per-run thread_id; it must be in the\n        # astream_events config (not just create_agent) for both initial run and resume.\n        thread_id = self._agent_thread_id()\n        get_pending_interrupt = None\n        stream_input: Any = input_dict\n        # A checkpointer is only present when the agent was built with interrupts enabled\n        # (the structured-output path builds without one); never probe its state otherwise.\n        interrupts_enabled = getattr(agent, \"checkpointer\", None) is not None\n        if interrupts_enabled and thread_id and self._gated_interrupt_on():\n            agent_config[\"configurable\"] = {\"thread_id\": thread_id}\n            get_pending_interrupt = self._pending_interrupt_getter(agent, agent_config)\n            if self._has_candidate_decision(thread_id):\n                # The injected decision must match the pending interrupt's nonce, so read\n                # the interrupt first; a matched decision resumes the checkpointed thread.\n                value, interrupt_id = await self._read_pending_interrupt(agent, agent_config)\n                decision = self._injected_agent_decision(thread_id, interrupt_id)\n                if decision is not None:\n                    action_requests = (value or {}).get(\"action_requests\") or []\n                    stream_input = Command(\n                        resume={\"decisions\": self._build_resume_decisions(decision, action_requests)}\n                    )\n        stream = adapt_graph_events_to_executor_shape(\n            agent.astream_events(stream_input, config=agent_config, version=\"v2\")\n        )\n        try:\n            result = await process_agent_events(\n                stream,\n                agent_message,\n                cast(\"SendMessageFunctionType\", self.send_message),\n                on_token_callback,\n                get_pending_interrupt=get_pending_interrupt,\n            )\n        except AgentPausedError as e:\n            # Why: retract the empty partial bubble (leaks as \"[]\"); the HITL card supersedes it, resume re-emits it.\n            paused_message = e.agent_message or agent_message\n            msg_id = paused_message.get_id()\n            if msg_id:\n                await delete_message(id_=msg_id)\n            await self._send_message_event(paused_message, category=\"remove_message\")\n            return self._suspend_for_tool_approval(e.request, paused_message)\n        except ExceptionWithMessageError as e:\n            # Drop the half-stored partial message from the DB (only if it was\n            # actually persisted) and tell the frontend to remove the stale bubble.\n            if hasattr(e, \"agent_message\"):\n                msg_id = e.agent_message.get_id()\n                if msg_id:\n                    await delete_message(id_=msg_id)\n                await self._send_message_event(e.agent_message, category=\"remove_message\")\n            # exception(), not error(f\"...{e}\"): this class's str() interpolates the model's\n            # partial completion, and passing the exception rather than formatting it is what\n            # gives the exported record an error.type to triage on.\n            logger.exception(\"Agent run failed after a partial message was emitted\")\n            raise\n\n        usage_data = token_usage_handler.get_usage()\n        if usage_data:\n            self._token_usage = usage_data\n            result.properties.usage = usage_data\n            # Only round-trip the DB when the message was stored (Chat Output wired).\n            # `_should_skip_message=True` leaves `result.get_id()` empty; persisting\n            # then would create a phantom row.\n            if result.get_id():\n                stored_result = await self._update_stored_message(result)\n                await self._send_message_event(stored_result)\n                result = stored_result\n\n        self.status = result\n        return result\n\n    def _build_initial_agent_message(self) -> Message:\n        \"\"\"Construct the placeholder agent Message that `process_agent_events` mutates.\"\"\"\n        if hasattr(self, \"graph\"):\n            session_id = self.graph.session_id\n        elif hasattr(self, \"_session_id\"):\n            session_id = self._session_id\n        else:\n            session_id = None\n\n        sender_name = get_chat_output_sender_name(self) or self.display_name or \"AI\"\n        return Message(\n            sender=MESSAGE_SENDER_AI,\n            sender_name=sender_name,\n            properties={\"icon\": \"Bot\", \"state\": \"partial\"},\n            # `text=\"\"` sentinel so MessageTable's no_content check accepts\n            # an in-flight agent message whose content_blocks haven't been\n            # populated yet. Mirrors ChatInput's convention.\n            text=\"\",\n            # Flat chronological event log; see lfx.base.agents.events.\n            content_blocks=[],\n            session_id=session_id or uuid.uuid4(),\n        )\n\n    def _selected_model_remediation_context(self) -> tuple[str | None, str | None, Any | None]:\n        \"\"\"Return provider/name plus a connected model target, when present.\"\"\"\n        try:\n            selected = self._resolve_selected_model()\n            if isinstance(selected, list) and selected and isinstance(selected[0], dict):\n                return selected[0].get(\"provider\"), selected[0].get(\"name\"), None\n\n            from langchain_core.language_models import BaseLanguageModel\n\n            if isinstance(selected, BaseLanguageModel):\n                model_name = None\n                for attr in (\"model_name\", \"model\", \"model_id\"):\n                    value = getattr(selected, attr, None)\n                    if isinstance(value, str) and value:\n                        model_name = value\n                        break\n                return self._connected_model_provider(selected), model_name, selected\n        except (AttributeError, TypeError, ValueError, KeyError, ImportError):\n            pass\n        return None, None, None\n\n    def _connected_model_provider(self, model: Any) -> str | None:\n        \"\"\"Resolve a connected model's provider from the source component.\n\n        Runtime model classes are not provider identities: OpenAI, OpenRouter,\n        and compatible endpoints can all produce ``ChatOpenAI``. The incoming\n        model edge preserves the source component, so prefer its explicit\n        provider override or selected-model metadata, then its provider display\n        name. If the Agent is used without graph provenance, leave the provider\n        unknown so provider-scoped remediations remain disabled.\n        \"\"\"\n        vertex = getattr(self, \"_vertex\", None)\n        if vertex is None:\n            return None\n        source_id = vertex.get_incoming_edge_by_target_param(\"model\")\n        if not source_id:\n            return None\n        source = vertex.graph.get_vertex(source_id)\n        component = getattr(source, \"custom_component\", None)\n        if component is None:\n            return None\n\n        candidate = getattr(component, \"provider\", None)\n\n        if not isinstance(candidate, str) or not candidate:\n            selected_model = getattr(component, \"model\", None)\n            if isinstance(selected_model, list) and selected_model and isinstance(selected_model[0], dict):\n                candidate = selected_model[0].get(\"provider\")\n\n        if not isinstance(candidate, str) or not candidate:\n            candidate = getattr(component, \"display_name\", None)\n        if not isinstance(candidate, str) or not candidate:\n            return None\n\n        from lfx.base.models.unified_models.provider_queries import get_model_provider_metadata\n\n        provider_metadata = get_model_provider_metadata().get(candidate, {})\n        expected_class = provider_metadata.get(\"mapping\", {}).get(\"model_class\")\n        model_classes = {base.__name__ for base in type(model).__mro__}\n        return candidate if expected_class in model_classes else None\n\n    async def _run_agent_with_model_remediation(\n        self,\n        run_once: Callable[[], Awaitable[Message]],\n    ) -> Message:\n        \"\"\"Run one Agent operation, retrying safe provider-validation failures.\n\n        Registered remediations must identify request-validation failures that\n        occur before tool execution. A failure that can happen after a tool runs\n        must instead retry at the model-call boundary to avoid repeating tool\n        side effects.\n        \"\"\"\n        from lfx.base.models.model_remediation import apply_overrides_to_model, find_remediation, remember\n\n        provider, model_name, connected_model = self._selected_model_remediation_context()\n        applied: set[str] = set()\n        while True:\n            try:\n                result = await run_once()\n            except (ValueError, TypeError, KeyError):\n                await logger.aexception(\"Agent run failed\")\n                raise\n            except Exception as exc:\n                # run_agent may wrap the provider error in ExceptionWithMessageError, whose\n                # str() carries the model's partial completion. Matching on it locally is fine;\n                # logging it is not, so the record below passes the exception instead.\n                error_text = f\"{exc} {getattr(exc, '__cause__', '') or ''}\"\n                remediation = find_remediation(error_text, provider, already_applied=applied)\n                if remediation is None:\n                    await logger.aexception(\"Agent run failed with no remediation available\")\n                    raise\n                if connected_model is not None and not apply_overrides_to_model(connected_model, remediation.overrides):\n                    await logger.aerror(\n                        f\"model.remediation.unapplied name={remediation.name} provider={provider} model={model_name}\"\n                    )\n                    raise\n                applied.add(remediation.name)\n                if connected_model is None:\n                    self._model_overrides = {\n                        **(getattr(self, \"_model_overrides\", None) or {}),\n                        **remediation.overrides,\n                    }\n                else:\n                    # Connected outputs are shared objects across downstream flow\n                    # branches. The mutation is intentionally flow-scoped: sibling\n                    # branches holding this model will see the matched override too.\n                    await logger.adebug(\n                        f\"model.remediation.shared_model name={remediation.name} provider={provider} model={model_name}\"\n                    )\n                await logger.awarning(\n                    f\"model.remediation.applied name={remediation.name} provider={provider} model={model_name}\"\n                )\n            else:\n                if connected_model is None and getattr(self, \"_model_overrides\", None):\n                    remember(provider, model_name, self._model_overrides)\n                return result\n\n    async def message_response(self) -> Message:\n        async def _run_once() -> Message:\n            llm_model, self.chat_history, self.tools = await self.get_agent_requirements()\n            self.set(\n                llm=llm_model,\n                tools=self.tools or [],\n                chat_history=self.chat_history,\n                input_value=self.input_value,\n                system_prompt=self._inject_dynamic_prompt_values(self.system_prompt),\n            )\n            agent = self.create_agent_runnable()\n            return await self.run_agent(agent)\n\n        result = await self._run_agent_with_model_remediation(_run_once)\n        self._agent_result = result\n        return result\n\n    async def json_response(self) -> Data:\n        \"\"\"Produce structured Data via native LLM structured output, with prompt-based fallback.\n\n        Native path (no tools, llm has with_structured_output) bypasses the agent loop and\n        returns provider-validated JSON. When tools are attached, falls back to running the\n        agent with a schema-augmented system prompt and parsing the final message content.\n        \"\"\"\n        from lfx.components.models_and_agents.structured_output.structured_output_orchestrator import (\n            orchestrate_structured_output,\n        )\n\n        try:\n            llm_model, self.chat_history, self.tools = await self.get_agent_requirements()\n        except (ValueError, TypeError) as exc:\n            await logger.aexception(\"json_response.requirements_failed\")\n            return Data(data={\"content\": \"\", \"error\": str(exc)})\n\n        injected_system_prompt = self._inject_dynamic_prompt_values(getattr(self, \"system_prompt\", \"\") or \"\") or \"\"\n        format_instructions = getattr(self, \"format_instructions\", \"\") or \"\"\n        output_schema = getattr(self, \"output_schema\", None) or []\n        has_tools = bool(self.tools)\n\n        async def _run_agent_for_fallback(augmented_prompt: str) -> str:\n            first_attempt = True\n\n            async def _run_once() -> Message:\n                nonlocal first_attempt, llm_model\n                if first_attempt:\n                    first_attempt = False\n                else:\n                    llm_model, self.chat_history, self.tools = await self.get_agent_requirements()\n                self.set(\n                    llm=llm_model,\n                    tools=self.tools or [],\n                    chat_history=self.chat_history,\n                    input_value=self.input_value,\n                    system_prompt=augmented_prompt,\n                )\n                # Structured output cannot suspend mid-parse: disable tool-approval interrupts.\n                agent_runnable = self.create_agent_runnable(allow_interrupts=False)\n                return await self.run_agent(agent_runnable)\n\n            with _suppress_send_message(self):\n                result = await self._run_agent_with_model_remediation(_run_once)\n            return _extract_text_content(result)\n\n        try:\n            return await orchestrate_structured_output(\n                llm=llm_model,\n                output_schema=output_schema,\n                system_prompt=injected_system_prompt,\n                format_instructions=format_instructions,\n                input_value=_extract_text_content(self.input_value),\n                run_prompt_fallback=_run_agent_for_fallback,\n                prefer_native=not has_tools,\n            )\n        except (\n            ExceptionWithMessageError,\n            ValueError,\n            TypeError,\n            NotImplementedError,\n            AttributeError,\n        ) as exc:\n            await logger.aexception(\"json_response.orchestration_failed\")\n            return Data(data={\"content\": \"\", \"error\": str(exc)})\n\n    async def get_memory_data(self):\n        # Scope by flow_id so default playground session names (e.g. \"New Session 0\")\n        # cannot leak chat history across unrelated flows. See issue #13059.\n        # The helper also returns [] when n_messages == 0, preserving the\n        # explicit \"memory disabled\" contract from MemoryComponent.retrieve_messages.\n        messages = await aget_agent_chat_history(\n            session_id=self.graph.session_id,\n            flow_id=getattr(self.graph, \"flow_id\", None),\n            context_id=self.context_id,\n            n_messages=self.n_messages,\n            user_id=_safe_graph_user_id(self),\n        )\n        return [\n            message for message in messages if getattr(message, \"id\", None) != getattr(self.input_value, \"id\", None)\n        ]\n\n    def update_input_types(self, build_config: dotdict) -> dotdict:\n        \"\"\"Update input types for all fields in build_config.\"\"\"\n        for key, value in build_config.items():\n            if isinstance(value, dict):\n                if value.get(\"input_types\") is None:\n                    build_config[key][\"input_types\"] = []\n            elif hasattr(value, \"input_types\") and value.input_types is None:\n                value.input_types = []\n        return build_config\n\n    async def update_build_config(\n        self,\n        build_config: dotdict,\n        field_value: list[dict],\n        field_name: str | None = None,\n    ) -> dotdict:\n        from lfx.services.model_provider_policy import ModelProviderPolicyPurpose\n\n        policy_parameters = dict(getattr(self, \"_parameters\", {}) or {})\n        if field_name:\n            policy_parameters[field_name] = field_value\n        await self.arequire_model_provider_policy(\n            ModelProviderPolicyPurpose.CONFIGURE,\n            parameters=policy_parameters,\n        )\n        # Update model options with caching (for all field changes).\n        # The tool-calling constraint lives on the ModelInput's ``filters``\n        # field (declared above); ``handle_model_input_update`` reads it\n        # and applies the filter to both the dropdown options and the\n        # sticky-default re-injection path.\n        build_config = handle_model_input_update(\n            component=self,\n            build_config=dict(build_config),\n            field_value=field_value,\n            field_name=field_name,\n        )\n        build_config = dotdict(build_config)\n\n        if field_name == \"model\":\n            build_config = self.update_input_types(build_config)\n\n            # Validate required keys. `verbose` was dropped from the input set\n            # (see `_agent_base_inputs` \u2014 the create_agent event stream already\n            # surfaces every step), so it is intentionally NOT required here.\n            # Saved flows that still carry a `verbose` value just ignore it on\n            # load.\n            default_keys = [\n                \"code\",\n                \"_type\",\n                \"model\",\n                \"tools\",\n                \"input_value\",\n                \"add_current_date_tool\",\n                \"add_calculator_tool\",\n                \"system_prompt\",\n                \"max_iterations\",\n                \"handle_parsing_errors\",\n            ]\n            missing_keys = [key for key in default_keys if key not in build_config]\n            if missing_keys:\n                msg = f\"Missing required keys in build_config: {missing_keys}\"\n                raise ValueError(msg)\n        await self._filter_legacy_provider_options(build_config)\n        return dotdict({k: v.to_dict() if hasattr(v, \"to_dict\") else v for k, v in build_config.items()})\n\n    async def _get_tools(self) -> list[Tool]:\n        component_toolkit = get_component_toolkit()\n\n        tools = component_toolkit(component=self).get_tools(\n            tool_name=\"Call_Agent\",\n            # here we do not use the shared callbacks as we are exposing the agent as a tool\n            callbacks=self.get_langchain_callbacks(),\n        )\n        if hasattr(self, \"tools_metadata\"):\n            tools = component_toolkit(component=self, metadata=self.tools_metadata).update_tools_metadata(tools=tools)\n\n        return tools\n"
              },
              "context_id": {
                "_input_type": "MessageTextInput",
                "advanced": true,
                "display_name": "Context ID",
                "dynamic": false,
                "info": "The context ID of the chat. Adds an extra layer to the local memory.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "name": "context_id",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "str",
                "value": ""
              },
              "format_instructions": {
                "_input_type": "MultilineInput",
                "advanced": true,
                "ai_enabled": false,
                "copy_field": false,
                "display_name": "Output Format Instructions",
                "dynamic": false,
                "info": "Generic Template for structured output formatting. Valid only with Structured response.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "multiline": true,
                "name": "format_instructions",
                "override_skip": false,
                "password": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "str",
                "value": "You are an AI that extracts structured JSON objects from unstructured text. Use a predefined schema with expected types (str, int, float, bool, dict). Extract ALL relevant instances that match the schema - if multiple patterns exist, capture them all. Fill missing or ambiguous values with defaults: null for missing values. Remove exact duplicates but keep variations that have different field values. Always return valid JSON in the expected format, never throw errors. If multiple objects can be extracted, return them all in the structured format."
              },
              "handle_parsing_errors": {
                "_input_type": "BoolInput",
                "advanced": true,
                "display_name": "Handle Parse Errors",
                "dynamic": false,
                "info": "Adds tool-execution retry as a safety net. `create_agent` already feeds tool-call validation errors back to the LLM automatically; this flag layers `ToolRetryMiddleware` on top so transient tool runtime failures are retried (max 2 retries).",
                "input_types": [],
                "list": false,
                "list_add_label": "Add More",
                "name": "handle_parsing_errors",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "bool",
                "value": true
              },
              "input_value": {
                "_input_type": "MessageInput",
                "advanced": false,
                "display_name": "Input",
                "dynamic": false,
                "info": "The input provided by the user for the agent to process.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "name": "input_value",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": true,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "str",
                "value": ""
              },
              "max_iterations": {
                "_input_type": "IntInput",
                "advanced": true,
                "display_name": "Max Iterations",
                "dynamic": false,
                "info": "Maximum number of model calls the agent can make before stopping (maps to `ModelCallLimitMiddleware.run_limit` on the create_agent path). Must be at least 1 \u2014 it is a safety cap, never 'unlimited'.",
                "input_types": [],
                "list": false,
                "list_add_label": "Add More",
                "name": "max_iterations",
                "override_skip": false,
                "placeholder": "",
                "range_spec": {
                  "max": 128000,
                  "min": 1,
                  "step": 1,
                  "step_type": "int"
                },
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "int",
                "value": 15
              },
              "max_tokens": {
                "_input_type": "IntInput",
                "advanced": true,
                "display_name": "Max Tokens",
                "dynamic": false,
                "info": "Maximum number of tokens to generate. Set to 0 for no explicit limit. Field name varies by provider.",
                "input_types": [],
                "list": false,
                "list_add_label": "Add More",
                "name": "max_tokens",
                "override_skip": false,
                "placeholder": "",
                "range_spec": {
                  "max": 128000,
                  "min": 1,
                  "step": 1,
                  "step_type": "int"
                },
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "int",
                "value": 0
              },
              "model": {
                "_input_type": "ModelInput",
                "advanced": false,
                "display_name": "Language Model",
                "dynamic": false,
                "external_options": {
                  "fields": {
                    "data": {
                      "node": {
                        "display_name": "Connect other models",
                        "icon": "CornerDownLeft",
                        "name": "connect_other_models"
                      }
                    }
                  }
                },
                "filters": {
                  "tool_calling": true
                },
                "info": "Select your model provider",
                "input_types": [
                  "LanguageModel"
                ],
                "list": false,
                "list_add_label": "Add More",
                "model_type": "language",
                "name": "model",
                "options": [
                  {
                    "category": "OpenAI",
                    "icon": "OpenAI",
                    "metadata": {
                      "api_key_param": "api_key",
                      "context_length": 128000,
                      "max_tokens_field_name": "max_tokens",
                      "model_class": "ChatOpenAI",
                      "model_name_param": "model",
                      "reasoning_models": [
                        "gpt-5.5-pro"
                      ]
                    },
                    "name": "gpt-5.5-pro",
                    "provider": "OpenAI"
                  },
                  {
                    "category": "OpenAI",
                    "icon": "OpenAI",
                    "metadata": {
                      "api_key_param": "api_key",
                      "context_length": 128000,
                      "max_tokens_field_name": "max_tokens",
                      "model_class": "ChatOpenAI",
                      "model_name_param": "model",
                      "reasoning_models": [
                        "gpt-5.5"
                      ]
                    },
                    "name": "gpt-5.5",
                    "provider": "OpenAI"
                  },
                  {
                    "category": "OpenAI",
                    "icon": "OpenAI",
                    "metadata": {
                      "api_key_param": "api_key",
                      "context_length": 128000,
                      "max_tokens_field_name": "max_tokens",
                      "model_class": "ChatOpenAI",
                      "model_name_param": "model",
                      "reasoning_models": [
                        "gpt-5.4-nano"
                      ]
                    },
                    "name": "gpt-5.4-nano",
                    "provider": "OpenAI"
                  },
                  {
                    "category": "OpenAI",
                    "icon": "OpenAI",
                    "metadata": {
                      "api_key_param": "api_key",
                      "context_length": 128000,
                      "max_tokens_field_name": "max_tokens",
                      "model_class": "ChatOpenAI",
                      "model_name_param": "model",
                      "reasoning_models": [
                        "gpt-5.4-mini"
                      ]
                    },
                    "name": "gpt-5.4-mini",
                    "provider": "OpenAI"
                  },
                  {
                    "category": "OpenAI",
                    "icon": "OpenAI",
                    "metadata": {
                      "api_key_param": "api_key",
                      "context_length": 128000,
                      "max_tokens_field_name": "max_tokens",
                      "model_class": "ChatOpenAI",
                      "model_name_param": "model",
                      "reasoning_models": [
                        "gpt-5.4"
                      ]
                    },
                    "name": "gpt-5.4",
                    "provider": "OpenAI"
                  },
                  {
                    "category": "Google Generative AI",
                    "icon": "GoogleGenerativeAI",
                    "metadata": {
                      "api_key_param": "google_api_key",
                      "context_length": 128000,
                      "max_tokens_field_name": "max_output_tokens",
                      "model_class": "ChatGoogleGenerativeAIFixed",
                      "model_name_param": "model"
                    },
                    "name": "gemini-3.5-flash",
                    "provider": "Google Generative AI"
                  },
                  {
                    "category": "Google Generative AI",
                    "icon": "GoogleGenerativeAI",
                    "metadata": {
                      "api_key_param": "google_api_key",
                      "context_length": 128000,
                      "max_tokens_field_name": "max_output_tokens",
                      "model_class": "ChatGoogleGenerativeAIFixed",
                      "model_name_param": "model"
                    },
                    "name": "gemini-3.1-flash-lite",
                    "provider": "Google Generative AI"
                  },
                  {
                    "category": "Google Generative AI",
                    "icon": "GoogleGenerativeAI",
                    "metadata": {
                      "api_key_param": "google_api_key",
                      "context_length": 128000,
                      "max_tokens_field_name": "max_output_tokens",
                      "model_class": "ChatGoogleGenerativeAIFixed",
                      "model_name_param": "model"
                    },
                    "name": "gemma-4-31b-it",
                    "provider": "Google Generative AI"
                  },
                  {
                    "category": "Google Generative AI",
                    "icon": "GoogleGenerativeAI",
                    "metadata": {
                      "api_key_param": "google_api_key",
                      "context_length": 128000,
                      "max_tokens_field_name": "max_output_tokens",
                      "model_class": "ChatGoogleGenerativeAIFixed",
                      "model_name_param": "model"
                    },
                    "name": "gemma-4-26b-a4b-it",
                    "provider": "Google Generative AI"
                  }
                ],
                "override_skip": false,
                "placeholder": "Setup Provider",
                "real_time_refresh": true,
                "refresh_button": true,
                "required": true,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_input": true,
                "track_in_telemetry": false,
                "type": "model",
                "value": ""
              },
              "n_messages": {
                "_input_type": "IntInput",
                "advanced": true,
                "display_name": "Number of Chat History Messages",
                "dynamic": false,
                "info": "Number of chat history messages to retrieve.",
                "input_types": [],
                "list": false,
                "list_add_label": "Add More",
                "name": "n_messages",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "int",
                "value": 100
              },
              "output_schema": {
                "_input_type": "TableInput",
                "advanced": true,
                "display_name": "Output Schema",
                "dynamic": false,
                "info": "Schema Validation: Define the structure and data types for structured output. No validation if no output schema.",
                "input_types": [
                  "DataFrame",
                  "Table"
                ],
                "is_list": true,
                "list_add_label": "Add More",
                "name": "output_schema",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "table_icon": "Table",
                "table_schema": [
                  {
                    "default": "field",
                    "description": "Specify the name of the output field.",
                    "display_name": "Name",
                    "edit_mode": "inline",
                    "formatter": "text",
                    "name": "name",
                    "type": "str"
                  },
                  {
                    "default": "description of field",
                    "description": "Describe the purpose of the output field.",
                    "display_name": "Description",
                    "edit_mode": "popover",
                    "formatter": "text",
                    "name": "description",
                    "type": "str"
                  },
                  {
                    "default": "str",
                    "description": "Indicate the data type of the output field (e.g., str, int, float, bool, dict).",
                    "display_name": "Type",
                    "edit_mode": "inline",
                    "formatter": "text",
                    "name": "type",
                    "options": [
                      "str",
                      "int",
                      "float",
                      "bool",
                      "dict"
                    ],
                    "type": "str"
                  },
                  {
                    "default": "False",
                    "description": "Set to True if this output field should be a list of the specified type.",
                    "display_name": "As List",
                    "edit_mode": "inline",
                    "formatter": "text",
                    "name": "multiple",
                    "type": "boolean"
                  }
                ],
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "trigger_icon": "Table",
                "trigger_text": "Open table",
                "type": "table",
                "value": []
              },
              "project_id": {
                "_input_type": "StrInput",
                "advanced": false,
                "display_name": "watsonx Project ID",
                "dynamic": false,
                "info": "The project ID associated with the foundation model (IBM watsonx.ai only)",
                "input_types": [],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "name": "project_id",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": false,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "str",
                "value": ""
              },
              "stream": {
                "_input_type": "BoolInput",
                "advanced": true,
                "display_name": "Stream",
                "dynamic": false,
                "info": "Stream the response from the model. Streaming works only in Chat.",
                "input_types": [],
                "list": false,
                "list_add_label": "Add More",
                "name": "stream",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "bool",
                "value": true
              },
              "system_prompt": {
                "_input_type": "MultilineInput",
                "advanced": false,
                "ai_enabled": false,
                "copy_field": false,
                "display_name": "Agent Instructions",
                "dynamic": false,
                "info": "System Prompt: Initial instructions and context provided to guide the agent's behavior. Supports dynamic placeholders: {current_date}, {model_name}, {optional_user_context}.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "multiline": true,
                "name": "system_prompt",
                "override_skip": false,
                "password": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "str",
                "value": "You are a helpful document Q&A assistant. Answer the user's question using the provided context retrieved from the knowledge base. Be concise and accurate. If the context does not contain the answer, say so plainly."
              },
              "tools": {
                "_input_type": "HandleInput",
                "advanced": false,
                "display_name": "Tools",
                "dynamic": false,
                "info": "These are the tools that the agent can use to help with tasks.",
                "input_types": [
                  "Tool"
                ],
                "list": true,
                "list_add_label": "Add More",
                "name": "tools",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "other",
                "value": ""
              }
            },
            "tool_mode": false
          },
          "selected_output": "response",
          "showNode": true,
          "type": "Agent"
        },
        "dragging": false,
        "id": "Agent-docqa",
        "measured": {
          "height": 430,
          "width": 320
        },
        "position": {
          "x": 1953.0297723292472,
          "y": 526.0945709281962
        },
        "positionAbsolute": {
          "x": 2000,
          "y": 500
        },
        "selected": false,
        "type": "genericNode"
      },
      {
        "data": {
          "id": "note-docqa",
          "node": {
            "description": "# \ud83d\udcc4 Document Q&A\n\nAsk questions about your documents and get answers grounded in their content,\nusing a Knowledge Base for retrieval.\n\n## Setup\n1. **Knowledge Base** \u2014 add your documents to a Knowledge Base first (via the\n   **Knowledge** module in Langflow, or a separate ingestion flow). This flow\n   only **retrieves** \u2014 it does not ingest.\n2. **Knowledge (Retrieve)** \u2014 select the Knowledge Base that holds your documents.\n3. **Agent** \u2014 add your LLM key (any provider in the Model dropdown).\n\n## How it works\n1. **Chat Input \u2192 Knowledge (Retrieve)** \u2014 your question pulls the most relevant\n   passages from the Knowledge Base.\n2. **Parser \u2192 Prompt** \u2014 the retrieved passages become the context, combined with\n   your question.\n3. **Agent \u2192 Chat Output** \u2014 the Agent answers using **only** the retrieved\n   context (it won't make things up beyond your documents).\n\n## Run\nMake sure your Knowledge Base is populated, then type a question in the Playground.",
            "display_name": "",
            "documentation": "",
            "i18n_key": "template_notes.document_q_a.0e21c05f",
            "template": {
              "backgroundColor": "blue"
            }
          },
          "type": "note"
        },
        "dragging": false,
        "height": 921,
        "id": "note-docqa",
        "measured": {
          "height": 921,
          "width": 495
        },
        "position": {
          "x": -311.3309982486865,
          "y": 45.21891418563928
        },
        "positionAbsolute": {
          "x": -220,
          "y": 40
        },
        "resizing": false,
        "selected": false,
        "type": "noteNode",
        "width": 495
      },
      {
        "data": {
          "id": "ChatOutput-9SFlE",
          "node": {
            "base_classes": [
              "Message"
            ],
            "beta": false,
            "conditional_paths": [],
            "custom_fields": {},
            "description": "Display a chat message in the Playground.",
            "display_name": "Chat Output",
            "documentation": "https://docs.langflow.org/chat-input-and-output",
            "edited": false,
            "field_order": [
              "input_value",
              "should_store_message",
              "sender",
              "sender_name",
              "session_id",
              "context_id",
              "data_template",
              "clean_data"
            ],
            "frozen": false,
            "icon": "MessagesSquare",
            "legacy": false,
            "lf_version": "1.10.0",
            "metadata": {
              "code_hash": "0aab1c78e1dc",
              "component_class_name": "ChatOutput",
              "dependencies": {
                "dependencies": [
                  {
                    "name": "orjson",
                    "version": "3.11.9"
                  },
                  {
                    "name": "fastapi",
                    "version": "0.139.2"
                  },
                  {
                    "name": "lfx",
                    "version": null
                  }
                ],
                "total_dependencies": 3
              },
              "module": "lfx.components.input_output.chat_output.ChatOutput"
            },
            "minimized": true,
            "output_types": [],
            "outputs": [
              {
                "allows_loop": false,
                "cache": true,
                "display_name": "Output Message",
                "group_outputs": false,
                "method": "message_response",
                "name": "message",
                "selected": "Message",
                "tool_mode": true,
                "types": [
                  "Message"
                ],
                "value": "__UNDEFINED__"
              }
            ],
            "pinned": false,
            "template": {
              "_type": "Component",
              "clean_data": {
                "_input_type": "BoolInput",
                "advanced": true,
                "display_name": "Basic Clean Data",
                "dynamic": false,
                "info": "Whether to clean data before converting to string.",
                "list": false,
                "list_add_label": "Add More",
                "name": "clean_data",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "bool",
                "value": true
              },
              "code": {
                "advanced": true,
                "api_editable": false,
                "dynamic": true,
                "fileTypes": [],
                "file_path": "",
                "info": "",
                "list": false,
                "load_from_db": false,
                "multiline": true,
                "name": "code",
                "password": false,
                "placeholder": "",
                "required": true,
                "show": true,
                "title_case": false,
                "type": "code",
                "value": "from collections.abc import Generator\nfrom typing import Any\n\nimport orjson\nfrom fastapi.encoders import jsonable_encoder\n\nfrom lfx.base.io.chat import ChatComponent\nfrom lfx.helpers.data import safe_convert\nfrom lfx.inputs.inputs import BoolInput, DropdownInput, HandleInput, MessageTextInput\nfrom lfx.schema.data import Data\nfrom lfx.schema.dataframe import DataFrame\nfrom lfx.schema.message import Message\nfrom lfx.schema.properties import Source\nfrom lfx.template.field.base import Output\nfrom lfx.utils.constants import (\n    MESSAGE_SENDER_AI,\n    MESSAGE_SENDER_NAME_AI,\n    MESSAGE_SENDER_USER,\n)\n\n\nclass ChatOutput(ChatComponent):\n    display_name = \"Chat Output\"\n    description = \"Display a chat message in the Playground.\"\n    documentation: str = \"https://docs.langflow.org/chat-input-and-output\"\n    icon = \"MessagesSquare\"\n    name = \"ChatOutput\"\n    minimized = True\n\n    inputs = [\n        HandleInput(\n            name=\"input_value\",\n            display_name=\"Inputs\",\n            info=\"Message to be passed as output.\",\n            input_types=[\"Data\", \"JSON\", \"DataFrame\", \"Table\", \"Message\"],\n            required=True,\n        ),\n        BoolInput(\n            name=\"should_store_message\",\n            display_name=\"Store Messages\",\n            info=\"Store the message in the history.\",\n            value=True,\n            advanced=True,\n        ),\n        DropdownInput(\n            name=\"sender\",\n            display_name=\"Sender Type\",\n            options=[MESSAGE_SENDER_AI, MESSAGE_SENDER_USER],\n            value=MESSAGE_SENDER_AI,\n            advanced=True,\n            info=\"Type of sender.\",\n        ),\n        MessageTextInput(\n            name=\"sender_name\",\n            display_name=\"Sender Name\",\n            info=\"Name of the sender.\",\n            value=MESSAGE_SENDER_NAME_AI,\n            advanced=True,\n        ),\n        MessageTextInput(\n            name=\"session_id\",\n            display_name=\"Session ID\",\n            info=\"The session ID of the chat. If empty, the current session ID parameter will be used.\",\n            advanced=True,\n        ),\n        MessageTextInput(\n            name=\"context_id\",\n            display_name=\"Context ID\",\n            info=\"The context ID of the chat. Adds an extra layer to the local memory.\",\n            value=\"\",\n            advanced=True,\n        ),\n        MessageTextInput(\n            name=\"data_template\",\n            display_name=\"Data Template\",\n            value=\"{text}\",\n            advanced=True,\n            info=\"Template to convert Data to Text. If left empty, it will be dynamically set to the Data's text key.\",\n        ),\n        BoolInput(\n            name=\"clean_data\",\n            display_name=\"Basic Clean Data\",\n            value=True,\n            advanced=True,\n            info=\"Whether to clean data before converting to string.\",\n        ),\n    ]\n    outputs = [\n        Output(\n            display_name=\"Output Message\",\n            name=\"message\",\n            method=\"message_response\",\n        ),\n    ]\n\n    def _build_source(self, id_: str | None, display_name: str | None, source: str | None) -> Source:\n        source_dict = {}\n        if id_:\n            source_dict[\"id\"] = id_\n        if display_name:\n            source_dict[\"display_name\"] = display_name\n        if source:\n            # Handle case where source is a ChatOpenAI object\n            if hasattr(source, \"model_name\"):\n                source_dict[\"source\"] = source.model_name\n            elif hasattr(source, \"model\"):\n                source_dict[\"source\"] = str(source.model)\n            else:\n                source_dict[\"source\"] = str(source)\n        return Source(**source_dict)\n\n    async def message_response(self) -> Message:\n        # First convert the input to string if needed\n        text = self.convert_to_string()\n\n        # Get source properties\n        source, _, display_name, source_id = self.get_properties_from_source_component()\n\n        # Create or use existing Message object\n        if isinstance(self.input_value, Message) and not self.is_connected_to_chat_input():\n            message = self.input_value\n            # Update message properties\n            message.text = text\n            # Preserve existing session_id from the incoming message if it exists\n            existing_session_id = message.session_id\n        else:\n            message = Message(text=text)\n            existing_session_id = None\n\n        # Set message properties\n        message.sender = self.sender\n        message.sender_name = self.sender_name\n        # Preserve session_id from incoming message, or use component/graph session_id\n        message.session_id = (\n            self.session_id or existing_session_id or (self.graph.session_id if hasattr(self, \"graph\") else None) or \"\"\n        )\n        message.context_id = self.context_id\n        message.flow_id = self.graph.flow_id if hasattr(self, \"graph\") else None\n        message.properties.source = self._build_source(source_id, display_name, source)\n\n        # Store message if needed\n        if message.session_id and self.should_store_message:\n            stored_message = await self.send_message(message)\n            self.message.value = stored_message\n            message = stored_message\n\n        # Set accumulated token usage from all upstream LLM vertices.\n        # This must happen AFTER send_message() because streaming captures\n        # usage from chunks and would overwrite accumulated totals.\n        if hasattr(self, \"_vertex\") and self._vertex is not None:\n            accumulated_usage = self._vertex._accumulate_upstream_token_usage()  # noqa: SLF001\n            if accumulated_usage:\n                message.properties.usage = accumulated_usage\n                if self.should_store_message and message.get_id():\n                    message = await self._update_stored_message(message)\n                    await self._send_message_event(message, id_=message.get_id())\n\n        self.status = message\n        return message\n\n    def _serialize_data(self, data: Data) -> str:\n        \"\"\"Serialize Data object to JSON string.\"\"\"\n        # Convert data.data to JSON-serializable format\n        serializable_data = jsonable_encoder(data.data)\n        # Serialize with orjson, enabling pretty printing with indentation\n        json_bytes = orjson.dumps(serializable_data, option=orjson.OPT_INDENT_2)\n        # Convert bytes to string and wrap in Markdown code blocks\n        return \"```json\\n\" + json_bytes.decode(\"utf-8\") + \"\\n```\"\n\n    def _validate_input(self) -> None:\n        \"\"\"Validate the input data and raise ValueError if invalid.\"\"\"\n        if self.input_value is None:\n            msg = \"Input data cannot be None\"\n            raise ValueError(msg)\n        if isinstance(self.input_value, list) and not all(\n            isinstance(item, Message | Data | DataFrame | str) for item in self.input_value\n        ):\n            invalid_types = [\n                type(item).__name__\n                for item in self.input_value\n                if not isinstance(item, Message | Data | DataFrame | str)\n            ]\n            msg = f\"Expected Data or DataFrame or Message or str, got {invalid_types}\"\n            raise TypeError(msg)\n        if not isinstance(\n            self.input_value,\n            Message | Data | DataFrame | str | list | Generator | type(None),\n        ):\n            type_name = type(self.input_value).__name__\n            msg = f\"Expected Data or DataFrame or Message or str, Generator or None, got {type_name}\"\n            raise TypeError(msg)\n\n    def convert_to_string(self) -> str | Generator[Any, None, None]:\n        \"\"\"Convert input data to string with proper error handling.\"\"\"\n        self._validate_input()\n        clean_data: bool = getattr(self, \"clean_data\", False)\n        if isinstance(self.input_value, list):\n            return \"\\n\".join([safe_convert(item, clean_data=clean_data) for item in self.input_value])\n        if isinstance(self.input_value, Generator):\n            return self.input_value\n        return safe_convert(self.input_value, clean_data=clean_data)\n"
              },
              "context_id": {
                "_input_type": "MessageTextInput",
                "advanced": true,
                "display_name": "Context ID",
                "dynamic": false,
                "info": "The context ID of the chat. Adds an extra layer to the local memory.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "name": "context_id",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "str",
                "value": ""
              },
              "data_template": {
                "_input_type": "MessageTextInput",
                "advanced": true,
                "display_name": "Data Template",
                "dynamic": false,
                "info": "Template to convert Data to Text. If left empty, it will be dynamically set to the Data's text key.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "name": "data_template",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "str",
                "value": "{text}"
              },
              "input_value": {
                "_input_type": "HandleInput",
                "advanced": false,
                "display_name": "Inputs",
                "dynamic": false,
                "info": "Message to be passed as output.",
                "input_types": [
                  "Data",
                  "JSON",
                  "DataFrame",
                  "Table",
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "name": "input_value",
                "override_skip": false,
                "placeholder": "",
                "required": true,
                "show": true,
                "title_case": false,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "other",
                "value": ""
              },
              "sender": {
                "_input_type": "DropdownInput",
                "advanced": true,
                "combobox": false,
                "dialog_inputs": {},
                "display_name": "Sender Type",
                "dynamic": false,
                "external_options": {},
                "info": "Type of sender.",
                "name": "sender",
                "options": [
                  "Machine",
                  "User"
                ],
                "options_metadata": [],
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "toggle": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "str",
                "value": "Machine"
              },
              "sender_name": {
                "_input_type": "MessageTextInput",
                "advanced": true,
                "display_name": "Sender Name",
                "dynamic": false,
                "info": "Name of the sender.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "name": "sender_name",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "str",
                "value": "AI"
              },
              "session_id": {
                "_input_type": "MessageTextInput",
                "advanced": true,
                "display_name": "Session ID",
                "dynamic": false,
                "info": "The session ID of the chat. If empty, the current session ID parameter will be used.",
                "input_types": [
                  "Message"
                ],
                "list": false,
                "list_add_label": "Add More",
                "load_from_db": false,
                "name": "session_id",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_input": true,
                "trace_as_metadata": true,
                "track_in_telemetry": false,
                "type": "str",
                "value": ""
              },
              "should_store_message": {
                "_input_type": "BoolInput",
                "advanced": true,
                "display_name": "Store Messages",
                "dynamic": false,
                "info": "Store the message in the history.",
                "list": false,
                "list_add_label": "Add More",
                "name": "should_store_message",
                "override_skip": false,
                "placeholder": "",
                "required": false,
                "show": true,
                "title_case": false,
                "tool_mode": false,
                "trace_as_metadata": true,
                "track_in_telemetry": true,
                "type": "bool",
                "value": true
              }
            },
            "tool_mode": false
          },
          "showNode": false,
          "type": "ChatOutput"
        },
        "dragging": false,
        "id": "ChatOutput-9SFlE",
        "measured": {
          "height": 52,
          "width": 192
        },
        "position": {
          "x": 2433.8178633975485,
          "y": 881.6374781085815
        },
        "selected": false,
        "type": "genericNode"
      }
    ],
    "viewport": {
      "x": 261.82274978885016,
      "y": 51.4335786381431,
      "zoom": 0.4995615689622061
    }
  },
  "description": "Ask questions about your own document using a built-in Knowledge Base (RAG). File ingests into the KB; an Agent answers from retrieved context. No external vector database required.",
  "endpoint_name": null,
  "id": "52ece3dc-8fe7-4cc7-8c9e-0a0e9c42d3ea",
  "is_component": false,
  "last_tested_version": "1.10.0",
  "locked": false,
  "name": "Document Q&A",
  "tags": [
    "rag",
    "q-a"
  ]
}