From 76ea0b9cde3215d53aaf575fad564f17c57fbc09 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?cls=5F=E5=AE=81=E6=B3=A2=E6=9C=AC=E6=9C=BA?= <908705107@qq.com> Date: Mon, 22 Jun 2026 09:06:17 +0800 Subject: [PATCH] feat: improve workflow run observability --- backend-fastapi/ai_platform/api/agent_api.py | 5 - .../src/api/ai-platform/ai-platform.ts | 2 + .../components/RunLogTimeline.vue | 160 +++++++++++++++--- .../workflow/shared/useRunReplay.ts | 51 +++++- 4 files changed, 182 insertions(+), 36 deletions(-) diff --git a/backend-fastapi/ai_platform/api/agent_api.py b/backend-fastapi/ai_platform/api/agent_api.py index afc47f3..444bede 100644 --- a/backend-fastapi/ai_platform/api/agent_api.py +++ b/backend-fastapi/ai_platform/api/agent_api.py @@ -437,11 +437,6 @@ async def publish_agent(agent_id: str, db: AsyncSession = Depends(get_db)): raise HTTPException(status_code=400, detail="模型不存在") if not model.is_active: raise HTTPException(status_code=400, detail="模型已禁用") - elif not await _resolve_default_chat_model_id(db): - raise HTTPException( - status_code=400, - detail="请先在模型配置中启用一个 chat 模型，或为智能体指定模型", - ) elif agent.mode == "dialog_flow": if not agent.workflow_id: raise HTTPException(status_code=400, detail="对话流模式需要配置工作流") diff --git a/web/apps/web-ele/src/api/ai-platform/ai-platform.ts b/web/apps/web-ele/src/api/ai-platform/ai-platform.ts index 4cba581..add4d66 100644 --- a/web/apps/web-ele/src/api/ai-platform/ai-platform.ts +++ b/web/apps/web-ele/src/api/ai-platform/ai-platform.ts @@ -280,6 +280,7 @@ export interface WorkflowRunEvent { } export interface ExecutionLogEntry { + type?: string; node_id: string; node_type: string; node_label?: string; @@ -292,6 +293,7 @@ export interface ExecutionLogEntry { timestamp?: string; branch?: string; branch_label?: string; + event?: WorkflowRunEvent; metadata?: Record; events?: WorkflowRunEvent[]; } diff --git a/web/apps/web-ele/src/views/ai-platform/workflow-runs/components/RunLogTimeline.vue b/web/apps/web-ele/src/views/ai-platform/workflow-runs/components/RunLogTimeline.vue index c99f60c..2a71af6 100644 --- a/web/apps/web-ele/src/views/ai-platform/workflow-runs/components/RunLogTimeline.vue +++ b/web/apps/web-ele/src/views/ai-platform/workflow-runs/components/RunLogTimeline.vue @@ -33,30 +33,75 @@ function formatDuration(ms?: number) { function stringifyBrief(value: any, maxLength = 180) { if (value === undefined || value === null || value === '') return ''; - const text = - typeof value === 'string' - ? value - : JSON.stringify(value, null, 2); + let text = ''; + if (typeof value === 'string') { + text = value; + } else { + try { + text = JSON.stringify(value, null, 2); + } catch { + text = String(value); + } + } return text.length > maxLength ? `${text.slice(0, maxLength)}...` : text; } function getCommunication(log: ExecutionLogEntry) { - return log.metadata?.communication || {}; + return log.metadata?.communication || log.event?.communication || {}; } function getMetaItems(log: ExecutionLogEntry) { const metadata = log.metadata || {}; - const items: Array<{ label: string; type?: 'info' | 'primary' | 'success' | 'warning' }> = []; + const items: Array<{ + label: string; + type?: 'danger' | 'info' | 'primary' | 'success' | 'warning'; + }> = []; const communication = getCommunication(log); + const collaboration = + metadata.collaboration || log.event?.collaboration || {}; + const branchLabel = + log.branch_label || + log.branch || + collaboration.branch_label || + collaboration.branch_id || + log.event?.branch_label || + log.event?.branch_id; + const subflowName = + metadata.subflow_name || + collaboration.subflow_name || + log.event?.subflow_name; + const providerName = + metadata.provider_name || + collaboration.provider_name || + log.event?.provider_name; + const modelName = + metadata.model || + metadata.model_name || + metadata.model_id || + collaboration.model || + collaboration.model_name || + collaboration.model_id || + log.event?.model || + log.event?.model_name || + log.event?.model_id; - if (log.branch_label || log.branch) { - items.push({ label: `分支: ${log.branch_label || log.branch}`, type: 'primary' }); + if (log.type || metadata.stream_event) { + items.push({ + label: getEventTypeLabel(log.type || log.event?.type), + type: log.status === 'failed' ? 'danger' : 'info', + }); } - if (metadata.subflow_name) { - items.push({ label: `子流程: ${metadata.subflow_name}`, type: 'success' }); + if (branchLabel) { + items.push({ label: `分支: ${branchLabel}`, type: 'primary' }); } - if (metadata.model || metadata.model_id) { - items.push({ label: `模型: ${metadata.model || metadata.model_id}`, type: 'info' }); + if (subflowName) { + items.push({ label: `子流程: ${subflowName}`, type: 'success' }); + } + if (providerName) { + items.push({ label: `提供商: ${providerName}`, type: 'info' }); + } + if (modelName) { + items.push({ label: `模型: ${modelName}`, type: 'info' }); } if (log.tokens_used) { items.push({ label: `Token ${log.tokens_used}`, type: 'info' }); @@ -99,34 +144,89 @@ function getEventTypeLabel(type?: string) { return labels[type || ''] || type || '事件'; } +function countItems(value: any) { + if (Array.isArray(value)) return value.length; + if (value && typeof value === 'object') return Object.keys(value).length; + return 0; +} + function getEventSummary(event: WorkflowRunEvent) { - if (event.content) return event.content; - if (event.message) return event.message; + const eventType = event.type || event.event?.type; + const waitingConfig = event.waiting_config || event.config; + if (event.error_message) return event.error_message; if (event.error) return event.error; + if (eventType === 'waiting_input') { + return stringifyBrief( + waitingConfig?.title || + waitingConfig?.question || + waitingConfig?.content || + waitingConfig || + '等待用户输入', + 160, + ); + } + if (eventType === 'parallel_start') { + return `启动 ${countItems(event.branches || event.branch_labels)} 个并行分支`; + } + if (eventType === 'parallel_complete') { + return `完成 ${countItems(event.branch_results)} 个并行分支`; + } + if (eventType === 'node_start') { + return `${event.node_label || event.node_id || '节点'} 开始执行`; + } + if (eventType === 'node_complete') { + return stringifyBrief( + event.outputs?.output || + event.outputs || + event.result || + `${event.node_label || event.node_id || '节点'} 执行完成`, + 160, + ); + } + if (eventType === 'complete') { + return stringifyBrief( + event.outputs?.output || event.outputs || '执行完成', + 160, + ); + } + if (event.content) return event.content; + if (event.message) return event.message; if (event.event) return getEventSummary(event.event); return stringifyBrief(event, 140); } function getLogEvents(log: ExecutionLogEntry): WorkflowRunEvent[] { + const primaryEvent = log.event ? [log.event] : []; const directEvents = Array.isArray(log.events) ? log.events : []; const metadataEvents = Array.isArray(log.metadata?.events) ? log.metadata.events : []; - return [...directEvents, ...metadataEvents].slice(0, 4); + return [...primaryEvent, ...directEvents, ...metadataEvents].filter(Boolean); +} + +function getVisibleEvents(log: ExecutionLogEntry) { + return getLogEvents(log).slice(0, 8); +} + +function getHiddenEventCount(log: ExecutionLogEntry) { + return Math.max(0, getLogEvents(log).length - 8); } function getLogDetail(log: ExecutionLogEntry) { const communication = getCommunication(log); if (log.error) return log.error; + if (log.event) return getEventSummary(log.event); if (log.status === 'waiting') { - return stringifyBrief(log.metadata?.waiting_config || log.metadata?.config || log.output); + return stringifyBrief( + log.metadata?.waiting_config || log.metadata?.config || log.output, + ); } return stringifyBrief( - log.output - || log.metadata?.branch_results - || log.metadata?.result - || communication.message, + log.output || + log.metadata?.branch_results || + log.metadata?.result || + communication.message, ); } @@ -147,7 +247,11 @@ function getLogDetail(log: ExecutionLogEntry) { diff --git a/web/apps/web-ele/src/views/ai-platform/workflow/shared/useRunReplay.ts b/web/apps/web-ele/src/views/ai-platform/workflow/shared/useRunReplay.ts index b95fce9..9cd85b2 100644 --- a/web/apps/web-ele/src/views/ai-platform/workflow/shared/useRunReplay.ts +++ b/web/apps/web-ele/src/views/ai-platform/workflow/shared/useRunReplay.ts @@ -1,6 +1,9 @@ import type { Edge, Node } from '@vue-flow/core'; -import type { ExecutionLogEntry, WorkflowRun } from '#/api/ai-platform/ai-platform'; +import type { + ExecutionLogEntry, + WorkflowRun, +} from '#/api/ai-platform/ai-platform'; function mapLogStatus( status: string, @@ -8,6 +11,8 @@ function mapLogStatus( if (status === 'completed') return 'success'; if (status === 'failed') return 'failed'; if (status === 'waiting') return 'running'; + if (status === 'running') return 'running'; + if (status === 'pending') return 'pending'; return 'success'; } @@ -17,7 +22,29 @@ function mapEdgeStatus( if (status === 'completed') return 'completed'; if (status === 'failed') return 'failed'; if (status === 'waiting') return 'running'; - return 'completed'; + if (status === 'running') return 'running'; + return undefined; +} + +function isStreamStartOnlyLog(log: ExecutionLogEntry) { + const metadata = log.metadata || {}; + return Boolean( + metadata.stream_event && + ['node_start', 'parallel_start', 'start'].includes(log.type || ''), + ); +} + +function shouldReplayLog(log: ExecutionLogEntry, logs: ExecutionLogEntry[]) { + if (!log?.node_id) return false; + + if (!isStreamStartOnlyLog(log)) return true; + + return !logs.some( + (item) => + item !== log && + item.node_id === log.node_id && + !isStreamStartOnlyLog(item), + ); } function resolveNodeOutput(log: ExecutionLogEntry, run: WorkflowRun) { @@ -95,7 +122,9 @@ export function applyRunReplay( edges: Edge[], run: WorkflowRun, ): { edges: Edge[]; nodes: Node[] } { - const nodeMap = new Map(nodes.map((node) => [node.id, { ...node, data: { ...node.data } }])); + const nodeMap = new Map( + nodes.map((node) => [node.id, { ...node, data: { ...node.data } }]), + ); const edgeList = edges.map((edge) => ({ ...edge, data: { ...(edge.data || {}), status: undefined as string | undefined }, @@ -103,18 +132,26 @@ export function applyRunReplay( const logs = (run.execution_log || []) as ExecutionLogEntry[]; const executedNodeIds: string[] = []; + const executedNodeLogs: ExecutionLogEntry[] = []; - logs.forEach((log, logIndex) => { + const replayLogs = logs.filter((log) => shouldReplayLog(log, logs)); + + replayLogs.forEach((log, logIndex) => { if (!log?.node_id) return; - executedNodeIds.push(log.node_id); const node = nodeMap.get(log.node_id); if (!node) return; + if (executedNodeIds[executedNodeIds.length - 1] !== log.node_id) { + executedNodeIds.push(log.node_id); + executedNodeLogs.push(log); + } else { + executedNodeLogs[executedNodeLogs.length - 1] = log; + } const isWaiting = log.status === 'waiting'; const isSuccess = log.status === 'completed'; const nodeOutput = resolveNodeOutput(log, run); - const nodeInputs = buildReplayInputs(logIndex, logs, run); + const nodeInputs = buildReplayInputs(logIndex, replayLogs, run); node.data = { ...node.data, @@ -155,7 +192,7 @@ export function applyRunReplay( for (let i = 1; i < executedNodeIds.length; i += 1) { const sourceId = executedNodeIds[i - 1]; const targetId = executedNodeIds[i]; - const log = logs[i]; + const log = executedNodeLogs[i]; const edgeStatus = mapEdgeStatus(log?.status || 'completed'); edgeList.forEach((edge, index) => {