From ada4c30f8540e319e671f8d403e41914a57561f5 Mon Sep 17 00:00:00 2001 From: Typer_Body Date: Wed, 6 May 2026 01:03:34 +0800 Subject: [PATCH] 1111 --- .../controller/groups/workflows/workflows.py | 13 + src/langbot/pkg/api/http/service/pipeline.py | 14 + src/langbot/pkg/api/http/service/workflow.py | 82 ++- src/langbot/pkg/workflow/executor.py | 61 +- src/langbot/pkg/workflow/node.py | 1 + .../pkg/workflow/nodes/call_pipeline.py | 193 ++++++- .../pkg/workflow/nodes/reply_message.py | 2 + .../dynamic-form/DynamicFormComponent.tsx | 100 ++-- .../dynamic-form/DynamicFormItemComponent.tsx | 130 ++++- .../WorkflowDebugDialog.tsx | 100 +++- .../node-configs/action-configs.ts | 146 +++-- .../WorkflowExecutionsTab.tsx | 521 ++++++++++++------ web/src/app/infra/entities/api/index.ts | 12 +- web/src/app/infra/entities/form/dynamic.ts | 1 + .../websocket/WorkflowWebSocketClient.ts | 58 +- web/src/i18n/locales/zh-Hans.ts | 10 +- 16 files changed, 1097 insertions(+), 347 deletions(-) diff --git a/src/langbot/pkg/api/http/controller/groups/workflows/workflows.py b/src/langbot/pkg/api/http/controller/groups/workflows/workflows.py index ec019fc7..7496f6ee 100644 --- a/src/langbot/pkg/api/http/controller/groups/workflows/workflows.py +++ b/src/langbot/pkg/api/http/controller/groups/workflows/workflows.py @@ -125,6 +125,19 @@ class WorkflowsRouterGroup(group.RouterGroup): ) return self.success(data=executions) + @self.route( + '//executions/', + methods=['GET'], + auth_type=group.AuthType.USER_TOKEN_OR_API_KEY, + ) + async def _(workflow_uuid: str, execution_uuid: str) -> str: + execution = await self.ap.workflow_service.get_execution(execution_uuid) + if execution is None: + return self.http_status(404, -1, 'execution not found') + if execution.get('workflow_uuid') != workflow_uuid: + return self.http_status(404, -1, 'execution not found in workflow') + return self.success(data={'execution': execution}) + # Get workflow versions @self.route('//versions', methods=['GET'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY) async def _(workflow_uuid: str) -> str: diff --git a/src/langbot/pkg/api/http/service/pipeline.py b/src/langbot/pkg/api/http/service/pipeline.py index ad75ffe7..6fae190a 100644 --- a/src/langbot/pkg/api/http/service/pipeline.py +++ b/src/langbot/pkg/api/http/service/pipeline.py @@ -73,6 +73,20 @@ class PipelineService: return self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline) + async def get_pipeline_by_name(self, pipeline_name: str) -> dict | None: + result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(persistence_pipeline.LegacyPipeline).where( + persistence_pipeline.LegacyPipeline.name == pipeline_name + ) + ) + + pipeline = result.first() + + if pipeline is None: + return None + + return self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline) + async def create_pipeline(self, pipeline_data: dict, default: bool = False) -> str: from ....utils import paths as path_utils diff --git a/src/langbot/pkg/api/http/service/workflow.py b/src/langbot/pkg/api/http/service/workflow.py index 24405467..869f1c25 100644 --- a/src/langbot/pkg/api/http/service/workflow.py +++ b/src/langbot/pkg/api/http/service/workflow.py @@ -284,7 +284,7 @@ class WorkflowService: 'uuid': execution_uuid, 'workflow_uuid': workflow_uuid, 'workflow_version': workflow_dict.get('version', 1), - 'status': ExecutionStatus.PENDING.value, + 'status': ExecutionStatus.RUNNING.value, 'trigger_type': trigger_type, 'trigger_data': trigger_data or {}, 'variables': {}, @@ -496,13 +496,7 @@ class WorkflowService: executions = result.all() return { - 'executions': [ - self.ap.persistence_mgr.serialize_model( - persistence_workflow.WorkflowExecution, - execution - ) - for execution in executions - ], + 'executions': [self._serialize_execution(execution) for execution in executions], 'total': total, } @@ -519,10 +513,17 @@ class WorkflowService: if execution is None: return None - return self.ap.persistence_mgr.serialize_model( - persistence_workflow.WorkflowExecution, - execution - ) + data = self._serialize_execution(execution) + + node_exec_query = sqlalchemy.select(persistence_workflow.WorkflowNodeExecution).where( + persistence_workflow.WorkflowNodeExecution.execution_uuid == execution_uuid + ).order_by(persistence_workflow.WorkflowNodeExecution.id.asc()) + node_exec_result = await self.ap.persistence_mgr.execute_async(node_exec_query) + node_executions = node_exec_result.all() + data['node_executions'] = [ + self._serialize_node_execution(node_exec) for node_exec in node_executions + ] + return data async def get_node_types(self) -> list[dict]: """Get all available node types""" @@ -837,6 +838,24 @@ class WorkflowService: result['variables'] = {} return result + + def _serialize_execution(self, execution) -> dict: + data = self.ap.persistence_mgr.serialize_model( + persistence_workflow.WorkflowExecution, + execution, + ) + data['started_at'] = data.get('start_time') + data['completed_at'] = data.get('end_time') + return data + + def _serialize_node_execution(self, node_execution) -> dict: + data = self.ap.persistence_mgr.serialize_model( + persistence_workflow.WorkflowNodeExecution, + node_execution, + ) + data['started_at'] = data.get('start_time') + data['completed_at'] = data.get('end_time') + return data async def update_workflow_extensions( self, @@ -1111,22 +1130,41 @@ class WorkflowService: execution = await self.get_execution(execution_uuid) if execution is None: raise ValueError(f'Execution {execution_uuid} not found') - + if execution.get('workflow_uuid') != workflow_uuid: + raise ValueError(f'Execution {execution_uuid} not found in workflow {workflow_uuid}') + query = sqlalchemy.select(persistence_workflow.WorkflowNodeExecution).where( persistence_workflow.WorkflowNodeExecution.execution_uuid == execution_uuid ).order_by( persistence_workflow.WorkflowNodeExecution.id.asc() ).limit(limit).offset(offset) - + result = await self.ap.persistence_mgr.execute_async(query) node_executions = result.all() - - logs = [ - self.ap.persistence_mgr.serialize_model( - persistence_workflow.WorkflowNodeExecution, - node_exec + + logs = [] + for node_exec in node_executions: + serialized = self._serialize_node_execution(node_exec) + timestamp = serialized.get('completed_at') or serialized.get('started_at') or execution.get('started_at') + level = 'error' if serialized.get('status') == 'failed' else 'info' + message = ( + f"{serialized.get('node_type')}::{serialized.get('node_id')} - {serialized.get('status')}" ) - for node_exec in node_executions - ] - + if serialized.get('error'): + message = f"{message} - {serialized.get('error')}" + logs.append( + { + 'id': str(serialized.get('id', serialized.get('node_id'))), + 'timestamp': timestamp, + 'level': level, + 'node_id': serialized.get('node_id'), + 'message': message, + 'data': { + 'inputs': serialized.get('inputs'), + 'outputs': serialized.get('outputs'), + 'retry_count': serialized.get('retry_count'), + }, + } + ) + return {'logs': logs, 'total': len(logs)} diff --git a/src/langbot/pkg/workflow/executor.py b/src/langbot/pkg/workflow/executor.py index e7d63f81..9250c44d 100644 --- a/src/langbot/pkg/workflow/executor.py +++ b/src/langbot/pkg/workflow/executor.py @@ -10,6 +10,8 @@ import uuid from datetime import datetime from typing import Any, Optional, TYPE_CHECKING +import sqlalchemy + from .entities import ( WorkflowDefinition, NodeDefinition, @@ -20,6 +22,7 @@ from .entities import ( NodeStatus, ExecutionStep, ) +from ..entity.persistence import workflow as persistence_workflow from .registry import NodeTypeRegistry if TYPE_CHECKING: @@ -346,6 +349,8 @@ class WorkflowExecutor: logger.warning(f"Circular dependency detected at node: {node.id}") context.node_states[node.id].status = NodeStatus.SKIPPED context.node_states[node.id].error = "Circular dependency detected" + context.node_states[node.id].end_time = datetime.now() + await self._persist_node_execution(node, context.node_states[node.id], context) return # Add node to current path @@ -353,7 +358,10 @@ class WorkflowExecutor: # Check if node should be skipped if await self._should_skip_node(node, context): - context.node_states[node.id].status = NodeStatus.SKIPPED + existing_state = context.node_states[node.id] + if existing_state.status == NodeStatus.SKIPPED: + existing_state.end_time = existing_state.end_time or datetime.now() + await self._persist_node_execution(node, existing_state, context) path.discard(node.id) return @@ -469,6 +477,7 @@ class WorkflowExecutor: node_state.error = f"Unknown node type: {node.type}" node_state.end_time = datetime.now() self._record_execution_step(node, node_state, context) + await self._persist_node_execution(node, node_state, context) return # Resolve inputs @@ -482,6 +491,7 @@ class WorkflowExecutor: node_state.error = "; ".join(validation_errors) node_state.end_time = datetime.now() self._record_execution_step(node, node_state, context) + await self._persist_node_execution(node, node_state, context) return # Execute with retries @@ -523,6 +533,7 @@ class WorkflowExecutor: ) self._record_execution_step(node, node_state, context) + await self._persist_node_execution(node, node_state, context) async def _resolve_inputs( self, @@ -738,6 +749,47 @@ class WorkflowExecutor: ) context.history.append(step) + async def _persist_node_execution( + self, + node: NodeDefinition, + node_state: NodeState, + context: ExecutionContext, + ): + """Persist node execution state for execution detail and logs.""" + if not self.ap: + return + + values = { + 'execution_uuid': context.execution_id, + 'node_id': node.id, + 'node_type': node.type, + 'status': node_state.status.value, + 'inputs': node_state.inputs, + 'outputs': node_state.outputs, + 'start_time': node_state.start_time, + 'end_time': node_state.end_time, + 'error': node_state.error, + 'retry_count': node_state.retry_count, + } + + existing_query = sqlalchemy.select(persistence_workflow.WorkflowNodeExecution).where( + persistence_workflow.WorkflowNodeExecution.execution_uuid == context.execution_id, + persistence_workflow.WorkflowNodeExecution.node_id == node.id, + ) + existing_result = await self.ap.persistence_mgr.execute_async(existing_query) + existing = existing_result.first() + + if existing is None: + await self.ap.persistence_mgr.execute_async( + sqlalchemy.insert(persistence_workflow.WorkflowNodeExecution).values(**values) + ) + else: + await self.ap.persistence_mgr.execute_async( + sqlalchemy.update(persistence_workflow.WorkflowNodeExecution) + .where(persistence_workflow.WorkflowNodeExecution.id == existing.id) + .values(**values) + ) + class ParallelExecutor: """Execute multiple branches in parallel""" @@ -997,8 +1049,8 @@ class DebugWorkflowExecutor(WorkflowExecutor): # Check if should skip if await self._should_skip_node(node, context): - context.node_states[node.id].status = NodeStatus.SKIPPED - debug_state.add_log('info', f'Skipping node: {node.id}', node_id=node.id) + if context.node_states[node.id].status == NodeStatus.SKIPPED: + debug_state.add_log('info', f'Skipping node: {node.id}', node_id=node.id) return # Check breakpoint @@ -1076,6 +1128,7 @@ class DebugWorkflowExecutor(WorkflowExecutor): node_state.end_time = datetime.now() debug_state.add_log('error', f'Unknown node type: {node.type}', node_id=node.id) self._record_execution_step(node, node_state, context) + await self._persist_node_execution(node, node_state, context) return # Resolve inputs @@ -1100,6 +1153,7 @@ class DebugWorkflowExecutor(WorkflowExecutor): node_id=node.id ) self._record_execution_step(node, node_state, context) + await self._persist_node_execution(node, node_state, context) return # Execute with retries @@ -1147,6 +1201,7 @@ class DebugWorkflowExecutor(WorkflowExecutor): ) self._record_execution_step(node, node_state, context) + await self._persist_node_execution(node, node_state, context) async def step_execute( self, diff --git a/src/langbot/pkg/workflow/node.py b/src/langbot/pkg/workflow/node.py index 717d8d06..fe22bccf 100644 --- a/src/langbot/pkg/workflow/node.py +++ b/src/langbot/pkg/workflow/node.py @@ -148,6 +148,7 @@ class WorkflowNode(abc.ABC): 'llm-model-selector': 'llm-model-selector', 'embedding-model-selector': 'embedding-model-selector', 'rerank-model-selector': 'rerank-model-selector', + 'pipeline-selector': 'pipeline-selector', 'knowledge-base-selector': 'knowledge-base-selector', 'knowledge-base-multi-selector': 'knowledge-base-multi-selector', 'bot-selector': 'bot-selector', diff --git a/src/langbot/pkg/workflow/nodes/call_pipeline.py b/src/langbot/pkg/workflow/nodes/call_pipeline.py index 6fc41675..8829564a 100644 --- a/src/langbot/pkg/workflow/nodes/call_pipeline.py +++ b/src/langbot/pkg/workflow/nodes/call_pipeline.py @@ -7,6 +7,13 @@ from __future__ import annotations from typing import Any, ClassVar +import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter +import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query +import langbot_plugin.api.entities.builtin.platform.entities as platform_entities +import langbot_plugin.api.entities.builtin.platform.events as platform_events +import langbot_plugin.api.entities.builtin.platform.message as platform_message +import langbot_plugin.api.entities.builtin.provider.session as provider_session + from ..entities import ExecutionContext from ..node import WorkflowNode, workflow_node, NodePort, NodeConfig @@ -30,7 +37,187 @@ class CallPipelineNode(WorkflowNode): config_schema: ClassVar[list[NodeConfig]] = [] async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]: - query = inputs.get("query", "") - pipeline_uuid = self.get_config("pipeline_uuid", "") + if not self.ap: + raise RuntimeError('Application instance not available — cannot call pipeline') - return {"response": f"[Pipeline {pipeline_uuid} response for: {query[:50]}...]", "result": {}} + raw_query = inputs.get('query', '') + query_text = str(raw_query or inputs.get('input') or '') + pipeline_ref = str(self.get_config('pipeline_uuid', '') or '').strip() + + if not pipeline_ref: + raise ValueError('No pipeline configured for call pipeline node') + + pipeline_data = await self.ap.pipeline_service.get_pipeline(pipeline_ref) + if pipeline_data is None: + pipeline_data = await self.ap.pipeline_service.get_pipeline_by_name(pipeline_ref) + if pipeline_data is None: + raise ValueError(f'Pipeline not found: {pipeline_ref}') + + pipeline_uuid = str(pipeline_data.get('uuid', '') or '') + if not pipeline_uuid: + raise ValueError(f'Pipeline UUID missing for: {pipeline_ref}') + + runtime_pipeline = await self.ap.pipeline_mgr.get_pipeline_by_uuid(pipeline_uuid) + if runtime_pipeline is None: + raise ValueError(f'Runtime pipeline not loaded: {pipeline_uuid}') + + adapter = _WorkflowPipelineCaptureAdapter(context=context) + adapter.bot_account_id = 'workflow-call-pipeline' + + message_event = self._build_message_event(query_text, context) + message_chain = message_event.message_chain + launcher_type = provider_session.LauncherTypes.GROUP if context.message_context and context.message_context.is_group else provider_session.LauncherTypes.PERSON + launcher_id = context.session_id or context.execution_id + sender_id = ( + context.message_context.sender_id + if context.message_context and context.message_context.sender_id + else context.user_id or f'workflow_{context.execution_id}' + ) + + query = pipeline_query.Query( + bot_uuid=context.bot_id, + query_id=-1, + launcher_type=launcher_type, + launcher_id=launcher_id, + sender_id=sender_id, + message_event=message_event, + message_chain=message_chain, + variables={ + '_called_from_workflow': True, + '_workflow_execution_id': context.execution_id, + '_workflow_id': context.workflow_id, + **dict(context.variables or {}), + }, + resp_messages=[], + resp_message_chain=[], + adapter=adapter, + pipeline_uuid=pipeline_uuid, + ) + + await runtime_pipeline.run(query) + + response_text = adapter.get_last_text_response() + result = { + 'pipeline_uuid': pipeline_uuid, + 'pipeline_name': pipeline_data.get('name', ''), + 'responses': adapter.responses, + 'query_text': query_text, + } + + return {'response': response_text, 'result': result} + + def _build_message_event( + self, + query_text: str, + context: ExecutionContext, + ) -> platform_events.MessageEvent: + message_chain_data = context.trigger_data.get('message_chain') or context.trigger_data.get('message', []) + if isinstance(message_chain_data, list) and message_chain_data: + message_chain = platform_message.MessageChain.model_validate(message_chain_data) + else: + message_chain = platform_message.MessageChain([platform_message.Plain(text=query_text)]) + + if context.message_context and context.message_context.is_group: + group = platform_entities.Group( + id=context.message_context.group_id or context.session_id or 'workflow_group', + name='Workflow Group', + permission=platform_entities.Permission.Member, + ) + sender = platform_entities.GroupMember( + id=context.message_context.sender_id, + member_name=context.message_context.sender_name or 'Workflow User', + permission=platform_entities.Permission.Member, + group=group, + ) + return platform_events.GroupMessage( + sender=sender, + message_chain=message_chain, + time=context.message_context.raw_message.get('time') if context.message_context.raw_message else None, + ) + + sender = platform_entities.Friend( + id=context.message_context.sender_id if context.message_context else context.user_id or 'workflow_user', + nickname=context.message_context.sender_name if context.message_context else 'Workflow User', + remark=context.message_context.sender_name if context.message_context else 'Workflow User', + ) + return platform_events.FriendMessage( + sender=sender, + message_chain=message_chain, + time=context.message_context.raw_message.get('time') if context.message_context and context.message_context.raw_message else None, + ) + + +class _WorkflowPipelineCaptureAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): + responses: list[dict[str, Any]] = [] + + def __init__(self, context: ExecutionContext): + super().__init__(config={}, logger=None) + self.context = context + self.responses = [] + + async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain): + payload = { + 'type': 'send', + 'target_type': target_type, + 'target_id': target_id, + 'content': str(message), + 'message_chain': message.model_dump(), + } + self.responses.append(payload) + return payload + + async def reply_message( + self, + message_source: platform_events.MessageEvent, + message: platform_message.MessageChain, + quote_origin: bool = False, + ): + payload = { + 'type': 'reply', + 'content': str(message), + 'message_chain': message.model_dump(), + 'quote_origin': quote_origin, + } + self.responses.append(payload) + return payload + + async def reply_message_chunk( + self, + message_source: platform_events.MessageEvent, + bot_message: dict, + message: platform_message.MessageChain, + quote_origin: bool = False, + is_final: bool = False, + ): + payload = { + 'type': 'reply_chunk', + 'content': str(message), + 'message_chain': message.model_dump(), + 'quote_origin': quote_origin, + 'is_final': is_final, + } + self.responses.append(payload) + return payload + + async def create_message_card(self, message_id, event: platform_events.MessageEvent) -> bool: + return False + + def register_listener(self, event_type, callback): + return None + + def unregister_listener(self, event_type, callback): + return None + + async def run_async(self): + return None + + async def is_stream_output_supported(self) -> bool: + return False + + async def kill(self) -> bool: + return True + + def get_last_text_response(self) -> str: + if not self.responses: + return '' + return str(self.responses[-1].get('content', '') or '') diff --git a/src/langbot/pkg/workflow/nodes/reply_message.py b/src/langbot/pkg/workflow/nodes/reply_message.py index 5ae52320..091e77e5 100644 --- a/src/langbot/pkg/workflow/nodes/reply_message.py +++ b/src/langbot/pkg/workflow/nodes/reply_message.py @@ -38,6 +38,8 @@ class ReplyMessageNode(WorkflowNode): message = inputs.get("input") if message in (None, ""): message = inputs.get("response") + if message in (None, ""): + message = inputs.get("content") if message in (None, "") and context.message_context: message = context.message_context.message_content if message is None: diff --git a/web/src/app/home/components/dynamic-form/DynamicFormComponent.tsx b/web/src/app/home/components/dynamic-form/DynamicFormComponent.tsx index 2e7ccea9..9a184816 100644 --- a/web/src/app/home/components/dynamic-form/DynamicFormComponent.tsx +++ b/web/src/app/home/components/dynamic-form/DynamicFormComponent.tsx @@ -14,7 +14,10 @@ import DynamicFormItemComponent from '@/app/home/components/dynamic-form/Dynamic import { useEffect, useMemo, useRef, useState } from 'react'; import { useTranslation } from 'react-i18next'; import i18n from 'i18next'; -import { resolveI18nLabel, maybeTranslateKey } from '@/app/home/workflows/components/workflow-editor/workflow-i18n'; +import { + resolveI18nLabel, + maybeTranslateKey, +} from '@/app/home/workflows/components/workflow-editor/workflow-i18n'; // Helper function to translate i18n key if the value is an i18n key string const translateIfKey = (value: string | undefined): string | undefined => { @@ -228,6 +231,7 @@ export default function DynamicFormComponent({ item.type === 'llm-model-selector' || item.type === 'embedding-model-selector' || item.type === 'rerank-model-selector' || + item.type === 'pipeline-selector' || item.type === 'knowledge-base-selector' || item.type === 'bot-selector' ) { @@ -286,6 +290,9 @@ export default function DynamicFormComponent({ case 'select': fieldSchema = z.string(); break; + case 'pipeline-selector': + fieldSchema = z.string(); + break; case 'llm-model-selector': fieldSchema = z.string(); break; @@ -490,9 +497,11 @@ export default function DynamicFormComponent({ label={extractAndTranslateI18n(config.label)} description={ config.description - ? (typeof config.description === 'string' - ? (config.description.startsWith('workflows.') ? String(t(config.description)) : config.description) - : extractAndTranslateI18n(config.description)) + ? typeof config.description === 'string' + ? config.description.startsWith('workflows.') + ? String(t(config.description)) + : config.description + : extractAndTranslateI18n(config.description) : undefined } url={webhookUrl} @@ -523,7 +532,9 @@ export default function DynamicFormComponent({ {config.description && (

{typeof config.description === 'string' - ? (config.description.startsWith('workflows.') ? String(t(config.description)) : translateIfKey(config.description)) + ? config.description.startsWith('workflows.') + ? String(t(config.description)) + : translateIfKey(config.description) : extractAndTranslateI18n(config.description)}

)} @@ -554,37 +565,54 @@ export default function DynamicFormComponent({ ? extractAndTranslateI18n(config.label) : config.name; return ( - - - {i18nLabel}{' '} - {config.required && *} - - -
- -
- - {config.description && (() => { - const desc = config.description; - if (typeof desc === 'string') { - if (desc.startsWith('workflows.')) { - return

{String(t(desc))}

; - } - return

{translateIfKey(desc) || desc}

; - } - return

{extractAndTranslateI18n(desc)}

; - })()} - - - ); + + + {i18nLabel}{' '} + {config.required && ( + * + )} + + +
+ +
+ + {config.description && + (() => { + const desc = config.description; + if (typeof desc === 'string') { + if (desc.startsWith('workflows.')) { + return ( +

+ {String(t(desc))} +

+ ); + } + return ( +

+ {translateIfKey(desc) || desc} +

+ ); + } + return ( +

+ {extractAndTranslateI18n(desc)} +

+ ); + })()} + + + ); }} /> ); diff --git a/web/src/app/home/components/dynamic-form/DynamicFormItemComponent.tsx b/web/src/app/home/components/dynamic-form/DynamicFormItemComponent.tsx index d6488653..ec7ffd66 100644 --- a/web/src/app/home/components/dynamic-form/DynamicFormItemComponent.tsx +++ b/web/src/app/home/components/dynamic-form/DynamicFormItemComponent.tsx @@ -25,11 +25,15 @@ import { KnowledgeBase, EmbeddingModel, RerankModel, + Pipeline, PluginTool, } from '@/app/infra/entities/api'; import { toast } from 'sonner'; import { useTranslation } from 'react-i18next'; -import { resolveI18nLabel, maybeTranslateKey } from '@/app/home/workflows/components/workflow-editor/workflow-i18n'; +import { + resolveI18nLabel, + maybeTranslateKey, +} from '@/app/home/workflows/components/workflow-editor/workflow-i18n'; import { Textarea } from '@/components/ui/textarea'; import { Card, CardContent } from '@/components/ui/card'; import { @@ -65,12 +69,11 @@ import { } from '@/components/ui/dropdown-menu'; import ModelsDialog from '@/app/home/components/models-dialog/ModelsDialog'; -const resolveOptionLabel = ( - label: unknown, - fallback: string, -): string => { +const resolveOptionLabel = (label: unknown, fallback: string): string => { if (!label || typeof label !== 'object') return fallback; - return resolveI18nLabel(label as Record | I18nObject) || fallback; + return ( + resolveI18nLabel(label as Record | I18nObject) || fallback + ); }; const getSelectedOptionLabel = ( @@ -87,7 +90,11 @@ const resolveModelLabel = (model: { name: string; display_name?: string; }): string => { - return maybeTranslateKey(model.display_name || model.name) || model.display_name || model.name; + return ( + maybeTranslateKey(model.display_name || model.name) || + model.display_name || + model.name + ); }; export default function DynamicFormItemComponent({ @@ -105,6 +112,7 @@ export default function DynamicFormItemComponent({ const [rerankModels, setRerankModels] = useState([]); const [knowledgeBases, setKnowledgeBases] = useState([]); const [bots, setBots] = useState([]); + const [pipelines, setPipelines] = useState([]); const [tools, setTools] = useState([]); const [uploading, setUploading] = useState(false); const [kbDialogOpen, setKbDialogOpen] = useState(false); @@ -258,6 +266,19 @@ export default function DynamicFormItemComponent({ } }, [config.type]); + useEffect(() => { + if (config.type === DynamicFormItemType.PIPELINE_SELECTOR) { + httpClient + .getPipelines() + .then((resp) => { + setPipelines(resp.pipelines); + }) + .catch((err) => { + toast.error(t('pipelines.loadPipelinesFailed') + err.msg); + }); + } + }, [config.type, t]); + useEffect(() => { if (config.type === DynamicFormItemType.TOOLS_SELECTOR) { httpClient @@ -308,7 +329,9 @@ export default function DynamicFormItemComponent({ onClick={() => field.onChange(option.name)} >
- {resolveOptionLabel(option.label, option.name)} + + {resolveOptionLabel(option.label, option.name)} + {option.name} @@ -320,7 +343,9 @@ export default function DynamicFormItemComponent({
); } - return ; + return ( + + ); case DynamicFormItemType.SECRET: const secretValue = typeof field.value === 'string' ? field.value : ''; @@ -346,16 +371,23 @@ export default function DynamicFormItemComponent({ onMouseDown={(e) => e.preventDefault()} onClick={() => setSecretVisible((prev) => !prev)} > - {secretVisible ? : } + {secretVisible ? ( + + ) : ( + + )} ); case DynamicFormItemType.TEXT: // Ensure value is always a string to avoid [object Object] display - const textValue = typeof field.value === 'string' - ? field.value - : (field.value != null ? JSON.stringify(field.value, null, 2) : ''); + const textValue = + typeof field.value === 'string' + ? field.value + : field.value != null + ? JSON.stringify(field.value, null, 2) + : ''; return (