WORKFLOWFree BlueprintLicense: MIT
Workflow to Langflow Graph to Content Aggregator
Workflow to Langflow Graph to Content Aggregator is a verified WORKFLOW blueprint on Promptally in Make.com & Langflow Graphs optimized for leading AI models. Multi-step automated node blueprint for Langflow Graph to Content Aggregator. Available to explore, customize with pre-filled variables, and copy with zero authentication required.
Created by @promptallyTechnical Specifications
| Specification | Value |
|---|---|
| Artifact type | WORKFLOW (Make.com & Langflow Graphs) |
| Compatible engines | Universal / Model Agnostic |
| Input variables | None (pre-filled) |
| Approximate tokens | 65,996 |
| Source file | Content Aggregator.json |
| Access tier | Free public (MIT) |
Pre-Filled Template
Content Aggregator.json
{
"data": {
"edges": [
{
"animated": false,
"className": "",
"data": {
"sourceHandle": {
"dataType": "ChatInput",
"id": "ChatInput-S2wgk",
"name": "message",
"output_types": [
"Message"
]
},
"targetHandle": {
"fieldName": "query",
"id": "WebSearch-ca",
"inputTypes": [
"Message"
],
"type": "str"
}
},
"id": "reactflow__edge-ChatInput-S2wgk{\u0153dataType\u0153:\u0153ChatInput\u0153,\u0153id\u0153:\u0153ChatInput-S2wgk\u0153,\u0153name\u0153:\u0153message\u0153,\u0153output_types\u0153:[\u0153Message\u0153]}-WebSearch-ca{\u0153fieldName\u0153:\u0153query\u0153,\u0153id\u0153:\u0153WebSearch-ca\u0153,\u0153inputTypes\u0153:[\u0153Message\u0153],\u0153type\u0153:\u0153str\u0153}",
"selected": false,
"source": "ChatInput-S2wgk",
"sourceHandle": "{\u0153dataType\u0153: \u0153ChatInput\u0153, \u0153id\u0153: \u0153ChatInput-S2wgk\u0153, \u0153name\u0153: \u0153message\u0153, \u0153output_types\u0153: [\u0153Message\u0153]}",
"target": "WebSearch-ca",
"targetHandle": "{\u0153fieldName\u0153: \u0153query\u0153, \u0153id\u0153: \u0153WebSearch-ca\u0153, \u0153inputTypes\u0153: [\u0153Message\u0153], \u0153type\u0153: \u0153str\u0153}"
},
{
"animated": false,
"className": "",
"data": {
"sourceHandle": {
"dataType": "UnifiedWebSearch",
"id": "WebSearch-ca",
"name": "results",
"output_types": [
"Table"
]
},
"targetHandle": {
"fieldName": "df",
"id": "BatchRun-ca",
"inputTypes": [
"DataFrame",
"Table"
],
"type": "other"
}
},
"id": "reactflow__edge-WebSearch-ca{\u0153dataType\u0153:\u0153UnifiedWebSearch\u0153,\u0153id\u0153:\u0153WebSearch-ca\u0153,\u0153name\u0153:\u0153results\u0153,\u0153output_types\u0153:[\u0153Table\u0153]}-BatchRun-ca{\u0153fieldName\u0153:\u0153df\u0153,\u0153id\u0153:\u0153BatchRun-ca\u0153,\u0153inputTypes\u0153:[\u0153DataFrame\u0153,\u0153Table\u0153],\u0153type\u0153:\u0153other\u0153}",
"selected": false,
"source": "WebSearch-ca",
"sourceHandle": "{\u0153dataType\u0153: \u0153UnifiedWebSearch\u0153, \u0153id\u0153: \u0153WebSearch-ca\u0153, \u0153name\u0153: \u0153results\u0153, \u0153output_types\u0153: [\u0153Table\u0153]}",
"target": "BatchRun-ca",
"targetHandle": "{\u0153fieldName\u0153: \u0153df\u0153, \u0153id\u0153: \u0153BatchRun-ca\u0153, \u0153inputTypes\u0153: [\u0153DataFrame\u0153, \u0153Table\u0153], \u0153type\u0153: \u0153other\u0153}"
},
{
"animated": false,
"className": "",
"data": {
"sourceHandle": {
"dataType": "BatchRunComponent",
"id": "BatchRun-ca",
"name": "batch_results",
"output_types": [
"Table"
]
},
"targetHandle": {
"fieldName": "input_data",
"id": "Parser-ca",
"inputTypes": [
"DataFrame",
"Table",
"Data",
"JSON"
],
"type": "other"
}
},
"id": "reactflow__edge-BatchRun-ca{\u0153dataType\u0153:\u0153BatchRunComponent\u0153,\u0153id\u0153:\u0153BatchRun-ca\u0153,\u0153name\u0153:\u0153batch_results\u0153,\u0153output_types\u0153:[\u0153Table\u0153]}-Parser-ca{\u0153fieldName\u0153:\u0153input_data\u0153,\u0153id\u0153:\u0153Parser-ca\u0153,\u0153inputTypes\u0153:[\u0153DataFrame\u0153,\u0153Table\u0153,\u0153Data\u0153,\u0153JSON\u0153],\u0153type\u0153:\u0153other\u0153}",
"selected": false,
"source": "BatchRun-ca",
"sourceHandle": "{\u0153dataType\u0153: \u0153BatchRunComponent\u0153, \u0153id\u0153: \u0153BatchRun-ca\u0153, \u0153name\u0153: \u0153batch_results\u0153, \u0153output_types\u0153: [\u0153Table\u0153]}",
"target": "Parser-ca",
"targetHandle": "{\u0153fieldName\u0153: \u0153input_data\u0153, \u0153id\u0153: \u0153Parser-ca\u0153, \u0153inputTypes\u0153: [\u0153DataFrame\u0153, \u0153Table\u0153, \u0153Data\u0153, \u0153JSON\u0153], \u0153type\u0153: \u0153other\u0153}"
},
{
"animated": false,
"className": "",
"data": {
"sourceHandle": {
"dataType": "ChatOutput",
"id": "ChatOutput-jNfAw",
"name": "message",
"output_types": [
"Message"
]
},
"targetHandle": {
"fieldName": "input",
"id": "SaveToFile-VVVVb",
"inputTypes": [
"Data",
"JSON",
"DataFrame",
"Table",
"Message"
],
"type": "other"
}
},
"id": "reactflow__edge-ChatOutput-jNfAw{\u0153dataType\u0153:\u0153ChatOutput\u0153,\u0153id\u0153:\u0153ChatOutput-jNfAw\u0153,\u0153name\u0153:\u0153message\u0153,\u0153output_types\u0153:[\u0153Message\u0153]}-SaveToFile-VVVVb{\u0153fieldName\u0153:\u0153input\u0153,\u0153id\u0153:\u0153SaveToFile-VVVVb\u0153,\u0153inputTypes\u0153:[\u0153Data\u0153,\u0153JSON\u0153,\u0153DataFrame\u0153,\u0153Table\u0153,\u0153Message\u0153],\u0153type\u0153:\u0153other\u0153}",
"selected": false,
"source": "ChatOutput-jNfAw",
"sourceHandle": "{\u0153dataType\u0153: \u0153ChatOutput\u0153, \u0153id\u0153: \u0153ChatOutput-jNfAw\u0153, \u0153name\u0153: \u0153message\u0153, \u0153output_types\u0153: [\u0153Message\u0153]}",
"target": "SaveToFile-VVVVb",
"targetHandle": "{\u0153fieldName\u0153: \u0153input\u0153, \u0153id\u0153: \u0153SaveToFile-VVVVb\u0153, \u0153inputTypes\u0153: [\u0153Data\u0153, \u0153JSON\u0153, \u0153DataFrame\u0153, \u0153Table\u0153, \u0153Message\u0153], \u0153type\u0153: \u0153other\u0153}"
},
{
"animated": false,
"className": "",
"data": {
"sourceHandle": {
"dataType": "parser",
"id": "Parser-ca",
"name": "parsed_text",
"output_types": [
"Message"
]
},
"targetHandle": {
"fieldName": "input_value",
"id": "Agent-ZH2Rd",
"inputTypes": [
"Message"
],
"type": "str"
}
},
"id": "xy-edge__Parser-ca{\u0153dataType\u0153:\u0153parser\u0153,\u0153id\u0153:\u0153Parser-ca\u0153,\u0153name\u0153:\u0153parsed_text\u0153,\u0153output_types\u0153:[\u0153Message\u0153]}-Agent-ZH2Rd{\u0153fieldName\u0153:\u0153input_value\u0153,\u0153id\u0153:\u0153Agent-ZH2Rd\u0153,\u0153inputTypes\u0153:[\u0153Message\u0153],\u0153type\u0153:\u0153str\u0153}",
"selected": false,
"source": "Parser-ca",
"sourceHandle": "{\u0153dataType\u0153: \u0153parser\u0153, \u0153id\u0153: \u0153Parser-ca\u0153, \u0153name\u0153: \u0153parsed_text\u0153, \u0153output_types\u0153: [\u0153Message\u0153]}",
"target": "Agent-ZH2Rd",
"targetHandle": "{\u0153fieldName\u0153: \u0153input_value\u0153, \u0153id\u0153: \u0153Agent-ZH2Rd\u0153, \u0153inputTypes\u0153: [\u0153Message\u0153], \u0153type\u0153: \u0153str\u0153}"
},
{
"animated": false,
"className": "",
"data": {
"sourceHandle": {
"dataType": "Agent",
"id": "Agent-ZH2Rd",
"name": "response",
"output_types": [
"Message"
]
},
"targetHandle": {
"fieldName": "input_value",
"id": "ChatOutput-jNfAw",
"inputTypes": [
"Data",
"JSON",
"DataFrame",
"Table",
"Message"
],
"type": "other"
}
},
"id": "xy-edge__Agent-ZH2Rd{\u0153dataType\u0153:\u0153Agent\u0153,\u0153id\u0153:\u0153Agent-ZH2Rd\u0153,\u0153name\u0153:\u0153response\u0153,\u0153output_types\u0153:[\u0153Message\u0153]}-ChatOutput-jNfAw{\u0153fieldName\u0153:\u0153input_value\u0153,\u0153id\u0153:\u0153ChatOutput-jNfAw\u0153,\u0153inputTypes\u0153:[\u0153Data\u0153,\u0153JSON\u0153,\u0153DataFrame\u0153,\u0153Table\u0153,\u0153Message\u0153],\u0153type\u0153:\u0153other\u0153}",
"selected": false,
"source": "Agent-ZH2Rd",
"sourceHandle": "{\u0153dataType\u0153: \u0153Agent\u0153, \u0153id\u0153: \u0153Agent-ZH2Rd\u0153, \u0153name\u0153: \u0153response\u0153, \u0153output_types\u0153: [\u0153Message\u0153]}",
"target": "ChatOutput-jNfAw",
"targetHandle": "{\u0153fieldName\u0153: \u0153input_value\u0153, \u0153id\u0153: \u0153ChatOutput-jNfAw\u0153, \u0153inputTypes\u0153: [\u0153Data\u0153, \u0153JSON\u0153, \u0153DataFrame\u0153, \u0153Table\u0153, \u0153Message\u0153], \u0153type\u0153: \u0153other\u0153}"
}
],
"nodes": [
{
"data": {
"id": "ChatInput-S2wgk",
"node": {
"base_classes": [
"Message"
],
"beta": false,
"category": "inputs",
"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",
"key": "ChatInput",
"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"
},
"minimized": true,
"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,
"score": 0.0020353564437605998,
"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": {
"_input_type": "FileInput",
"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,
"list_add_label": "Add More",
"name": "files",
"placeholder": "",
"required": false,
"show": true,
"temp_file": true,
"title_case": false,
"trace_as_metadata": true,
"type": "file",
"value": ""
},
"input_value": {
"_input_type": "MultilineInput",
"advanced": false,
"display_name": "Input Text",
"dynamic": false,
"info": "Message to be passed as input.",
"input_types": [],
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"multiline": true,
"name": "input_value",
"placeholder": "",
"required": false,
"show": true,
"title_case": false,
"tool_mode": false,
"trace_as_input": true,
"trace_as_metadata": true,
"type": "str",
"value": "latest developments in artificial intelligence"
},
"sender": {
"_input_type": "DropdownInput",
"advanced": true,
"combobox": false,
"dialog_inputs": {},
"display_name": "Sender Type",
"dynamic": false,
"info": "Type of sender.",
"name": "sender",
"options": [
"Machine",
"User"
],
"options_metadata": [],
"placeholder": "",
"required": false,
"show": true,
"title_case": false,
"tool_mode": false,
"trace_as_metadata": true,
"type": "str",
"value": "User"
},
"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",
"placeholder": "",
"required": false,
"show": true,
"title_case": false,
"tool_mode": false,
"trace_as_input": true,
"trace_as_metadata": true,
"type": "str",
"value": "User"
},
"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",
"placeholder": "",
"required": false,
"show": true,
"title_case": false,
"tool_mode": 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,
"list_add_label": "Add More",
"name": "should_store_message",
"placeholder": "",
"required": false,
"show": true,
"title_case": false,
"tool_mode": false,
"trace_as_metadata": true,
"type": "bool",
"value": true
}
},
"tool_mode": false
},
"selected_output": "message",
"showNode": false,
"type": "ChatInput"
},
"dragging": false,
"id": "ChatInput-S2wgk",
"measured": {
"height": 52,
"width": 192
},
"position": {
"x": 232.57443082311732,
"y": 366.5148861646235
},
"positionAbsolute": {
"x": 200,
"y": 360
},
"selected": false,
"type": "genericNode"
},
{
"data": {
"id": "WebSearch-ca",
"node": {
"base_classes": [
"Table"
],
"beta": false,
"conditional_paths": [],
"custom_fields": {},
"description": "Search the web, news, or RSS feeds.",
"display_name": "Web Search",
"documentation": "https://docs.langflow.org/web-search",
"edited": false,
"field_order": [
"search_mode",
"query",
"hl",
"gl",
"ceid",
"topic",
"location",
"timeout",
"max_results",
"max_content_length"
],
"frozen": false,
"icon": "search",
"legacy": false,
"lf_version": "1.10.0",
"metadata": {
"code_hash": "523dd690abca",
"component_class_name": "WebSearchComponent",
"dependencies": {
"dependencies": [
{
"name": "httpx",
"version": "0.28.1"
},
{
"name": "pandas",
"version": "2.3.3"
},
{
"name": "requests",
"version": "2.34.2"
},
{
"name": "bs4",
"version": "4.15.0"
},
{
"name": "pydantic",
"version": "2.13.4"
},
{
"name": "lfx",
"version": null
}
],
"total_dependencies": 6
},
"module": "lfx.components.data_source.web_search.WebSearchComponent"
},
"minimized": false,
"output_types": [],
"outputs": [
{
"allows_loop": false,
"cache": true,
"display_name": "Results",
"group_outputs": false,
"method": "perform_search",
"name": "results",
"selected": "Table",
"tool_mode": true,
"types": [
"Table"
],
"value": "__UNDEFINED__"
}
],
"pinned": false,
"template": {
"_type": "Component",
"ceid": {
"_input_type": "MessageTextInput",
"advanced": true,
"display_name": "Country:Language (ceid)",
"dynamic": false,
"info": "e.g. US:en, FR:fr. Default: US:en.",
"input_types": [],
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "ceid",
"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": "US:en"
},
"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 Web Search Component.\n\nThis component consolidates Web Search, News Search, and RSS Reader into a single\ncomponent with tabs for different search modes.\n\"\"\"\n\nimport re\nfrom typing import Any\nfrom urllib.parse import parse_qs, quote_plus, unquote, urlparse\n\nimport httpx\nimport pandas as pd\nimport requests\nfrom bs4 import BeautifulSoup\nfrom pydantic import SecretStr\n\nfrom lfx.custom import Component\nfrom lfx.io import IntInput, MessageTextInput, Output, TabInput\nfrom lfx.schema import Data, DataFrame, Message\nfrom lfx.utils.request_utils import get_user_agent\nfrom lfx.utils.ssrf_protection import SSRFProtectionError, is_ssrf_protection_enabled, validate_and_resolve_url\nfrom lfx.utils.ssrf_transport import create_ssrf_protected_sync_client, pin_host_for_url\n\nDEFAULT_MAX_RESULTS = 5\nDEFAULT_MAX_CONTENT_LENGTH = 2000\nTRUNCATION_SUFFIX = \"... [truncated]\"\nSHORT_TRUNCATION_SUFFIX = \"\u2026\"\nLIMIT_DEFAULTS = {\n \"max_results\": DEFAULT_MAX_RESULTS,\n \"max_content_length\": DEFAULT_MAX_CONTENT_LENGTH,\n}\n\n\nclass WebSearchComponent(Component):\n display_name = \"Web Search\"\n description = \"Search the web, news, or RSS feeds.\"\n documentation: str = \"https://docs.langflow.org/web-search\"\n icon = \"search\"\n name = \"UnifiedWebSearch\"\n\n inputs = [\n TabInput(\n name=\"search_mode\",\n display_name=\"Search Mode\",\n options=[\"Web\", \"News\", \"RSS\"],\n info=\"Choose search mode: Web (DuckDuckGo), News (Google News), or RSS (Feed Reader)\",\n value=\"Web\",\n real_time_refresh=True,\n tool_mode=True,\n ),\n MessageTextInput(\n name=\"query\",\n display_name=\"Search Query\",\n info=\"Search keywords for news articles.\",\n tool_mode=True,\n required=True,\n ),\n MessageTextInput(\n name=\"hl\",\n display_name=\"Language (hl)\",\n info=\"Language code, e.g. en-US, fr, de. Default: en-US.\",\n tool_mode=False,\n input_types=[],\n required=False,\n advanced=True,\n ),\n MessageTextInput(\n name=\"gl\",\n display_name=\"Country (gl)\",\n info=\"Country code, e.g. US, FR, DE. Default: US.\",\n tool_mode=False,\n input_types=[],\n required=False,\n advanced=True,\n ),\n MessageTextInput(\n name=\"ceid\",\n display_name=\"Country:Language (ceid)\",\n info=\"e.g. US:en, FR:fr. Default: US:en.\",\n tool_mode=False,\n value=\"US:en\",\n input_types=[],\n required=False,\n advanced=True,\n ),\n MessageTextInput(\n name=\"topic\",\n display_name=\"Topic\",\n info=\"One of: WORLD, NATION, BUSINESS, TECHNOLOGY, ENTERTAINMENT, SCIENCE, SPORTS, HEALTH.\",\n tool_mode=False,\n input_types=[],\n required=False,\n advanced=True,\n ),\n MessageTextInput(\n name=\"location\",\n display_name=\"Location (Geo)\",\n info=\"City, state, or country for location-based news. Leave blank for keyword search.\",\n tool_mode=False,\n input_types=[],\n required=False,\n advanced=True,\n ),\n IntInput(\n name=\"timeout\",\n display_name=\"Timeout\",\n info=\"Timeout for the request in seconds.\",\n value=5,\n required=False,\n advanced=True,\n ),\n IntInput(\n name=\"max_results\",\n display_name=\"Max Results\",\n info=\"Maximum number of results to return. Set to 0 to return every result.\",\n value=DEFAULT_MAX_RESULTS,\n required=False,\n advanced=True,\n ),\n IntInput(\n name=\"max_content_length\",\n display_name=\"Max Content Length\",\n info=(\n \"Maximum number of characters kept from each scraped page or feed field. \"\n \"Set to 0 to keep the full text.\"\n ),\n value=DEFAULT_MAX_CONTENT_LENGTH,\n required=False,\n advanced=True,\n ),\n ]\n\n outputs = [Output(name=\"results\", display_name=\"Results\", method=\"perform_search\")]\n\n def __init__(self, **kwargs):\n super().__init__(**kwargs)\n\n def update_build_config(self, build_config: dict, field_value: Any, field_name: str | None = None) -> dict:\n \"\"\"Update input visibility based on search mode.\"\"\"\n if field_name == \"search_mode\":\n # Show/hide inputs based on search mode\n is_news = field_value == \"News\"\n is_rss = field_value == \"RSS\"\n\n # Update query field info based on mode\n if is_rss:\n build_config[\"query\"][\"info\"] = \"RSS feed URL to parse\"\n build_config[\"query\"][\"display_name\"] = \"RSS Feed URL\"\n elif is_news:\n build_config[\"query\"][\"info\"] = \"Search keywords for news articles.\"\n build_config[\"query\"][\"display_name\"] = \"Search Query\"\n else: # Web\n build_config[\"query\"][\"info\"] = \"Keywords to search for\"\n build_config[\"query\"][\"display_name\"] = \"Search Query\"\n\n # Keep news-specific fields as advanced (matching original News Search component)\n # They remain advanced=True in all modes, just like in the original component\n\n return build_config\n\n def validate_url(self, string: str) -> bool:\n \"\"\"Validate URL format.\"\"\"\n url_regex = re.compile(\n r\"^(https?:\\/\\/)?\" r\"(www\\.)?\" r\"([a-zA-Z0-9.-]+)\" r\"(\\.[a-zA-Z]{2,})?\" r\"(:\\d+)?\" r\"(\\/[^\\s]*)?$\",\n re.IGNORECASE,\n )\n return bool(url_regex.match(string))\n\n def ensure_url(self, url: str) -> str:\n \"\"\"Ensure URL has proper protocol.\"\"\"\n if not url.startswith((\"http://\", \"https://\")):\n url = \"https://\" + url\n if not self.validate_url(url):\n msg = f\"Invalid URL: {url}\"\n raise ValueError(msg)\n return url\n\n def _build_safe_client(self, url: str, validated_ips: list[str]) -> httpx.Client:\n \"\"\"Create a sync HTTP client with DNS pinning when SSRF protection applies.\"\"\"\n if is_ssrf_protection_enabled() and validated_ips:\n hostname = pin_host_for_url(url)\n if hostname:\n return create_ssrf_protected_sync_client(hostname=hostname, validated_ips=validated_ips)\n return httpx.Client()\n\n def _safe_get_url(self, url: str, *, headers: dict[str, str] | None = None) -> httpx.Response:\n \"\"\"Validate an arbitrary URL and fetch it without following unvalidated redirects.\"\"\"\n url = self.ensure_url(url)\n try:\n safe_url, validated_ips = validate_and_resolve_url(url)\n except SSRFProtectionError as e:\n msg = f\"SSRF Protection: {e}\"\n raise ValueError(msg) from e\n\n with self._build_safe_client(safe_url, validated_ips) as client:\n return client.get(safe_url, headers=headers, timeout=self.timeout, follow_redirects=False)\n\n def _sanitize_query(self, query: str) -> str:\n \"\"\"Sanitize search query.\"\"\"\n return re.sub(r'[<>\"\\']', \"\", query.strip())\n\n def clean_html(self, html_string: str) -> str:\n \"\"\"Remove HTML tags from text.\"\"\"\n return BeautifulSoup(html_string, \"html.parser\").get_text(separator=\" \", strip=True)\n\n @staticmethod\n def _normalize_limit(raw: Any, default: int) -> int:\n \"\"\"Normalize a limit so only an explicit zero disables it.\"\"\"\n if isinstance(raw, Message):\n raw = raw.text\n elif isinstance(raw, Data):\n raw = raw.data.get(raw.text_key, \"\")\n if raw is None or isinstance(raw, bool) or (isinstance(raw, str) and not raw.strip()):\n return default\n if isinstance(raw, str):\n try:\n limit = int(raw)\n except ValueError:\n try:\n numeric = float(raw)\n except ValueError:\n return default\n if not numeric.is_integer():\n return default\n limit = int(numeric)\n else:\n try:\n limit = int(raw)\n except (TypeError, ValueError, OverflowError):\n return default\n if raw != limit:\n return default\n return default if limit < 0 else limit\n\n def set_attributes(self, params: dict) -> None:\n \"\"\"Normalize limits before IntInput validation mutates or rejects them.\"\"\"\n for name, default in LIMIT_DEFAULTS.items():\n if name in params and not isinstance(params[name], SecretStr):\n params[name] = self._normalize_limit(params[name], default)\n super().set_attributes(params)\n\n def _get_limit(self, name: str, default: int) -> int | None:\n \"\"\"Resolve a size limit, returning None only when the user explicitly opted out with zero.\n\n Flows saved before these inputs existed carry no value at all, so the default has to apply.\n \"\"\"\n limit = self._normalize_limit(getattr(self, name, None), default)\n return None if limit == 0 else limit\n\n def _truncate_content(self, content: str, limit: int | None) -> str:\n \"\"\"Bound a scraped page so a single result cannot flood the caller's context window.\"\"\"\n if limit is None or len(content) <= limit:\n return content\n if limit < len(TRUNCATION_SUFFIX):\n return content[: limit - len(SHORT_TRUNCATION_SUFFIX)] + SHORT_TRUNCATION_SUFFIX\n if limit == len(TRUNCATION_SUFFIX):\n return TRUNCATION_SUFFIX\n return content[: limit - len(TRUNCATION_SUFFIX)] + TRUNCATION_SUFFIX\n\n def perform_web_search(self) -> DataFrame:\n \"\"\"Perform DuckDuckGo web search.\"\"\"\n query = self._sanitize_query(self.query)\n if not query:\n msg = \"Empty search query\"\n raise ValueError(msg)\n\n headers = {\"User-Agent\": get_user_agent()}\n params = {\"q\": query, \"kl\": \"us-en\"}\n url = \"https://html.duckduckgo.com/html/\"\n\n try:\n response = requests.get(url, params=params, headers=headers, timeout=self.timeout)\n response.raise_for_status()\n except requests.RequestException as e:\n self.status = f\"Failed request: {e!s}\"\n return DataFrame(pd.DataFrame([{\"title\": \"Error\", \"link\": \"\", \"snippet\": str(e), \"content\": \"\"}]))\n\n if not response.text or \"text/html\" not in response.headers.get(\"content-type\", \"\").lower():\n self.status = \"No results found\"\n return DataFrame(\n pd.DataFrame([{\"title\": \"Error\", \"link\": \"\", \"snippet\": \"No results found\", \"content\": \"\"}])\n )\n\n soup = BeautifulSoup(response.text, \"html.parser\")\n results = []\n max_results = self._get_limit(\"max_results\", DEFAULT_MAX_RESULTS)\n max_content_length = self._get_limit(\"max_content_length\", DEFAULT_MAX_CONTENT_LENGTH)\n\n for result in soup.select(\"div.result\"):\n if max_results is not None and len(results) >= max_results:\n break\n title_tag = result.select_one(\"a.result__a\")\n snippet_tag = result.select_one(\"a.result__snippet\")\n if title_tag:\n raw_link = title_tag.get(\"href\", \"\")\n parsed = urlparse(raw_link)\n uddg = parse_qs(parsed.query).get(\"uddg\", [\"\"])[0]\n decoded_link = unquote(uddg) if uddg else raw_link\n\n try:\n final_url = self.ensure_url(decoded_link)\n page = self._safe_get_url(final_url, headers=headers)\n page.raise_for_status()\n content = BeautifulSoup(page.text, \"lxml\").get_text(separator=\" \", strip=True)\n except (httpx.HTTPError, ValueError) as e:\n final_url = decoded_link\n if \"SSRF Protection\" in str(e):\n content = f\"(Blocked by SSRF protection: {e!s})\"\n else:\n content = f\"(Failed to fetch: {e!s})\"\n\n results.append(\n {\n \"title\": title_tag.get_text(strip=True),\n \"link\": final_url,\n \"snippet\": snippet_tag.get_text(strip=True) if snippet_tag else \"\",\n \"content\": self._truncate_content(content, max_content_length),\n }\n )\n\n return DataFrame(pd.DataFrame(results))\n\n def perform_news_search(self) -> DataFrame:\n \"\"\"Perform Google News search.\"\"\"\n query = getattr(self, \"query\", \"\")\n hl = getattr(self, \"hl\", \"en-US\") or \"en-US\"\n gl = getattr(self, \"gl\", \"US\") or \"US\"\n topic = getattr(self, \"topic\", None)\n location = getattr(self, \"location\", None)\n\n ceid = f\"{gl}:{hl.split('-')[0]}\"\n\n # Build RSS URL based on parameters\n if topic:\n # Topic-based feed\n base_url = f\"https://news.google.com/rss/headlines/section/topic/{quote_plus(topic.upper())}\"\n params = f\"?hl={hl}&gl={gl}&ceid={ceid}\"\n rss_url = base_url + params\n elif location:\n # Location-based feed\n base_url = f\"https://news.google.com/rss/headlines/section/geo/{quote_plus(location)}\"\n params = f\"?hl={hl}&gl={gl}&ceid={ceid}\"\n rss_url = base_url + params\n elif query:\n # Keyword search feed\n base_url = \"https://news.google.com/rss/search?q=\"\n query_encoded = quote_plus(query)\n params = f\"&hl={hl}&gl={gl}&ceid={ceid}\"\n rss_url = f\"{base_url}{query_encoded}{params}\"\n else:\n self.status = \"No search query, topic, or location provided.\"\n return DataFrame(\n pd.DataFrame(\n [{\"title\": \"Error\", \"link\": \"\", \"published\": \"\", \"summary\": \"No search parameters provided\"}]\n )\n )\n\n try:\n response = requests.get(rss_url, timeout=self.timeout)\n response.raise_for_status()\n soup = BeautifulSoup(response.content, \"xml\")\n items = soup.find_all(\"item\")\n except requests.RequestException as e:\n self.status = f\"Failed to fetch news: {e}\"\n return DataFrame(pd.DataFrame([{\"title\": \"Error\", \"link\": \"\", \"published\": \"\", \"summary\": str(e)}]))\n\n if not items:\n self.status = \"No news articles found.\"\n return DataFrame(pd.DataFrame([{\"title\": \"No articles found\", \"link\": \"\", \"published\": \"\", \"summary\": \"\"}]))\n\n articles = []\n max_results = self._get_limit(\"max_results\", DEFAULT_MAX_RESULTS)\n max_content_length = self._get_limit(\"max_content_length\", DEFAULT_MAX_CONTENT_LENGTH)\n for item in items:\n if max_results is not None and len(articles) >= max_results:\n break\n try:\n title = self._truncate_content(\n self.clean_html(item.title.text if item.title else \"\"), max_content_length\n )\n link = self._truncate_content(item.link.text if item.link else \"\", max_content_length)\n published = self._truncate_content(item.pubDate.text if item.pubDate else \"\", max_content_length)\n summary = self._truncate_content(\n self.clean_html(item.description.text if item.description else \"\"), max_content_length\n )\n articles.append({\"title\": title, \"link\": link, \"published\": published, \"summary\": summary})\n except (AttributeError, ValueError, TypeError) as e:\n self.log(f\"Error parsing article: {e!s}\")\n continue\n\n return DataFrame(pd.DataFrame(articles))\n\n def perform_rss_read(self) -> DataFrame:\n \"\"\"Read RSS feed.\"\"\"\n rss_url = getattr(self, \"query\", \"\")\n if not rss_url:\n return DataFrame(\n pd.DataFrame([{\"title\": \"Error\", \"link\": \"\", \"published\": \"\", \"summary\": \"No RSS URL provided\"}])\n )\n\n try:\n response = self._safe_get_url(rss_url)\n response.raise_for_status()\n if not response.content.strip():\n msg = \"Empty response received\"\n raise ValueError(msg)\n\n # Validate XML\n try:\n BeautifulSoup(response.content, \"xml\")\n except Exception as e:\n msg = f\"Invalid XML response: {e}\"\n raise ValueError(msg) from e\n\n soup = BeautifulSoup(response.content, \"xml\")\n items = soup.find_all(\"item\")\n except (httpx.HTTPError, ValueError) as e:\n self.status = f\"Failed to fetch RSS: {e}\"\n return DataFrame(pd.DataFrame([{\"title\": \"Error\", \"link\": \"\", \"published\": \"\", \"summary\": str(e)}]))\n\n max_results = self._get_limit(\"max_results\", DEFAULT_MAX_RESULTS)\n max_content_length = self._get_limit(\"max_content_length\", DEFAULT_MAX_CONTENT_LENGTH)\n articles = [\n {\n \"title\": self._truncate_content(item.title.text if item.title else \"\", max_content_length),\n \"link\": self._truncate_content(item.link.text if item.link else \"\", max_content_length),\n \"published\": self._truncate_content(item.pubDate.text if item.pubDate else \"\", max_content_length),\n \"summary\": self._truncate_content(\n item.description.text if item.description else \"\", max_content_length\n ),\n }\n for item in (items if max_results is None else items[:max_results])\n ]\n\n # Ensure DataFrame has correct columns even if empty\n df_articles = pd.DataFrame(articles, columns=[\"title\", \"link\", \"published\", \"summary\"])\n self.log(f\"Fetched {len(df_articles)} articles.\")\n return DataFrame(df_articles)\n\n def perform_search(self) -> DataFrame:\n \"\"\"Main search method that routes to appropriate search function based on mode.\"\"\"\n search_mode = getattr(self, \"search_mode\", \"Web\")\n\n if search_mode == \"Web\":\n return self.perform_web_search()\n if search_mode == \"News\":\n return self.perform_news_search()\n if search_mode == \"RSS\":\n return self.perform_rss_read()\n # Fallback to web search\n return self.perform_web_search()\n"
},
"gl": {
"_input_type": "MessageTextInput",
"advanced": true,
"display_name": "Country (gl)",
"dynamic": false,
"info": "Country code, e.g. US, FR, DE. Default: US.",
"input_types": [],
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "gl",
"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": ""
},
"hl": {
"_input_type": "MessageTextInput",
"advanced": true,
"display_name": "Language (hl)",
"dynamic": false,
"info": "Language code, e.g. en-US, fr, de. Default: en-US.",
"input_types": [],
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "hl",
"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": ""
},
"location": {
"_input_type": "MessageTextInput",
"advanced": true,
"display_name": "Location (Geo)",
"dynamic": false,
"info": "City, state, or country for location-based news. Leave blank for keyword search.",
"input_types": [],
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "location",
"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": ""
},
"max_content_length": {
"_input_type": "IntInput",
"advanced": true,
"api_editable": false,
"display_name": "Max Content Length",
"dynamic": false,
"info": "Maximum number of characters kept from each scraped page or feed field. Set to 0 to keep the full text.",
"list": false,
"list_add_label": "Add More",
"name": "max_content_length",
"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": 2000
},
"max_results": {
"_input_type": "IntInput",
"advanced": true,
"api_editable": false,
"display_name": "Max Results",
"dynamic": false,
"info": "Maximum number of results to return. Set to 0 to return every result.",
"list": false,
"list_add_label": "Add More",
"name": "max_results",
"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": 5
},
"query": {
"_input_type": "MessageTextInput",
"advanced": false,
"display_name": "Search Query",
"dynamic": false,
"info": "Search keywords for news articles.",
"input_types": [
"Message"
],
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "query",
"override_skip": false,
"placeholder": "",
"required": true,
"show": true,
"title_case": false,
"tool_mode": true,
"trace_as_input": true,
"trace_as_metadata": true,
"track_in_telemetry": false,
"type": "str",
"value": ""
},
"search_mode": {
"_input_type": "TabInput",
"advanced": false,
"display_name": "Search Mode",
"dynamic": false,
"info": "Choose search mode: Web (DuckDuckGo), News (Google News), or RSS (Feed Reader)",
"name": "search_mode",
"options": [
"Web",
"News",
"RSS"
],
"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": "News"
},
"timeout": {
"_input_type": "IntInput",
"advanced": true,
"display_name": "Timeout",
"dynamic": false,
"info": "Timeout for the request in seconds.",
"list": false,
"list_add_label": "Add More",
"name": "timeout",
"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": 5
},
"topic": {
"_input_type": "MessageTextInput",
"advanced": true,
"display_name": "Topic",
"dynamic": false,
"info": "One of: WORLD, NATION, BUSINESS, TECHNOLOGY, ENTERTAINMENT, SCIENCE, SPORTS, HEALTH.",
"input_types": [],
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "topic",
"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": ""
}
},
"tool_mode": false
},
"type": "UnifiedWebSearch"
},
"dragging": false,
"id": "WebSearch-ca",
"measured": {
"height": 286,
"width": 320
},
"position": {
"x": 560,
"y": 360
},
"positionAbsolute": {
"x": 560,
"y": 360
},
"selected": false,
"type": "genericNode"
},
{
"data": {
"id": "BatchRun-ca",
"node": {
"base_classes": [
"DataFrame",
"Table"
],
"beta": false,
"conditional_paths": [],
"custom_fields": {},
"description": "Runs an LLM on each row of a DataFrame column. If no column is specified, all columns are used.",
"display_name": "Batch Run",
"documentation": "https://docs.langflow.org/batch-run",
"edited": false,
"field_order": [
"model",
"api_key",
"base_url_ibm_watsonx",
"project_id",
"system_message",
"df",
"column_name",
"output_column_name",
"enable_metadata"
],
"frozen": false,
"icon": "List",
"last_updated": "2026-06-25T21:08:27.240Z",
"legacy": false,
"lf_version": "1.10.0",
"metadata": {
"code_hash": "1b10e577e096",
"component_class_name": "BatchRunComponent",
"dependencies": {
"dependencies": [
{
"name": "toml",
"version": "0.10.2"
},
{
"name": "lfx",
"version": null
}
],
"total_dependencies": 2
},
"module": "lfx.components.llm_operations.batch_run.BatchRunComponent"
},
"minimized": false,
"output_types": [],
"outputs": [
{
"allows_loop": false,
"cache": true,
"display_name": "LLM Results",
"group_outputs": false,
"method": "run_batch",
"name": "batch_results",
"selected": "Table",
"tool_mode": true,
"types": [
"Table"
],
"value": "__UNDEFINED__"
}
],
"pinned": false,
"template": {
"_type": "Component",
"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": false,
"dialog_inputs": {},
"display_name": "watsonx API Endpoint",
"dynamic": false,
"external_options": {},
"info": "The base URL of the API (IBM watsonx.ai only)",
"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\nfrom typing import Any, cast\n\nimport toml # type: ignore[import-untyped]\n\nfrom lfx.base.models.unified_models import (\n get_llm,\n handle_model_input_update,\n)\nfrom lfx.base.models.watsonx_constants import IBM_WATSONX_URLS\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.io import (\n BoolInput,\n DataFrameInput,\n DropdownInput,\n MessageTextInput,\n ModelInput,\n MultilineInput,\n Output,\n SecretStrInput,\n StrInput,\n)\nfrom lfx.log.logger import logger\nfrom lfx.schema.dataframe import DataFrame\nfrom lfx.schema.token_usage import accumulate_usage, extract_usage_from_message\n\n\nclass BatchRunComponent(Component):\n display_name = \"Batch Run\"\n description = \"Runs an LLM on each row of a DataFrame column. If no column is specified, all columns are used.\"\n documentation: str = \"https://docs.langflow.org/batch-run\"\n icon = \"List\"\n\n inputs = [\n ModelInput(\n name=\"model\",\n display_name=\"Language Model\",\n info=\"Select your model provider\",\n real_time_refresh=True,\n required=True,\n ),\n SecretStrInput(\n name=\"api_key\",\n display_name=\"API Key\",\n info=\"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 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_message\",\n display_name=\"Instructions\",\n info=\"Multi-line system instruction for all rows in the DataFrame.\",\n required=False,\n ),\n DataFrameInput(\n name=\"df\",\n display_name=\"Table\",\n info=\"The DataFrame whose column (specified by 'column_name') we'll treat as text messages.\",\n required=True,\n ),\n MessageTextInput(\n name=\"column_name\",\n display_name=\"Column Name\",\n info=(\n \"The name of the DataFrame column to treat as text messages. \"\n \"If empty, all columns will be formatted in TOML.\"\n ),\n required=False,\n advanced=False,\n ),\n MessageTextInput(\n name=\"output_column_name\",\n display_name=\"Output Column Name\",\n info=\"Name of the column where the model's response will be stored.\",\n value=\"model_response\",\n required=False,\n advanced=True,\n ),\n BoolInput(\n name=\"enable_metadata\",\n display_name=\"Enable Metadata\",\n info=\"If True, add metadata to the output DataFrame.\",\n value=False,\n required=False,\n advanced=True,\n ),\n ]\n\n outputs = [\n Output(\n display_name=\"LLM Results\",\n name=\"batch_results\",\n method=\"run_batch\",\n info=\"A DataFrame with all original columns plus the model's response column.\",\n ),\n ]\n\n def update_build_config(self, build_config: dict, field_value: str, field_name: str | None = None):\n \"\"\"Dynamically update build config with user-filtered model options.\"\"\"\n return handle_model_input_update(self, build_config, field_value, field_name)\n\n def _format_row_as_toml(self, row: dict[str, Any]) -> str:\n \"\"\"Convert a dictionary (row) into a TOML-formatted string.\"\"\"\n formatted_dict = {str(col): {\"value\": str(val)} for col, val in row.items()}\n return toml.dumps(formatted_dict)\n\n def _create_base_row(\n self, original_row: dict[str, Any], model_response: str = \"\", batch_index: int = -1\n ) -> dict[str, Any]:\n \"\"\"Create a base row with original columns and additional metadata.\"\"\"\n row = original_row.copy()\n row[self.output_column_name] = model_response\n row[\"batch_index\"] = batch_index\n return row\n\n def _add_metadata(\n self, row: dict[str, Any], *, success: bool = True, system_msg: str = \"\", error: str | None = None\n ) -> None:\n \"\"\"Add metadata to a row if enabled.\"\"\"\n if not self.enable_metadata:\n return\n\n if success:\n row[\"metadata\"] = {\n \"has_system_message\": bool(system_msg),\n \"input_length\": len(row.get(\"text_input\", \"\")),\n \"response_length\": len(row[self.output_column_name]),\n \"processing_status\": \"success\",\n }\n else:\n row[\"metadata\"] = {\n \"error\": error,\n \"processing_status\": \"failed\",\n }\n\n async def run_batch(self) -> DataFrame:\n \"\"\"Process each row in df[column_name] with the language model asynchronously.\"\"\"\n # Check if model is already an instance (for testing) or needs to be instantiated\n if isinstance(self.model, list):\n model = get_llm(\n model=self.model,\n user_id=self.user_id,\n api_key=self.api_key,\n watsonx_url=getattr(self, \"base_url_ibm_watsonx\", None),\n watsonx_project_id=getattr(self, \"project_id\", None),\n )\n else:\n # Model is already an instance (typically in tests)\n model = self.model\n\n system_msg = self.system_message or \"\"\n df: DataFrame = self.df\n col_name = self.column_name or \"\"\n\n # Validate inputs first\n if not isinstance(df, DataFrame):\n msg = f\"Expected DataFrame input, got {type(df)}\"\n raise TypeError(msg)\n\n if col_name and col_name not in df.columns:\n msg = f\"Column '{col_name}' not found in the DataFrame. Available columns: {', '.join(df.columns)}\"\n raise ValueError(msg)\n\n try:\n # Determine text input for each row\n if col_name:\n user_texts = df[col_name].astype(str).tolist()\n else:\n user_texts = [\n self._format_row_as_toml(cast(\"dict[str, Any]\", row)) for row in df.to_dict(orient=\"records\")\n ]\n\n total_rows = len(user_texts)\n await logger.ainfo(f\"Processing {total_rows} rows with batch run\")\n\n # Prepare the batch of conversations\n conversations = [\n [{\"role\": \"system\", \"content\": system_msg}, {\"role\": \"user\", \"content\": text}]\n if system_msg\n else [{\"role\": \"user\", \"content\": text}]\n for text in user_texts\n ]\n\n # Configure the model with project info and callbacks\n # Some models (e.g., ChatWatsonx) may have serialization issues with with_config()\n # due to SecretStr or other non-serializable attributes\n try:\n model = model.with_config(\n {\n \"run_name\": self.display_name,\n \"project_name\": self.get_project_name(),\n \"callbacks\": self.get_langchain_callbacks(),\n }\n )\n except (TypeError, ValueError, AttributeError) as e:\n # Log warning and continue without configuration\n await logger.awarning(\n f\"Could not configure model with callbacks and project info: {e!s}. \"\n \"Proceeding with batch processing without configuration.\"\n )\n # Process batches and track progress\n responses_with_idx = list(\n zip(\n range(len(conversations)),\n await model.abatch(list(conversations)),\n strict=True,\n )\n )\n\n # Sort by index to maintain order\n responses_with_idx.sort(key=lambda x: x[0])\n\n # Build the final data with enhanced metadata\n rows: list[dict[str, Any]] = []\n for idx, (original_row, response) in enumerate(\n zip(df.to_dict(orient=\"records\"), responses_with_idx, strict=False)\n ):\n response_msg = response[1]\n self._token_usage = accumulate_usage(self._token_usage, extract_usage_from_message(response_msg))\n response_text = response_msg.content if hasattr(response_msg, \"content\") else str(response_msg)\n row = self._create_base_row(\n cast(\"dict[str, Any]\", original_row), model_response=response_text, batch_index=idx\n )\n self._add_metadata(row, success=True, system_msg=system_msg)\n rows.append(row)\n\n # Log progress\n if (idx + 1) % max(1, total_rows // 10) == 0:\n await logger.ainfo(f\"Processed {idx + 1}/{total_rows} rows\")\n\n await logger.ainfo(\"Batch processing completed successfully\")\n return DataFrame(rows)\n\n except (KeyError, AttributeError) as e:\n # Handle data structure and attribute access errors\n await logger.aerror(f\"Data processing error: {e!s}\")\n error_row = self._create_base_row(dict.fromkeys(df.columns, \"\"), model_response=\"\", batch_index=-1)\n self._add_metadata(error_row, success=False, error=str(e))\n return DataFrame([error_row])\n"
},
"column_name": {
"_input_type": "MessageTextInput",
"advanced": false,
"display_name": "Column Name",
"dynamic": false,
"info": "The name of the DataFrame column to treat as text messages. If empty, all columns will be formatted in TOML.",
"input_types": [
"Message"
],
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "column_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": "summary"
},
"df": {
"_input_type": "DataFrameInput",
"advanced": false,
"display_name": "Table",
"dynamic": false,
"info": "The DataFrame whose column (specified by 'column_name') we'll treat as text messages.",
"input_types": [
"DataFrame",
"Table"
],
"list": false,
"list_add_label": "Add More",
"name": "df",
"override_skip": false,
"placeholder": "",
"required": true,
"show": true,
"title_case": false,
"tool_mode": false,
"trace_as_input": true,
"trace_as_metadata": true,
"track_in_telemetry": false,
"type": "other",
"value": ""
},
"enable_metadata": {
"_input_type": "BoolInput",
"advanced": true,
"display_name": "Enable Metadata",
"dynamic": false,
"info": "If True, add metadata to the output DataFrame.",
"list": false,
"list_add_label": "Add More",
"name": "enable_metadata",
"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
},
"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"
}
}
}
},
"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": "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": ""
},
"output_column_name": {
"_input_type": "MessageTextInput",
"advanced": true,
"display_name": "Output Column Name",
"dynamic": false,
"info": "Name of the column where the model's response will be stored.",
"input_types": [
"Message"
],
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "output_column_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": "brief"
},
"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)",
"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": ""
},
"system_message": {
"_input_type": "MultilineInput",
"advanced": false,
"ai_enabled": false,
"copy_field": false,
"display_name": "Instructions",
"dynamic": false,
"info": "Multi-line system instruction for all rows in the DataFrame.",
"input_types": [
"Message"
],
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"multiline": true,
"name": "system_message",
"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": "Write one concise, neutral sentence capturing the key point of this news item. Output only the sentence, with no preamble."
}
},
"tool_mode": false
},
"showNode": true,
"type": "BatchRunComponent"
},
"dragging": false,
"id": "BatchRun-ca",
"measured": {
"height": 430,
"width": 320
},
"position": {
"x": 920,
"y": 360
},
"positionAbsolute": {
"x": 920,
"y": 360
},
"selected": false,
"type": "genericNode"
},
{
"data": {
"id": "Parser-ca",
"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": "## {title}\n{brief}\n_{published}_ - {link}\n"
},
"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-ca",
"measured": {
"height": 330,
"width": 320
},
"position": {
"x": 1280,
"y": 360
},
"positionAbsolute": {
"x": 1280,
"y": 360
},
"selected": false,
"type": "genericNode"
},
{
"data": {
"id": "Agent-ZH2Rd",
"node": {
"base_classes": [
"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-25T21:08:27.244Z",
"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": false
},
"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,
"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": "",
"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",
"name": "name",
"type": "str"
},
{
"default": "description of field",
"description": "Describe the purpose of the output field.",
"display_name": "Description",
"edit_mode": "popover",
"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",
"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",
"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": false
},
"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,
"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 content aggregator. You receive a list of news items, each with a title, a one-line brief, a date, and a link. Produce a clean, well-organized digest:\n\n- Start with a one-sentence overview of the main themes.\n- Then list the items as concise bullets, grouped by theme where it makes sense, each keeping its title, brief, date, and link.\n- Be neutral and factual. Do not add items or facts beyond those provided.\n\nFormat the digest in clean Markdown."
},
"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-ZH2Rd",
"measured": {
"height": 430,
"width": 320
},
"position": {
"x": 1640,
"y": 360
},
"positionAbsolute": {
"x": 1640,
"y": 360
},
"selected": false,
"type": "genericNode"
},
{
"data": {
"id": "ChatOutput-jNfAw",
"node": {
"base_classes": [
"Message"
],
"beta": false,
"conditional_paths": [],
"custom_fields": {},
"description": "Display a chat message in the Playground.",
"display_name": "Chat Output",
"documentation": "",
"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.4.3",
"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",
"placeholder": "",
"required": false,
"show": true,
"title_case": false,
"tool_mode": false,
"trace_as_metadata": 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",
"placeholder": "",
"required": false,
"show": true,
"title_case": false,
"tool_mode": false,
"trace_as_input": true,
"trace_as_metadata": true,
"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",
"placeholder": "",
"required": false,
"show": true,
"title_case": false,
"tool_mode": false,
"trace_as_input": true,
"trace_as_metadata": true,
"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",
"placeholder": "",
"required": true,
"show": true,
"title_case": false,
"trace_as_metadata": true,
"type": "other",
"value": ""
},
"sender": {
"_input_type": "DropdownInput",
"advanced": true,
"combobox": false,
"dialog_inputs": {},
"display_name": "Sender Type",
"dynamic": false,
"info": "Type of sender.",
"name": "sender",
"options": [
"Machine",
"User"
],
"options_metadata": [],
"placeholder": "",
"required": false,
"show": true,
"title_case": false,
"tool_mode": false,
"trace_as_metadata": 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",
"placeholder": "",
"required": false,
"show": true,
"title_case": false,
"tool_mode": false,
"trace_as_input": true,
"trace_as_metadata": true,
"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",
"placeholder": "",
"required": false,
"show": true,
"title_case": false,
"tool_mode": 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,
"list_add_label": "Add More",
"name": "should_store_message",
"placeholder": "",
"required": false,
"show": true,
"title_case": false,
"tool_mode": false,
"trace_as_metadata": true,
"type": "bool",
"value": true
}
},
"tool_mode": false
},
"selected_output": "message",
"showNode": false,
"type": "ChatOutput"
},
"dragging": false,
"id": "ChatOutput-jNfAw",
"measured": {
"height": 52,
"width": 192
},
"position": {
"x": 2017.3730297723291,
"y": 438.17863397548166
},
"positionAbsolute": {
"x": 2000,
"y": 360
},
"selected": false,
"type": "genericNode"
},
{
"data": {
"id": "SaveToFile-VVVVb",
"node": {
"base_classes": [
"Message"
],
"beta": false,
"conditional_paths": [],
"custom_fields": {},
"description": "Save content to a file in the specified format and return its path.",
"display_name": "Write File",
"documentation": "https://docs.langflow.org/components-processing#save-file",
"edited": false,
"field_order": [
"storage_location",
"input",
"file_name",
"file_format",
"append_mode",
"local_format",
"aws_format",
"gdrive_format",
"aws_access_key_id",
"aws_secret_access_key",
"bucket_name",
"aws_region",
"s3_prefix",
"service_account_key",
"folder_id"
],
"frozen": false,
"icon": "file-text",
"last_updated": "2025-09-30T16:16:26.172Z",
"legacy": false,
"metadata": {
"code_hash": "6fb7a6da11ce",
"component_class_name": "SaveToFileComponent",
"dependencies": {
"dependencies": [
{
"name": "orjson",
"version": "3.11.9"
},
{
"name": "pandas",
"version": "2.3.3"
},
{
"name": "fastapi",
"version": "0.139.2"
},
{
"name": "lfx",
"version": null
},
{
"name": "langflow",
"version": null
},
{
"name": "boto3",
"version": "1.43.46"
},
{
"name": "googleapiclient",
"version": "2.198.0"
}
],
"total_dependencies": 7
},
"module": "lfx.components.files_and_knowledge.save_file.SaveToFileComponent"
},
"minimized": false,
"output_types": [],
"outputs": [
{
"allows_loop": false,
"cache": true,
"display_name": "File Path",
"group_outputs": false,
"method": "save_to_file",
"name": "message",
"selected": "Message",
"tool_mode": true,
"types": [
"Message"
],
"value": "__UNDEFINED__"
}
],
"pinned": false,
"template": {
"_type": "Component",
"append_mode": {
"_input_type": "BoolInput",
"advanced": false,
"display_name": "Append",
"dynamic": false,
"info": "Append to file if it exists (only for Local storage with plain text formats). Not supported for cloud storage (AWS/Google Drive).",
"list": false,
"list_add_label": "Add More",
"name": "append_mode",
"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
},
"aws_access_key_id": {
"_input_type": "SecretStrInput",
"advanced": true,
"display_name": "AWS Access Key ID",
"dynamic": false,
"info": "Optional. Falls back to the AWS_ACCESS_KEY_ID environment variable.",
"input_types": [],
"load_from_db": false,
"name": "aws_access_key_id",
"password": true,
"placeholder": "",
"required": false,
"show": false,
"title_case": false,
"type": "str",
"value": ""
},
"aws_format": {
"_input_type": "DropdownInput",
"advanced": false,
"combobox": false,
"dialog_inputs": {},
"display_name": "File Format",
"dynamic": false,
"external_options": {},
"info": "Select the file format for AWS S3 storage.",
"name": "aws_format",
"options": [
"txt",
"json",
"csv",
"xml",
"html",
"md",
"yaml",
"log",
"tsv",
"jsonl",
"parquet",
"xlsx",
"zip"
],
"options_metadata": [],
"placeholder": "",
"required": false,
"show": false,
"title_case": false,
"toggle": false,
"tool_mode": false,
"trace_as_metadata": true,
"type": "str",
"value": "txt"
},
"aws_region": {
"_input_type": "StrInput",
"advanced": true,
"display_name": "AWS Region",
"dynamic": false,
"info": "AWS region (e.g., us-east-1, eu-west-1).",
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "aws_region",
"placeholder": "",
"required": false,
"show": false,
"title_case": false,
"tool_mode": false,
"trace_as_metadata": true,
"type": "str",
"value": ""
},
"aws_secret_access_key": {
"_input_type": "SecretStrInput",
"advanced": true,
"display_name": "AWS Secret Key",
"dynamic": false,
"info": "Optional. Falls back to the AWS_SECRET_ACCESS_KEY environment variable.",
"input_types": [],
"load_from_db": false,
"name": "aws_secret_access_key",
"password": true,
"placeholder": "",
"required": false,
"show": false,
"title_case": false,
"type": "str",
"value": ""
},
"bucket_name": {
"_input_type": "StrInput",
"advanced": true,
"display_name": "S3 Bucket Name",
"dynamic": false,
"info": "Optional. Falls back to the configured object storage bucket.",
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "bucket_name",
"placeholder": "",
"required": false,
"show": false,
"title_case": false,
"tool_mode": false,
"trace_as_metadata": true,
"type": "str",
"value": ""
},
"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": "import contextlib\nimport json\nfrom collections.abc import AsyncIterator, Iterator\nfrom pathlib import Path, PurePath, PureWindowsPath\nfrom typing import Any\n\nimport orjson\nimport pandas as pd\nfrom fastapi import UploadFile\nfrom fastapi.encoders import jsonable_encoder\n\nfrom lfx.custom import Component\nfrom lfx.inputs import SortableListInput\nfrom lfx.inputs.inputs import DataFrameInput\nfrom lfx.io import BoolInput, DropdownInput, SecretStrInput, StrInput\nfrom lfx.schema import Data, DataFrame, Message\nfrom lfx.services.deps import get_settings_service, get_storage_service, session_scope\nfrom lfx.template.field.base import Output\nfrom lfx.utils.end_user_storage import end_user_folder_segment, record_end_user_folder\nfrom lfx.utils.file_path_security import component_file_access_scopes, enforce_local_file_access\nfrom lfx.utils.validate_cloud import is_astra_cloud_environment\n\n\ndef _get_storage_location_options():\n \"\"\"Get storage location options, filtering out Local if in Astra cloud environment.\"\"\"\n all_options = [{\"name\": \"AWS\", \"icon\": \"Amazon\"}, {\"name\": \"Google Drive\", \"icon\": \"google\"}]\n if is_astra_cloud_environment():\n return all_options\n return [{\"name\": \"Local\", \"icon\": \"hard-drive\"}, *all_options]\n\n\ndef _get_default_storage_location() -> list[dict[str, str]]:\n \"\"\"Return the default storage selection for the component template.\"\"\"\n return [_get_storage_location_options()[0]]\n\n\ndef _is_default_storage(storage_name: str) -> bool:\n \"\"\"Check whether a storage type is the default selection.\"\"\"\n return _get_default_storage_location()[0][\"name\"] == storage_name\n\n\nclass SaveToFileComponent(Component):\n display_name = \"Write File\"\n description = \"Save content to a file in the specified format and return its path.\"\n documentation: str = \"https://docs.langflow.org/write-file\"\n icon = \"file-text\"\n name = \"SaveToFile\"\n\n # File format options for different storage types\n LOCAL_DATA_FORMAT_CHOICES = [\"csv\", \"excel\", \"json\", \"markdown\"]\n LOCAL_MESSAGE_FORMAT_CHOICES = [\"txt\", \"html\", \"json\", \"markdown\"]\n AWS_FORMAT_CHOICES = [\n \"txt\",\n \"json\",\n \"csv\",\n \"xml\",\n \"html\",\n \"md\",\n \"yaml\",\n \"log\",\n \"tsv\",\n \"jsonl\",\n \"parquet\",\n \"xlsx\",\n \"zip\",\n ]\n GDRIVE_FORMAT_CHOICES = [\"txt\", \"html\", \"json\", \"csv\", \"xlsx\", \"slides\", \"docs\", \"jpg\", \"mp3\"]\n\n inputs = [\n SortableListInput(\n name=\"storage_location\",\n display_name=\"Storage Location\",\n placeholder=\"Select Location\",\n info=\"Choose where to save the file.\",\n options=_get_storage_location_options(),\n real_time_refresh=True,\n limit=1,\n value=_get_default_storage_location(),\n advanced=True,\n ),\n # Common inputs\n DataFrameInput(\n name=\"input\",\n display_name=\"File Content\",\n info=(\n \"The content to save. Accepts a DataFrame, Data, or Message object directly. \"\n 'Can also accept a JSON string (e.g. \\'[{\"col1\": \"val1\"}]\\') which will be '\n \"parsed into a DataFrame, or plain text which will be saved as a Message.\"\n ),\n input_types=[\"Data\", \"JSON\", \"DataFrame\", \"Table\", \"Message\"],\n required=True,\n tool_mode=True,\n ),\n StrInput(\n name=\"file_name\",\n display_name=\"File Name\",\n info=\"File name without extension (e.g. 'report'). Extension is added automatically.\",\n required=True,\n show=True,\n tool_mode=True,\n ),\n StrInput(\n name=\"file_format\",\n display_name=\"File Format (Tool)\",\n info=\"Output format: 'csv', 'json', 'txt', 'html', 'excel', 'markdown'. Overrides pre-configured format.\",\n required=False,\n show=False,\n tool_mode=True,\n ),\n BoolInput(\n name=\"append_mode\",\n display_name=\"Append\",\n info=(\n \"Append to file if it exists (only for Local storage with plain text formats). \"\n \"Not supported for cloud storage (AWS/Google Drive).\"\n ),\n value=False,\n show=_is_default_storage(\"Local\"),\n ),\n # Format inputs (dynamic based on storage location)\n DropdownInput(\n name=\"local_format\",\n display_name=\"File Format\",\n options=list(dict.fromkeys(LOCAL_DATA_FORMAT_CHOICES + LOCAL_MESSAGE_FORMAT_CHOICES)),\n info=\"Select the file format for local storage.\",\n value=\"json\",\n show=_is_default_storage(\"Local\"),\n ),\n DropdownInput(\n name=\"aws_format\",\n display_name=\"File Format\",\n options=AWS_FORMAT_CHOICES,\n info=\"Select the file format for AWS S3 storage.\",\n value=\"txt\",\n show=_is_default_storage(\"AWS\"),\n ),\n DropdownInput(\n name=\"gdrive_format\",\n display_name=\"File Format\",\n options=GDRIVE_FORMAT_CHOICES,\n info=\"Select the file format for Google Drive storage.\",\n value=\"txt\",\n show=_is_default_storage(\"Google Drive\"),\n ),\n # AWS S3 specific inputs\n SecretStrInput(\n name=\"aws_access_key_id\",\n display_name=\"AWS Access Key ID\",\n info=\"Optional. Falls back to the AWS_ACCESS_KEY_ID environment variable.\",\n show=_is_default_storage(\"AWS\"),\n advanced=not _is_default_storage(\"AWS\"),\n required=False,\n ),\n SecretStrInput(\n name=\"aws_secret_access_key\",\n display_name=\"AWS Secret Key\",\n info=\"Optional. Falls back to the AWS_SECRET_ACCESS_KEY environment variable.\",\n show=_is_default_storage(\"AWS\"),\n advanced=not _is_default_storage(\"AWS\"),\n required=False,\n ),\n StrInput(\n name=\"bucket_name\",\n display_name=\"S3 Bucket Name\",\n info=\"Optional. Falls back to the configured object storage bucket.\",\n show=_is_default_storage(\"AWS\"),\n advanced=not _is_default_storage(\"AWS\"),\n required=False,\n ),\n StrInput(\n name=\"aws_region\",\n display_name=\"AWS Region\",\n info=\"AWS region (e.g., us-east-1, eu-west-1).\",\n show=_is_default_storage(\"AWS\"),\n advanced=not _is_default_storage(\"AWS\"),\n ),\n StrInput(\n name=\"s3_prefix\",\n display_name=\"S3 Prefix\",\n info=\"Prefix for all files in S3.\",\n show=_is_default_storage(\"AWS\"),\n advanced=not _is_default_storage(\"AWS\"),\n ),\n # Google Drive specific inputs\n SecretStrInput(\n name=\"service_account_key\",\n display_name=\"GCP Credentials Secret Key\",\n info=\"Your Google Cloud Platform service account JSON key as a secret string (complete JSON content).\",\n show=_is_default_storage(\"Google Drive\"),\n advanced=not _is_default_storage(\"Google Drive\"),\n required=True,\n ),\n StrInput(\n name=\"folder_id\",\n display_name=\"Google Drive Folder ID\",\n info=(\n \"The Google Drive folder ID where the file will be uploaded. \"\n \"The folder must be shared with the service account email.\"\n ),\n required=True,\n show=_is_default_storage(\"Google Drive\"),\n advanced=not _is_default_storage(\"Google Drive\"),\n ),\n ]\n\n outputs = [\n Output(\n display_name=\"File Path\",\n name=\"message\",\n method=\"save_to_file\",\n # Tool-facing documentation: ``build_description`` prefers output\n # ``info`` over the component description when this component is\n # exposed as an agent tool, so the argument reference lives here\n # instead of bloating the UI card description.\n info=(\n \"Save data to a file. \"\n \"Arguments: 'input' \u2014 the content to save (pass a DataFrame directly, or a JSON string \"\n \"for tabular data, or plain text for messages); \"\n \"'file_name' \u2014 the name to save as, without extension (e.g. 'report'); \"\n \"'file_format' \u2014 output format: 'csv', 'json', 'txt', 'html', 'excel', 'markdown' (optional). \"\n \"Returns a confirmation with the file path or URL.\"\n ),\n )\n ]\n\n def update_build_config(self, build_config, field_value, field_name=None):\n \"\"\"Update build configuration to show/hide fields based on storage location selection.\"\"\"\n # Update options dynamically based on cloud environment\n # This ensures options are refreshed when build_config is updated\n if \"storage_location\" in build_config:\n updated_options = _get_storage_location_options()\n build_config[\"storage_location\"][\"options\"] = updated_options\n\n # When tool_mode is toggled, hide storage-specific format dropdowns\n # (the agent uses the unified file_format input instead)\n if field_name == \"tool_mode\":\n format_fields = [\"local_format\", \"aws_format\", \"gdrive_format\"]\n for f_name in format_fields:\n if f_name in build_config:\n build_config[f_name][\"show\"] = not bool(field_value)\n return build_config\n\n if field_name != \"storage_location\":\n return build_config\n\n # Extract selected storage location\n selected = [location[\"name\"] for location in field_value] if isinstance(field_value, list) else []\n\n # Hide all dynamic fields first\n dynamic_fields = [\n \"file_name\", # Common fields (input is always visible)\n \"append_mode\",\n \"local_format\",\n \"aws_format\",\n \"gdrive_format\",\n \"aws_access_key_id\",\n \"aws_secret_access_key\",\n \"bucket_name\",\n \"aws_region\",\n \"s3_prefix\",\n \"service_account_key\",\n \"folder_id\",\n ]\n\n for f_name in dynamic_fields:\n if f_name in build_config:\n build_config[f_name][\"show\"] = False\n\n # Show fields based on selected storage location\n is_tool_mode = build_config.get(\"tools_metadata\", {}).get(\"show\", False)\n\n if len(selected) == 1:\n location = selected[0]\n\n # Show file_name when any storage location is selected\n if \"file_name\" in build_config:\n build_config[\"file_name\"][\"show\"] = True\n\n # Show append_mode only for Local storage (not supported for cloud storage)\n if \"append_mode\" in build_config:\n build_config[\"append_mode\"][\"show\"] = location == \"Local\"\n\n if location == \"Local\":\n if \"local_format\" in build_config:\n build_config[\"local_format\"][\"show\"] = not is_tool_mode\n\n elif location == \"AWS\":\n aws_fields = [\n \"aws_format\",\n \"aws_access_key_id\",\n \"aws_secret_access_key\",\n \"bucket_name\",\n \"aws_region\",\n \"s3_prefix\",\n ]\n for f_name in aws_fields:\n if f_name in build_config:\n show = f_name != \"aws_format\" or not is_tool_mode\n build_config[f_name][\"show\"] = show\n build_config[f_name][\"advanced\"] = False\n\n elif location == \"Google Drive\":\n gdrive_fields = [\"gdrive_format\", \"service_account_key\", \"folder_id\"]\n for f_name in gdrive_fields:\n if f_name in build_config:\n show = f_name != \"gdrive_format\" or not is_tool_mode\n build_config[f_name][\"show\"] = show\n build_config[f_name][\"advanced\"] = False\n\n return build_config\n\n async def save_to_file(self) -> Message:\n \"\"\"Save the input to a file and upload it, returning a confirmation message.\"\"\"\n # Validate inputs\n if not self.file_name:\n msg = \"File name must be provided.\"\n raise ValueError(msg)\n if not self._get_input_type():\n msg = \"Input type is not set.\"\n raise ValueError(msg)\n\n # Get selected storage location\n storage_location = self._get_selected_storage_location()\n if not storage_location:\n msg = \"Storage location must be selected.\"\n raise ValueError(msg)\n\n # Check if Local storage is disabled in cloud environment\n if storage_location == \"Local\" and is_astra_cloud_environment():\n msg = \"Local storage is not available in cloud environment. Please use AWS or Google Drive.\"\n raise ValueError(msg)\n\n # Route to appropriate save method based on storage location\n if storage_location == \"Local\":\n return await self._save_to_local()\n if storage_location == \"AWS\":\n return await self._save_to_aws()\n if storage_location == \"Google Drive\":\n return await self._save_to_google_drive()\n msg = f\"Unsupported storage location: {storage_location}\"\n raise ValueError(msg)\n\n def _get_input_type(self) -> str:\n \"\"\"Determine the input type based on the provided input.\"\"\"\n # Use exact type checking (type() is) instead of isinstance() to avoid inheritance issues.\n # Since Message inherits from Data, isinstance(message, Data) would return True for Message objects,\n # causing Message inputs to be incorrectly identified as Data type.\n if type(self.input) is DataFrame:\n return \"DataFrame\"\n if type(self.input) is Message:\n return \"Message\"\n if type(self.input) is Data:\n return \"Data\"\n # When invoked by a code agent (e.g. OpenDsStar), the input may be a raw\n # pandas DataFrame rather than Langflow's DataFrame wrapper.\n if isinstance(self.input, pd.DataFrame):\n self.input = DataFrame(self.input)\n return \"DataFrame\"\n # When invoked as a tool, the agent passes a string. Try to parse it as\n # tabular JSON (list of objects) \u2192 DataFrame, otherwise wrap as Message.\n if isinstance(self.input, str):\n self.input = self._coerce_string_input(self.input)\n return self._get_input_type()\n msg = f\"Unsupported input type: {type(self.input)}\"\n raise ValueError(msg)\n\n def _coerce_string_input(self, value: str) -> DataFrame | Message:\n \"\"\"Convert a raw string (from agent tool call) into a DataFrame or Message.\n\n Tries to parse as JSON first \u2014 a list of objects or a single object becomes\n a DataFrame. Anything else is wrapped in a Message.\n \"\"\"\n try:\n parsed = json.loads(value)\n if isinstance(parsed, list) and parsed and isinstance(parsed[0], dict):\n return DataFrame(pd.DataFrame(parsed))\n if isinstance(parsed, dict):\n return DataFrame(pd.DataFrame([parsed]))\n except (json.JSONDecodeError, ValueError):\n pass\n return Message(text=value)\n\n def _get_default_format(self) -> str:\n \"\"\"Return the default file format based on input type.\"\"\"\n if self._get_input_type() == \"DataFrame\":\n return \"csv\"\n if self._get_input_type() == \"Data\":\n return \"json\"\n if self._get_input_type() == \"Message\":\n return \"json\"\n return \"json\" # Fallback\n\n def _adjust_file_path_with_format(self, path: Path, fmt: str) -> Path:\n \"\"\"Adjust the file path to include the correct extension.\"\"\"\n file_extension = path.suffix.lower().lstrip(\".\")\n if fmt == \"excel\":\n return Path(f\"{path}.xlsx\").expanduser() if file_extension not in [\"xlsx\", \"xls\"] else path\n return Path(f\"{path}.{fmt}\").expanduser() if file_extension != fmt else path\n\n def _get_safe_local_file_name(self) -> str:\n \"\"\"Return a local file basename, rejecting paths before any file access.\"\"\"\n file_name = str(self.file_name).strip()\n if not file_name:\n msg = \"File name must be provided.\"\n raise ValueError(msg)\n\n local_path = PurePath(file_name)\n windows_path = PureWindowsPath(file_name)\n path_parts = (*local_path.parts, *windows_path.parts)\n\n if (\n \"\\x00\" in file_name\n or local_path.is_absolute()\n or windows_path.is_absolute()\n or windows_path.drive\n or len(local_path.parts) != 1\n or len(windows_path.parts) != 1\n or any(part in {\"\", \".\", \"..\"} for part in path_parts)\n ):\n msg = \"Local file name must be a file name only, without paths or parent directory references.\"\n raise ValueError(msg)\n\n return file_name\n\n def _is_plain_text_format(self, fmt: str) -> bool:\n \"\"\"Check if a file format is plain text (supports appending).\"\"\"\n plain_text_formats = [\"txt\", \"json\", \"markdown\", \"md\", \"csv\", \"xml\", \"html\", \"yaml\", \"log\", \"tsv\", \"jsonl\"]\n return fmt.lower() in plain_text_formats\n\n async def _upload_file(self, file_path: Path) -> Any:\n \"\"\"Upload the saved file using the upload_user_file service.\n\n Returns the ``UploadFileResponse`` (with the durable storage ``path`` and\n ``provider``) so the caller can report the real destination and, for\n remote backends, discard the local staging copy.\n \"\"\"\n from langflow.api.v2.files import upload_user_file\n from langflow.services.database.models.user.crud import get_user_by_id\n\n # Ensure the file exists\n if not file_path.exists():\n msg = f\"File not found: {file_path}\"\n raise FileNotFoundError(msg)\n\n # Upload the file - always use append=False because the local file already contains\n # the correct content (either new or appended locally)\n with file_path.open(\"rb\") as f:\n async with session_scope() as db:\n if not self.user_id:\n msg = \"User ID is required for file saving.\"\n raise ValueError(msg)\n current_user = await get_user_by_id(db, self.user_id)\n\n return await upload_user_file(\n file=UploadFile(filename=file_path.name, file=f, size=file_path.stat().st_size),\n session=db,\n current_user=current_user,\n storage_service=get_storage_service(),\n settings_service=get_settings_service(),\n append=False,\n )\n\n def _save_dataframe(self, dataframe: DataFrame, path: Path, fmt: str) -> str:\n \"\"\"Save a DataFrame to the specified file format.\"\"\"\n append_mode = getattr(self, \"append_mode\", False)\n should_append = append_mode and path.exists() and self._is_plain_text_format(fmt)\n\n if fmt == \"csv\":\n dataframe.to_csv(path, index=False, mode=\"a\" if should_append else \"w\", header=not should_append)\n elif fmt == \"excel\":\n dataframe.to_excel(path, index=False, engine=\"openpyxl\")\n elif fmt == \"json\":\n if should_append:\n # Read and parse existing JSON\n existing_data = []\n try:\n existing_content = path.read_text(encoding=\"utf-8\").strip()\n if existing_content:\n parsed = json.loads(existing_content)\n # Handle case where existing content is a single object\n if isinstance(parsed, dict):\n existing_data = [parsed]\n elif isinstance(parsed, list):\n existing_data = parsed\n except (json.JSONDecodeError, FileNotFoundError):\n # Treat parse errors or missing file as empty array\n existing_data = []\n\n # Append new data\n new_records = json.loads(dataframe.to_json(orient=\"records\"))\n existing_data.extend(new_records)\n\n # Write back as a single JSON array\n path.write_text(json.dumps(existing_data, indent=2), encoding=\"utf-8\")\n else:\n dataframe.to_json(path, orient=\"records\", indent=2)\n elif fmt == \"markdown\":\n content = dataframe.to_markdown(index=False)\n if should_append:\n path.write_text(path.read_text(encoding=\"utf-8\") + \"\\n\\n\" + content, encoding=\"utf-8\")\n else:\n path.write_text(content, encoding=\"utf-8\")\n else:\n msg = f\"Unsupported DataFrame format: {fmt}\"\n raise ValueError(msg)\n action = \"appended to\" if should_append else \"saved successfully as\"\n return f\"DataFrame {action} '{path}'\"\n\n def _save_data(self, data: Data, path: Path, fmt: str) -> str:\n \"\"\"Save a Data object to the specified file format.\"\"\"\n append_mode = getattr(self, \"append_mode\", False)\n should_append = append_mode and path.exists() and self._is_plain_text_format(fmt)\n\n if fmt == \"csv\":\n pd.DataFrame(data.data).to_csv(\n path,\n index=False,\n mode=\"a\" if should_append else \"w\",\n header=not should_append,\n )\n elif fmt == \"excel\":\n pd.DataFrame(data.data).to_excel(path, index=False, engine=\"openpyxl\")\n elif fmt == \"json\":\n new_data = jsonable_encoder(data.data)\n if should_append:\n # Read and parse existing JSON\n existing_data = []\n try:\n existing_content = path.read_text(encoding=\"utf-8\").strip()\n if existing_content:\n parsed = json.loads(existing_content)\n # Handle case where existing content is a single object\n if isinstance(parsed, dict):\n existing_data = [parsed]\n elif isinstance(parsed, list):\n existing_data = parsed\n except (json.JSONDecodeError, FileNotFoundError):\n # Treat parse errors or missing file as empty array\n existing_data = []\n\n # Append new data\n if isinstance(new_data, list):\n existing_data.extend(new_data)\n else:\n existing_data.append(new_data)\n\n # Write back as a single JSON array\n path.write_text(json.dumps(existing_data, indent=2), encoding=\"utf-8\")\n else:\n content = orjson.dumps(new_data, option=orjson.OPT_INDENT_2).decode(\"utf-8\")\n path.write_text(content, encoding=\"utf-8\")\n elif fmt == \"markdown\":\n content = pd.DataFrame(data.data).to_markdown(index=False)\n if should_append:\n path.write_text(path.read_text(encoding=\"utf-8\") + \"\\n\\n\" + content, encoding=\"utf-8\")\n else:\n path.write_text(content, encoding=\"utf-8\")\n else:\n msg = f\"Unsupported Data format: {fmt}\"\n raise ValueError(msg)\n action = \"appended to\" if should_append else \"saved successfully as\"\n return f\"Data {action} '{path}'\"\n\n async def _save_message(self, message: Message, path: Path, fmt: str) -> str:\n \"\"\"Save a Message to the specified file format, handling async iterators.\"\"\"\n content = \"\"\n stream = message.text_stream if hasattr(message, \"text_stream\") else None\n if stream is not None and isinstance(stream, AsyncIterator):\n async for item in stream:\n content += str(item) + \" \"\n content = content.strip()\n elif stream is not None and isinstance(stream, Iterator):\n content = \" \".join(str(item) for item in stream)\n else:\n content = str(message.text)\n\n append_mode = getattr(self, \"append_mode\", False)\n should_append = append_mode and path.exists() and self._is_plain_text_format(fmt)\n\n if fmt in (\"txt\", \"html\"):\n if should_append:\n path.write_text(path.read_text(encoding=\"utf-8\") + \"\\n\" + content, encoding=\"utf-8\")\n else:\n path.write_text(content, encoding=\"utf-8\")\n elif fmt == \"json\":\n new_message = {\"message\": content}\n if should_append:\n # Read and parse existing JSON\n existing_data = []\n try:\n existing_content = path.read_text(encoding=\"utf-8\").strip()\n if existing_content:\n parsed = json.loads(existing_content)\n # Handle case where existing content is a single object\n if isinstance(parsed, dict):\n existing_data = [parsed]\n elif isinstance(parsed, list):\n existing_data = parsed\n except (json.JSONDecodeError, FileNotFoundError):\n # Treat parse errors or missing file as empty array\n existing_data = []\n\n # Append new message\n existing_data.append(new_message)\n\n # Write back as a single JSON array\n path.write_text(json.dumps(existing_data, indent=2), encoding=\"utf-8\")\n else:\n path.write_text(json.dumps(new_message, indent=2), encoding=\"utf-8\")\n elif fmt == \"markdown\":\n md_content = f\"**Message:**\\n\\n{content}\"\n if should_append:\n path.write_text(path.read_text(encoding=\"utf-8\") + \"\\n\\n\" + md_content, encoding=\"utf-8\")\n else:\n path.write_text(md_content, encoding=\"utf-8\")\n else:\n msg = f\"Unsupported Message format: {fmt}\"\n raise ValueError(msg)\n action = \"appended to\" if should_append else \"saved successfully as\"\n return f\"Message {action} '{path}'\"\n\n def _get_selected_storage_location(self) -> str:\n \"\"\"Get the selected storage location from the SortableListInput.\"\"\"\n if hasattr(self, \"storage_location\") and self.storage_location:\n if isinstance(self.storage_location, list) and len(self.storage_location) > 0:\n return self.storage_location[0].get(\"name\", \"\")\n if isinstance(self.storage_location, dict):\n return self.storage_location.get(\"name\", \"\")\n return \"\"\n\n def _get_file_format_for_location(self, location: str) -> str:\n \"\"\"Get the appropriate file format based on storage location.\n\n If the agent set file_format via tool mode, that takes priority.\n \"\"\"\n agent_format = getattr(self, \"file_format\", None)\n if agent_format:\n return agent_format\n if location == \"Local\":\n return getattr(self, \"local_format\", None) or self._get_default_format()\n if location == \"AWS\":\n return getattr(self, \"aws_format\", \"txt\")\n if location == \"Google Drive\":\n return getattr(self, \"gdrive_format\", \"txt\")\n return self._get_default_format()\n\n def _serving_end_user_segment(self) -> str | None:\n \"\"\"Filesystem-safe end-user id to namespace saved files under, or ``None``.\n\n On the serving plane an identified end user owns the files they write during their\n session, so their id (``graph.end_user_id``) namespaces the save destination instead\n of the shared service account. Encode the complete identity so distinct ids never\n share a folder, including on case-insensitive filesystems. ``None`` off / editor / anonymous, so\n the destination falls back to the SID scope exactly as before (strict BC).\n \"\"\"\n end_user = self._serving_end_user_id()\n if not end_user:\n return None\n return end_user_folder_segment(end_user)\n\n def _serving_end_user_id(self) -> str:\n graph = getattr(getattr(self, \"_vertex\", None), \"graph\", None)\n end_user = getattr(graph, \"end_user_id\", None)\n return str(end_user) if end_user else \"\"\n\n async def _save_to_local(self) -> Message:\n \"\"\"Save file to local storage (original functionality).\"\"\"\n file_format = self._get_file_format_for_location(\"Local\")\n\n # Validate file format based on input type\n allowed_formats = (\n self.LOCAL_MESSAGE_FORMAT_CHOICES if self._get_input_type() == \"Message\" else self.LOCAL_DATA_FORMAT_CHOICES\n )\n if file_format not in allowed_formats:\n msg = f\"Invalid file format '{file_format}' for {self._get_input_type()}. Allowed: {allowed_formats}\"\n raise ValueError(msg)\n\n # Prepare file path. file_name is tenant-controlled and this writes to local disk.\n settings = get_settings_service().settings\n scope_ids = component_file_access_scopes(self)\n # Serving plane: an identified end user owns their files, so their encoded id\n # becomes the namespace folder AND an allowed access scope for enforce_local_file_access.\n # None off / editor / anonymous -> scope_ids unchanged -> SID namespace as before (BC).\n end_user_segment = self._serving_end_user_segment()\n if end_user_segment:\n scope_ids = (end_user_segment, *scope_ids)\n file_path = Path(self._get_safe_local_file_name()).expanduser()\n in_end_user_folder = False\n if settings.restrict_local_file_access and not file_path.is_absolute():\n # New files belong to the authenticated user's storage namespace. If no\n # user/flow scope exists, ``enforce_local_file_access`` below fails closed.\n scope_root = Path(scope_ids[0]) if scope_ids else Path()\n file_path = Path(settings.config_dir) / scope_root / file_path\n in_end_user_folder = end_user_segment is not None\n file_path = self._adjust_file_path_with_format(file_path, file_format)\n file_path = enforce_local_file_access(file_path, scope_ids=scope_ids)\n if in_end_user_folder:\n # Erasure deletes this folder only for the end users recorded as its owners.\n record_end_user_folder(\n Path(settings.config_dir), end_user_segment, self._serving_end_user_id(), exclusive=True\n )\n if not file_path.parent.exists():\n file_path.parent.mkdir(parents=True, exist_ok=True)\n\n # Save the input to file based on type\n if self._get_input_type() == \"DataFrame\":\n confirmation = self._save_dataframe(self.input, file_path, file_format)\n elif self._get_input_type() == \"Data\":\n confirmation = self._save_data(self.input, file_path, file_format)\n elif self._get_input_type() == \"Message\":\n confirmation = await self._save_message(self.input, file_path, file_format)\n else:\n msg = f\"Unsupported input type: {self._get_input_type()}\"\n raise ValueError(msg)\n\n # Upload the saved file into the configured storage backend.\n upload_result = await self._upload_file(file_path)\n\n # When the backend is remote (e.g. S3) the local file was only a\n # serialization staging area \u2014 the durable copy now lives in the storage\n # service. Delete the staging file (matching AWS-mode cleanup) and report\n # the storage destination instead of the misleading local path.\n # Why: on disposable executor pods the local copy is both redundant and\n # unreachable by other pods, so leaving it behind leaks a file and makes\n # the returned path point somewhere that does not durably exist. For local\n # storage the staged file IS the durable artifact, so it is kept as-is.\n #\n # Exception: append_mode accumulates content by reading the persisted local\n # file across calls (should_append hinges on path.exists()). Deleting it\n # would make the next append silently fall back to overwrite and lose the\n # accumulated content \u2014 so the staging file is preserved when appending.\n settings = get_settings_service().settings\n if settings.storage_type != \"local\":\n if not getattr(self, \"append_mode\", False):\n with contextlib.suppress(OSError):\n file_path.unlink()\n destination = upload_result.path if upload_result is not None else file_path.name\n provider = (\n upload_result.provider\n if upload_result is not None and upload_result.provider\n else settings.storage_type\n ).upper()\n action = (\n \"appended to\"\n if getattr(self, \"append_mode\", False) and self._is_plain_text_format(file_format)\n else \"saved successfully as\"\n )\n return Message(text=f\"{self._get_input_type()} {action} '{destination}' in {provider} storage\")\n\n # Local storage: the staged file is the durable artifact \u2014 keep it and\n # report its on-disk path.\n final_path = Path.cwd() / file_path if not file_path.is_absolute() else file_path\n return Message(text=f\"{confirmation} at {final_path}\")\n\n async def _save_to_aws(self) -> Message:\n \"\"\"Save file to AWS S3 using S3 functionality.\"\"\"\n import os\n\n import boto3\n\n # Get AWS credentials from component inputs or fall back to environment variables\n aws_access_key_id = getattr(self, \"aws_access_key_id\", None)\n if aws_access_key_id and hasattr(aws_access_key_id, \"get_secret_value\"):\n aws_access_key_id = aws_access_key_id.get_secret_value()\n access_key_from_environment = not aws_access_key_id\n if access_key_from_environment:\n aws_access_key_id = os.getenv(\"AWS_ACCESS_KEY_ID\")\n\n aws_secret_access_key = getattr(self, \"aws_secret_access_key\", None)\n if aws_secret_access_key and hasattr(aws_secret_access_key, \"get_secret_value\"):\n aws_secret_access_key = aws_secret_access_key.get_secret_value()\n secret_key_from_environment = not aws_secret_access_key\n if secret_key_from_environment:\n aws_secret_access_key = os.getenv(\"AWS_SECRET_ACCESS_KEY\")\n\n bucket_name = getattr(self, \"bucket_name\", None)\n if not bucket_name:\n # Try to get from storage service settings\n settings = get_settings_service().settings\n bucket_name = settings.object_storage_bucket_name\n\n # Validate AWS credentials\n if not aws_access_key_id:\n msg = (\n \"AWS Access Key ID is required for S3 storage. Provide it as a component input \"\n \"or set AWS_ACCESS_KEY_ID environment variable.\"\n )\n raise ValueError(msg)\n if not aws_secret_access_key:\n msg = (\n \"AWS Secret Key is required for S3 storage. Provide it as a component input \"\n \"or set AWS_SECRET_ACCESS_KEY environment variable.\"\n )\n raise ValueError(msg)\n if not bucket_name:\n msg = (\n \"S3 Bucket Name is required for S3 storage. Provide it as a component input \"\n \"or set LANGFLOW_OBJECT_STORAGE_BUCKET_NAME environment variable.\"\n )\n raise ValueError(msg)\n\n # Create S3 client from the resolved component or fallback values\n client_config: dict[str, Any] = {\n \"aws_access_key_id\": str(aws_access_key_id),\n \"aws_secret_access_key\": str(aws_secret_access_key),\n }\n if access_key_from_environment and secret_key_from_environment:\n aws_session_token = os.getenv(\"AWS_SESSION_TOKEN\")\n if aws_session_token:\n client_config[\"aws_session_token\"] = aws_session_token\n\n # Get region from component input, environment variable, or settings\n aws_region = getattr(self, \"aws_region\", None)\n if not aws_region:\n aws_region = os.getenv(\"AWS_DEFAULT_REGION\") or os.getenv(\"AWS_REGION\")\n if aws_region:\n client_config[\"region_name\"] = str(aws_region)\n\n s3_client = boto3.client(\"s3\", **client_config)\n\n # Extract content\n content = self._extract_content_for_upload()\n file_format = self._get_file_format_for_location(\"AWS\")\n\n # Generate file path. Namespace by user_id so concurrent multi-user runs\n # writing the same file_name don't overwrite each other \u2014 mirrors how the\n # platform storage service isolates files under \"{user_id}/\". Layout:\n # {s3_prefix}/{user_id}/{file_name}.{ext}\n file_name = f\"{self.file_name}.{file_format}\"\n path_segments = []\n if hasattr(self, \"s3_prefix\") and self.s3_prefix:\n path_segments.append(self.s3_prefix.rstrip(\"/\"))\n # Serving plane: an identified end user owns their files (encoded id); else the SID.\n user_id = self._serving_end_user_segment() or self.user_id\n if user_id and str(user_id) != \"None\":\n path_segments.append(str(user_id))\n path_segments.append(file_name)\n file_path = \"/\".join(path_segments)\n\n # Create temporary file\n import tempfile\n\n with tempfile.NamedTemporaryFile(\n mode=\"w\", encoding=\"utf-8\", suffix=f\".{file_format}\", delete=False\n ) as temp_file:\n temp_file.write(content)\n temp_file_path = temp_file.name\n\n try:\n # Upload to S3\n s3_client.upload_file(temp_file_path, bucket_name, file_path)\n s3_url = f\"s3://{bucket_name}/{file_path}\"\n return Message(text=f\"File successfully uploaded to {s3_url}\")\n finally:\n # Clean up temp file\n if Path(temp_file_path).exists():\n Path(temp_file_path).unlink()\n\n async def _save_to_google_drive(self) -> Message:\n \"\"\"Save file to Google Drive using Google Drive functionality.\"\"\"\n import tempfile\n\n from googleapiclient.http import MediaFileUpload\n\n from lfx.base.data.cloud_storage_utils import create_google_drive_service\n\n # Validate Google Drive credentials\n if not getattr(self, \"service_account_key\", None):\n msg = \"GCP Credentials Secret Key is required for Google Drive storage\"\n raise ValueError(msg)\n if not getattr(self, \"folder_id\", None):\n msg = \"Google Drive Folder ID is required for Google Drive storage\"\n raise ValueError(msg)\n\n # Create Google Drive service with full drive scope (needed for folder operations)\n drive_service, credentials = create_google_drive_service(\n self.service_account_key, scopes=[\"https://www.googleapis.com/auth/drive\"], return_credentials=True\n )\n\n # Extract content and format\n content = self._extract_content_for_upload()\n file_format = self._get_file_format_for_location(\"Google Drive\")\n\n # Handle special Google Drive formats\n if file_format in [\"slides\", \"docs\"]:\n return await self._save_to_google_apps(drive_service, credentials, content, file_format)\n\n # Create temporary file\n file_path = f\"{self.file_name}.{file_format}\"\n with tempfile.NamedTemporaryFile(\n mode=\"w\",\n encoding=\"utf-8\",\n suffix=f\".{file_format}\",\n delete=False,\n ) as temp_file:\n temp_file.write(content)\n temp_file_path = temp_file.name\n\n try:\n # Upload to Google Drive\n # Note: We skip explicit folder verification since it requires broader permissions.\n # If the folder doesn't exist or isn't accessible, the create() call will fail with a clear error.\n file_metadata = {\"name\": file_path, \"parents\": [self.folder_id]}\n media = MediaFileUpload(temp_file_path, resumable=True)\n\n try:\n uploaded_file = (\n drive_service.files().create(body=file_metadata, media_body=media, fields=\"id\").execute()\n )\n except Exception as e:\n msg = (\n f\"Unable to upload file to Google Drive folder '{self.folder_id}'. \"\n f\"Error: {e!s}. \"\n \"Please ensure: 1) The folder ID is correct, 2) The folder exists, \"\n \"3) The service account has been granted access to this folder.\"\n )\n raise ValueError(msg) from e\n\n file_id = uploaded_file.get(\"id\")\n file_url = f\"https://drive.google.com/file/d/{file_id}/view\"\n return Message(text=f\"File successfully uploaded to Google Drive: {file_url}\")\n finally:\n # Clean up temp file\n if Path(temp_file_path).exists():\n Path(temp_file_path).unlink()\n\n async def _save_to_google_apps(self, drive_service, credentials, content: str, app_type: str) -> Message:\n \"\"\"Save content to Google Apps (Slides or Docs).\"\"\"\n import time\n\n if app_type == \"slides\":\n from googleapiclient.discovery import build\n\n slides_service = build(\"slides\", \"v1\", credentials=credentials)\n\n file_metadata = {\n \"name\": self.file_name,\n \"mimeType\": \"application/vnd.google-apps.presentation\",\n \"parents\": [self.folder_id],\n }\n\n created_file = drive_service.files().create(body=file_metadata, fields=\"id\").execute()\n presentation_id = created_file[\"id\"]\n\n time.sleep(2) # Wait for file to be available # noqa: ASYNC251\n\n presentation = slides_service.presentations().get(presentationId=presentation_id).execute()\n slide_id = presentation[\"slides\"][0][\"objectId\"]\n\n # Add content to slide\n requests = [\n {\n \"createShape\": {\n \"objectId\": \"TextBox_01\",\n \"shapeType\": \"TEXT_BOX\",\n \"elementProperties\": {\n \"pageObjectId\": slide_id,\n \"size\": {\n \"height\": {\"magnitude\": 3000000, \"unit\": \"EMU\"},\n \"width\": {\"magnitude\": 6000000, \"unit\": \"EMU\"},\n },\n \"transform\": {\n \"scaleX\": 1,\n \"scaleY\": 1,\n \"translateX\": 1000000,\n \"translateY\": 1000000,\n \"unit\": \"EMU\",\n },\n },\n }\n },\n {\"insertText\": {\"objectId\": \"TextBox_01\", \"insertionIndex\": 0, \"text\": content}},\n ]\n\n slides_service.presentations().batchUpdate(\n presentationId=presentation_id, body={\"requests\": requests}\n ).execute()\n file_url = f\"https://docs.google.com/presentation/d/{presentation_id}/edit\"\n\n elif app_type == \"docs\":\n from googleapiclient.discovery import build\n\n docs_service = build(\"docs\", \"v1\", credentials=credentials)\n\n file_metadata = {\n \"name\": self.file_name,\n \"mimeType\": \"application/vnd.google-apps.document\",\n \"parents\": [self.folder_id],\n }\n\n created_file = drive_service.files().create(body=file_metadata, fields=\"id\").execute()\n document_id = created_file[\"id\"]\n\n time.sleep(2) # Wait for file to be available # noqa: ASYNC251\n\n # Add content to document\n requests = [{\"insertText\": {\"location\": {\"index\": 1}, \"text\": content}}]\n docs_service.documents().batchUpdate(documentId=document_id, body={\"requests\": requests}).execute()\n file_url = f\"https://docs.google.com/document/d/{document_id}/edit\"\n\n return Message(text=f\"File successfully created in Google {app_type.title()}: {file_url}\")\n\n def _extract_content_for_upload(self) -> str:\n \"\"\"Extract content from input for upload to cloud services.\"\"\"\n if self._get_input_type() == \"DataFrame\":\n return self.input.to_csv(index=False)\n if self._get_input_type() == \"Data\":\n if hasattr(self.input, \"data\") and self.input.data:\n if isinstance(self.input.data, dict):\n import json\n\n return json.dumps(self.input.data, indent=2, ensure_ascii=False)\n return str(self.input.data)\n return str(self.input)\n if self._get_input_type() == \"Message\":\n return str(self.input.text) if self.input.text else str(self.input)\n return str(self.input)\n"
},
"file_format": {
"_input_type": "StrInput",
"advanced": false,
"display_name": "File Format (Tool)",
"dynamic": false,
"info": "Output format: 'csv', 'json', 'txt', 'html', 'excel', 'markdown'. Overrides pre-configured format.",
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "file_format",
"placeholder": "",
"required": false,
"show": false,
"title_case": false,
"tool_mode": true,
"trace_as_metadata": true,
"type": "str",
"value": ""
},
"file_name": {
"_input_type": "StrInput",
"advanced": false,
"display_name": "File Name",
"dynamic": false,
"info": "File name without extension (e.g. 'report'). Extension is added automatically.",
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "file_name",
"placeholder": "",
"required": true,
"show": true,
"title_case": false,
"tool_mode": false,
"trace_as_metadata": true,
"type": "str",
"value": "news-aggregated"
},
"folder_id": {
"_input_type": "StrInput",
"advanced": true,
"display_name": "Google Drive Folder ID",
"dynamic": false,
"info": "The Google Drive folder ID where the file will be uploaded. The folder must be shared with the service account email.",
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "folder_id",
"placeholder": "",
"required": true,
"show": false,
"title_case": false,
"tool_mode": false,
"trace_as_metadata": true,
"type": "str",
"value": ""
},
"gdrive_format": {
"_input_type": "DropdownInput",
"advanced": false,
"combobox": false,
"dialog_inputs": {},
"display_name": "File Format",
"dynamic": false,
"external_options": {},
"info": "Select the file format for Google Drive storage.",
"name": "gdrive_format",
"options": [
"txt",
"html",
"json",
"csv",
"xlsx",
"slides",
"docs",
"jpg",
"mp3"
],
"options_metadata": [],
"placeholder": "",
"required": false,
"show": false,
"title_case": false,
"toggle": false,
"tool_mode": false,
"trace_as_metadata": true,
"type": "str",
"value": "txt"
},
"input": {
"_input_type": "HandleInput",
"advanced": false,
"display_name": "File Content",
"dynamic": true,
"info": "The content to save. Accepts a DataFrame, Data, or Message object directly. Can also accept a JSON string (e.g. '[{\"col1\": \"val1\"}]') which will be parsed into a DataFrame, or plain text which will be saved as a Message.",
"input_types": [
"Data",
"JSON",
"DataFrame",
"Table",
"Message"
],
"list": false,
"list_add_label": "Add More",
"name": "input",
"placeholder": "",
"required": true,
"show": true,
"title_case": false,
"trace_as_metadata": true,
"type": "other",
"value": ""
},
"local_format": {
"_input_type": "DropdownInput",
"advanced": false,
"combobox": false,
"dialog_inputs": {},
"display_name": "File Format",
"dynamic": false,
"external_options": {},
"info": "Select the file format for local storage.",
"name": "local_format",
"options": [
"csv",
"excel",
"json",
"markdown",
"txt",
"html"
],
"options_metadata": [],
"placeholder": "",
"required": false,
"show": true,
"title_case": false,
"toggle": false,
"tool_mode": false,
"trace_as_metadata": true,
"type": "str",
"value": "json"
},
"s3_prefix": {
"_input_type": "StrInput",
"advanced": true,
"display_name": "S3 Prefix",
"dynamic": false,
"info": "Prefix for all files in S3.",
"list": false,
"list_add_label": "Add More",
"load_from_db": false,
"name": "s3_prefix",
"placeholder": "",
"required": false,
"show": false,
"title_case": false,
"tool_mode": false,
"trace_as_metadata": true,
"type": "str",
"value": ""
},
"service_account_key": {
"_input_type": "SecretStrInput",
"advanced": true,
"display_name": "GCP Credentials Secret Key",
"dynamic": false,
"info": "Your Google Cloud Platform service account JSON key as a secret string (complete JSON content).",
"input_types": [],
"load_from_db": false,
"name": "service_account_key",
"password": true,
"placeholder": "",
"required": true,
"show": false,
"title_case": false,
"type": "str",
"value": ""
},
"storage_location": {
"_input_type": "SortableListInput",
"advanced": false,
"display_name": "Storage Location",
"dynamic": false,
"info": "Choose where to save the file.",
"limit": 1,
"name": "storage_location",
"options": [
{
"icon": "hard-drive",
"name": "Local"
},
{
"icon": "Amazon",
"name": "AWS"
},
{
"icon": "google",
"name": "Google Drive"
}
],
"placeholder": "Select Location",
"real_time_refresh": true,
"required": false,
"search_category": [],
"show": true,
"title_case": false,
"tool_mode": false,
"trace_as_metadata": true,
"type": "sortableList",
"value": [
{
"chosen": false,
"icon": "hard-drive",
"name": "Local",
"selected": false
}
]
}
},
"tool_mode": false
},
"showNode": true,
"type": "SaveToFile"
},
"dragging": false,
"id": "SaveToFile-VVVVb",
"measured": {
"height": 512,
"width": 320
},
"position": {
"x": 2360,
"y": 360
},
"positionAbsolute": {
"x": 2360,
"y": 360
},
"selected": false,
"type": "genericNode"
},
{
"data": {
"id": "note-ca",
"node": {
"description": "# \ud83d\udcf0 Content Aggregator\n\nSearch the news on a topic, summarize each article, and compose a clean aggregated digest - then save it to a file.\n\n## Setup\n1. **Agent** and **Batch Run** - add your LLM key on each (any provider in the Model dropdown).\n2. **Web Search** - needs no key. Set its **Search Mode** to *News* (already selected); optionally set a **Topic** (Business, Technology, ...).\n3. **Save to File** - choose where to write the digest.\n\n## How it works\n1. **Chat Input -> Web Search (News)** - finds recent articles on your topic (returns title, link, date, summary).\n2. **Batch Run** - summarizes every article in one line.\n3. **Parser -> Agent** - the summaries are formatted and the Agent curates them into a clean digest.\n4. **Chat Output -> Save to File** - the digest is shown and saved.\n\n## Run\nType a topic (e.g. \"artificial intelligence\") and run.",
"display_name": "",
"documentation": "",
"i18n_key": "template_notes.content_aggregator.cbed07a7",
"template": {
"backgroundColor": "blue"
}
},
"type": "note"
},
"dragging": false,
"height": 959,
"id": "note-ca",
"measured": {
"height": 959,
"width": 628
},
"position": {
"x": -414.5709281961471,
"y": -53.99299474605954
},
"positionAbsolute": {
"x": 200,
"y": -130
},
"resizing": false,
"selected": false,
"type": "noteNode",
"width": 628
}
],
"viewport": {
"x": -148.0967741935483,
"y": 22.280645161290295,
"zoom": 0.46048387096774196
}
},
"description": "Aggregate recent content on a topic: Web Search (News) gathers articles, a Batch Run summarizes each one, and an Agent curates them into a clean digest that is shown and saved to a file.",
"endpoint_name": null,
"id": "0c580ed0-5bb0-4aa3-916b-af3cca4a34b4",
"is_component": false,
"last_tested_version": "1.10.0",
"locked": false,
"name": "Content Aggregator",
"tags": [
"web-scraping",
"agents"
]
}How to use this workflow
- Review the pre-filled example and source files above.
- Copy the template or download
Content Aggregator.json, then use it in your model, IDE, or workflow tool. - Compare your result with the sample output when one is provided.
Frequently Asked Questions
How do I use Workflow to Langflow Graph to Content Aggregator?
Open the workflow on Promptally, review the pre-filled example, then copy the template or download Content Aggregator.json and paste it into your model, IDE or workflow tool. Workflow to Langflow Graph to Content Aggregator is a static blueprint with no dynamic variables. Copy it and use it as-is.
Is Workflow to Langflow Graph to Content Aggregator free?
Yes. Workflow to Langflow Graph to Content Aggregator is free to view, copy and download on Promptally without creating an account. It is published under the MIT license.
What variables does Workflow to Langflow Graph to Content Aggregator need?
Workflow to Langflow Graph to Content Aggregator is a static blueprint with no dynamic variables. Copy it and use it as-is.