feat: improve workflow run observability

This commit is contained in:
2026-06-22 03:02:08 +08:00
parent 1dd7805712
commit ae29c24529
3 changed files with 349 additions and 4 deletions
@@ -211,6 +211,14 @@ def _resolve_handoff_target(
return next_nodes[0] if next_nodes else None return next_nodes[0] if next_nodes else None
def _execution_log_event_type(status: Optional[str]) -> str:
if status == 'waiting':
return 'waiting_input'
if status == 'failed':
return 'error'
return 'node_complete'
def _make_execution_log_entry( def _make_execution_log_entry(
node_map: Dict[str, Any], node_map: Dict[str, Any],
node_id: str, node_id: str,
@@ -219,6 +227,7 @@ def _make_execution_log_entry(
) -> dict: ) -> dict:
extra.setdefault('timestamp', datetime.now().isoformat()) extra.setdefault('timestamp', datetime.now().isoformat())
metadata = dict(extra.pop('metadata', {}) or {}) metadata = dict(extra.pop('metadata', {}) or {})
event_type = _execution_log_event_type(extra.get('status'))
collaboration = _node_collaboration_metadata(node_map, node_id, node_type) collaboration = _node_collaboration_metadata(node_map, node_id, node_type)
branch_id = extra.get('branch') or extra.get('branch_id') branch_id = extra.get('branch') or extra.get('branch_id')
branch_label = extra.get('branch_label') branch_label = extra.get('branch_label')
@@ -231,7 +240,7 @@ def _make_execution_log_entry(
node_map, node_map,
node_id, node_id,
node_type, node_type,
event_type='node_complete', event_type=event_type,
status=extra.get('status'), status=extra.get('status'),
branch_id=branch_id, branch_id=branch_id,
branch_label=branch_label, branch_label=branch_label,
@@ -2508,6 +2517,13 @@ class AIWorkflowService:
start_time = time.time() start_time = time.time()
result = await node_instance.execute_async(context) result = await node_instance.execute_async(context)
elapsed = int((time.time() - start_time) * 1000) elapsed = int((time.time() - start_time) * 1000)
target_node_id = _resolve_handoff_target(
current_node_id,
result,
edge_map,
parallel_edge_map,
node_map,
)
# 处理节点事件 # 处理节点事件
if result.events: if result.events:
@@ -226,6 +226,49 @@ export function useEventHandler(config: EventHandlerConfig) {
!event.error && !event.error &&
(!event.status || event.status === 'success' || event.status === 'completed'); (!event.status || event.status === 'success' || event.status === 'completed');
const getEventDisplayName = (event: StreamEvent) => {
const typeMap: Record<string, string> = {
complete: '执行完成',
loop_start: '循环开始',
node_event: '节点事件',
resume: '恢复执行',
start: '开始执行',
};
return (
event.node_label ||
event.event?.message ||
event.event?.content ||
event.event?.type ||
event.message ||
event.content ||
typeMap[event.type] ||
event.type ||
'运行事件'
);
};
const appendObservedStep = (
event: StreamEvent,
msgId: string,
status: 'completed' | 'failed' | 'running' = 'completed',
) => {
currentSteps.value.push({
type: status === 'failed' ? 'error' : 'observation',
content: getEventDisplayName(event),
node_id: event.node_id,
node_type: event.node_type,
...buildReasoningMeta(event),
params: event.config || event.waiting_config || event.event || event.params,
output: event.outputs || event.output || event.result,
status,
timestamp: new Date().toISOString(),
});
updateAssistantMessage(msgId, {
reasoning_steps: [...currentSteps.value],
...buildMessageMeta(event),
});
};
/** 处理流式事件 */ /** 处理流式事件 */
const handleStreamEvent = (event: StreamEvent, msgId: string) => { const handleStreamEvent = (event: StreamEvent, msgId: string) => {
switch (event.type) { switch (event.type) {
@@ -600,7 +643,11 @@ export function useEventHandler(config: EventHandlerConfig) {
(s) => s.node_id === event.node_id && s.type === 'parallel_start', (s) => s.node_id === event.node_id && s.type === 'parallel_start',
); );
const branchResults = event.branch_results || {}; const branchResults = event.branch_results || {};
const content = `并行分支完成:${getItemCount(branchResults)} 个分支`; const nodeSucceeded = isSuccessEvent(event);
const content = nodeSucceeded
? `并行分支完成:${getItemCount(branchResults)} 个分支`
: `并行分支失败:${getStreamErrorMessage(event)}`;
const nodeStatus = nodeSucceeded ? 'completed' : 'failed';
if (existingIndex === -1) { if (existingIndex === -1) {
currentSteps.value.push({ currentSteps.value.push({
type: 'parallel_complete', type: 'parallel_complete',
@@ -612,7 +659,7 @@ export function useEventHandler(config: EventHandlerConfig) {
branch_results: branchResults, branch_results: branchResults,
total_tokens: event.total_tokens, total_tokens: event.total_tokens,
}, },
status: 'completed', status: nodeStatus,
timestamp: new Date().toISOString(), timestamp: new Date().toISOString(),
}); });
} else { } else {
@@ -625,7 +672,7 @@ export function useEventHandler(config: EventHandlerConfig) {
branch_results: branchResults, branch_results: branchResults,
total_tokens: event.total_tokens, total_tokens: event.total_tokens,
}, },
status: 'completed', status: nodeStatus,
}; };
} }
updateAssistantMessage(msgId, { updateAssistantMessage(msgId, {
@@ -673,6 +720,8 @@ export function useEventHandler(config: EventHandlerConfig) {
}; };
messages.value.push(newMsg); messages.value.push(newMsg);
} }
} else if (nodeEvent) {
appendObservedStep(event, msgId);
} }
break; break;
} }
@@ -719,6 +768,12 @@ export function useEventHandler(config: EventHandlerConfig) {
break; break;
} }
case 'resume': {
currentRunId.value = event.run_id || currentRunId.value;
appendObservedStep(event, msgId, 'running');
break;
}
// ========== 对话流交互事件 ========== // ========== 对话流交互事件 ==========
case 'waiting_input': { case 'waiting_input': {
waitingForInput.value = true; waitingForInput.value = true;
@@ -752,6 +807,24 @@ export function useEventHandler(config: EventHandlerConfig) {
running.value = false; running.value = false;
break; break;
} }
default: {
if (
event.type &&
(event.content ||
event.message ||
event.error ||
event.error_message ||
event.event)
) {
appendObservedStep(
event,
msgId,
event.error || event.error_message ? 'failed' : 'completed',
);
}
break;
}
} }
}; };
@@ -95,6 +95,115 @@ const selectedNodeLabel = computed(() => {
const log = executionLogs.value.find((item) => item.node_id === nodeId); const log = executionLogs.value.find((item) => item.node_id === nodeId);
return log?.node_label || log?.node_id || nodeId; return log?.node_label || log?.node_id || nodeId;
}); });
const runEventTimeline = computed(() => {
if (!run.value) return [];
const items: Array<{
channel: string;
elapsed?: number;
eventType: string;
index?: number;
key: string;
nodeId?: string;
status: string;
summary: string;
timestamp?: string;
title: string;
}> = [
{
channel: triggerLabel(run.value.trigger_type),
eventType: 'start',
key: 'start',
status: run.value.status === 'running' ? 'running' : 'completed',
summary: previewValue(run.value.inputs),
timestamp: run.value.started_at,
title: '开始运行',
},
];
let estimatedOffset = 0;
for (const [index, log] of executionLogs.value.entries()) {
const communication = getCommunication(log);
const collaboration = getCollaboration(log);
const eventType =
communication.event ||
(log.status === 'waiting'
? 'waiting_input'
: log.status === 'failed'
? 'error'
: 'node_complete');
const branch =
communication.branch_label ||
log.branch_label ||
log.branch ||
collaboration.branch_label;
const subflowName = collaboration.subflow_name || getLogMeta(log).subflow_name;
const targetName = getTargetName(log);
const timestamp =
log.timestamp ||
communication.timestamp ||
estimateLogTimestamp(estimatedOffset);
estimatedOffset += Number(log.elapsed_time || 0);
const summaryParts = [
branch ? `分支:${branch}` : '',
subflowName ? `子流程:${subflowName}` : '',
targetName ? `交接:${targetName}` : '',
communication.message ? `摘要:${communication.message}` : '',
getLogError(log) ? `错误:${getLogError(log)}` : '',
].filter(Boolean);
items.push({
channel: communication.channel || collaboration.collaboration_mode || log.node_type || '',
elapsed: log.elapsed_time,
eventType,
index,
key: `${log.node_id || 'event'}-${index}-${eventType}`,
nodeId: log.node_id,
status: normalizeRunEventStatus(eventType, log.status),
summary: summaryParts.join(' · ') || previewValue(log.output),
timestamp,
title:
log.node_label ||
getActorName(log) ||
getRunEventTypeLabel(eventType),
});
}
if (run.value.status === 'waiting') {
items.push({
channel: 'human_input',
eventType: 'waiting_input',
key: 'waiting',
nodeId: run.value.current_node_id,
status: 'waiting',
summary: previewValue(run.value.waiting_config),
title: '等待用户输入',
});
} else if (run.value.status === 'failed') {
items.push({
channel: 'error',
eventType: 'error',
key: 'failed',
status: 'failed',
summary: run.value.error_message || '工作流执行失败',
timestamp: run.value.completed_at,
title: '运行失败',
});
} else if (run.value.status === 'completed') {
items.push({
channel: 'workflow',
elapsed: run.value.elapsed_time,
eventType: 'complete',
key: 'complete',
status: 'completed',
summary: previewValue(run.value.outputs),
timestamp: run.value.completed_at,
title: '运行完成',
});
}
return items;
});
const agentSummaries = computed(() => { const agentSummaries = computed(() => {
const agents = new Map< const agents = new Map<
string, string,
@@ -380,6 +489,29 @@ function getLogStatusText(status?: string) {
return status ? map[status] || status : '-'; return status ? map[status] || status : '-';
} }
function getRunEventTypeLabel(type?: string) {
const map: Record<string, string> = {
complete: '完成',
error: '错误',
llm_chunk: '模型输出',
node_complete: '节点完成',
node_event: '节点事件',
node_start: '节点开始',
parallel_complete: '并行完成',
parallel_start: '并行开始',
start: '开始',
waiting_input: '等待输入',
};
return type ? map[type] || type : '-';
}
function normalizeRunEventStatus(eventType?: string, status?: string) {
if (eventType === 'waiting_input' || status === 'waiting') return 'waiting';
if (eventType === 'error' || status === 'failed') return 'failed';
if (eventType === 'start' || status === 'running') return 'running';
return 'completed';
}
function previewValue(value: any) { function previewValue(value: any) {
if (value === undefined || value === null || value === '') return '-'; if (value === undefined || value === null || value === '') return '-';
if (typeof value === 'string') return value; if (typeof value === 'string') return value;
@@ -483,6 +615,11 @@ function selectLog(nodeId?: string, index?: number) {
} }
} }
function selectRunEvent(item: { index?: number; nodeId?: string }) {
if (!item.nodeId) return;
selectLog(item.nodeId, item.index);
}
defineExpose({ open }); defineExpose({ open });
</script> </script>
@@ -566,6 +703,58 @@ defineExpose({ open });
</ZqDescItem> </ZqDescItem>
</ZqDesc> </ZqDesc>
<div class="run-event-panel border-border rounded-lg border p-3">
<div class="mb-3 flex flex-wrap items-center justify-between gap-2">
<div>
<div class="text-sm font-medium">运行事件</div>
<div class="text-muted-foreground mt-1 text-xs">
串联展示开始节点等待错误并行和子流程事件
</div>
</div>
<ElTag size="small">{{ runEventTimeline.length }}</ElTag>
</div>
<div class="run-event-list">
<button
v-for="item in runEventTimeline"
:key="item.key"
class="run-event-item"
:class="{
'is-active': item.nodeId && item.nodeId === activeNodeId,
'is-clickable': item.nodeId,
}"
type="button"
@click="selectRunEvent(item)"
>
<span
class="run-event-dot"
:class="`is-${getLogStatusType(item.status)}`"
></span>
<span class="min-w-0">
<span class="flex min-w-0 items-center gap-2">
<span class="truncate text-xs font-medium">
{{ item.title }}
</span>
<ElTag size="small" type="info">
{{ getRunEventTypeLabel(item.eventType) }}
</ElTag>
</span>
<span class="text-muted-foreground mt-1 block truncate text-xs">
{{ item.summary }}
</span>
</span>
<span class="text-muted-foreground shrink-0 text-right text-xs">
<span class="block">{{ item.channel || '-' }}</span>
<span class="block">
{{ formatTimelineTime(item.timestamp) }}
<template v-if="item.elapsed">
· {{ formatDuration(item.elapsed) }}
</template>
</span>
</span>
</button>
</div>
</div>
<div class="run-report-grid grid gap-4 lg:grid-cols-[minmax(0,1fr)_minmax(0,1fr)_360px]"> <div class="run-report-grid grid gap-4 lg:grid-cols-[minmax(0,1fr)_minmax(0,1fr)_360px]">
<div class="border-border bg-muted/30 rounded-lg border p-3"> <div class="border-border bg-muted/30 rounded-lg border p-3">
<div class="mb-2 flex items-center justify-between gap-2"> <div class="mb-2 flex items-center justify-between gap-2">
@@ -961,6 +1150,67 @@ defineExpose({ open });
padding: 6px 10px; padding: 6px 10px;
} }
.workflow-run-detail-dialog .run-event-list {
display: grid;
grid-auto-columns: minmax(240px, 320px);
grid-auto-flow: column;
gap: 8px;
overflow-x: auto;
padding-bottom: 2px;
}
.workflow-run-detail-dialog .run-event-item {
display: grid;
grid-template-columns: 8px minmax(0, 1fr) auto;
align-items: center;
gap: 8px;
width: 100%;
min-height: 58px;
border: 1px solid var(--el-border-color-lighter);
border-radius: 8px;
background: var(--el-fill-color-blank);
padding: 8px 10px;
text-align: left;
}
.workflow-run-detail-dialog .run-event-item.is-clickable {
cursor: pointer;
}
.workflow-run-detail-dialog .run-event-item.is-clickable:hover {
border-color: var(--el-color-primary-light-5);
background: var(--el-fill-color-light);
}
.workflow-run-detail-dialog .run-event-item.is-active {
border-color: var(--el-color-primary);
background: var(--el-color-primary-light-9);
box-shadow: 0 0 0 2px var(--el-color-primary-light-7);
}
.workflow-run-detail-dialog .run-event-dot {
width: 8px;
height: 8px;
border-radius: 999px;
background: var(--el-text-color-placeholder);
}
.workflow-run-detail-dialog .run-event-dot.is-success {
background: var(--el-color-success);
}
.workflow-run-detail-dialog .run-event-dot.is-danger {
background: var(--el-color-danger);
}
.workflow-run-detail-dialog .run-event-dot.is-warning {
background: var(--el-color-warning);
}
.workflow-run-detail-dialog .run-event-dot.is-primary {
background: var(--el-color-primary);
}
.workflow-run-detail-dialog .collaboration-timeline { .workflow-run-detail-dialog .collaboration-timeline {
display: flex; display: flex;
flex-direction: column; flex-direction: column;
@@ -1143,6 +1393,12 @@ defineExpose({ open });
grid-template-columns: 1fr; grid-template-columns: 1fr;
} }
.workflow-run-detail-dialog .run-event-list {
grid-auto-flow: row;
grid-template-columns: 1fr;
overflow-x: visible;
}
.workflow-run-detail-dialog .collaboration-flow { .workflow-run-detail-dialog .collaboration-flow {
flex-wrap: wrap; flex-wrap: wrap;
overflow-x: visible; overflow-x: visible;