feat: expose agent collaboration metadata

2026-06-14 15:35:24 +08:00
parent 3f75c966c2
commit 7b7c82309c
6 changed files with 546 additions and 107 deletions
@@ -50,16 +50,63 @@ def _node_label_from_map(node_map: Dict[str, Any], node_id: str) -> str:
    return data.get('label') or node.get('label') or node_id


+def _node_collaboration_metadata(node_map: Dict[str, Any], node_id: str, node_type: str) -> dict:
+    node = node_map.get(node_id) or {}
+    data = node.get('data') or {}
+    metadata: Dict[str, Any] = {}
+
+    agent_code = (data.get('agent_code') or '').strip()
+    if agent_code:
+        metadata['agent_code'] = agent_code
+    if data.get('agent_name'):
+        metadata['agent_name'] = data.get('agent_name')
+
+    model_id = data.get('model_id')
+    if model_id:
+        metadata['model_id'] = str(model_id)
+
+    if data.get('subflow_name'):
+        metadata['subflow_name'] = data.get('subflow_name')
+    if data.get('show_subflow_messages') is not None:
+        metadata['show_subflow_messages'] = data.get('show_subflow_messages')
+    if data.get('forward_interactive') is not None:
+        metadata['forward_interactive'] = data.get('forward_interactive')
+
+    branches = data.get('branches') or []
+    if node_type == 'parallel' and branches:
+        metadata['branches'] = [
+            {
+                'id': branch.get('id'),
+                'name': branch.get('name') or branch.get('id'),
+            }
+            for branch in branches
+            if branch.get('id')
+        ]
+
+    if node_type in ('template', 'llm') and agent_code:
+        metadata['collaboration_role'] = data.get('label') or node_id
+
+    if node_type == 'subflow':
+        metadata['collaboration_role'] = data.get('label') or node_id
+
+    return metadata
+
+
 def _make_execution_log_entry(
    node_map: Dict[str, Any],
    node_id: str,
    node_type: str,
    **extra: Any,
 ) -> dict:
+    metadata = dict(extra.pop('metadata', {}) or {})
+    collaboration = _node_collaboration_metadata(node_map, node_id, node_type)
+    if collaboration:
+        metadata['collaboration'] = collaboration
    return {
        'node_id': node_id,
        'node_type': node_type,
        'node_label': _node_label_from_map(node_map, node_id),
+        'metadata': metadata,
        **extra,
    }

@@ -68,6 +115,55 @@ def _node_result_metadata(result: NodeResult) -> dict:
    return dict(result.metadata or {})


+def _event_collaboration_metadata(
+    node_map: Dict[str, Any],
+    node_id: str,
+    node_type: str,
+    result: Optional[NodeResult] = None,
+) -> dict:
+    metadata = _node_collaboration_metadata(node_map, node_id, node_type)
+    if not result:
+        return metadata
+
+    result_metadata = result.metadata or {}
+    for key in (
+        'agent_code',
+        'agent_name',
+        'model',
+        'model_id',
+        'output_variable',
+        'subflow_name',
+    ):
+        value = result_metadata.get(key)
+        if value is not None and value != '':
+            metadata[key] = value
+    return metadata
+
+
+def _parallel_branch_labels(node_map: Dict[str, Any], parallel_node_id: str) -> Dict[str, str]:
+    node = node_map.get(parallel_node_id) or {}
+    branches = (node.get('data') or {}).get('branches') or []
+    labels: Dict[str, str] = {}
+    for branch in branches:
+        branch_id = branch.get('id')
+        if branch_id:
+            labels[branch_id] = branch.get('name') or branch_id
+    return labels
+
+
+def _merge_event_collaboration(
+    event: Dict[str, Any],
+    node_map: Dict[str, Any],
+    node_id: str,
+    node_type: str,
+    result: Optional[NodeResult] = None,
+) -> Dict[str, Any]:
+    collaboration = _event_collaboration_metadata(node_map, node_id, node_type, result)
+    if collaboration:
+        event['collaboration'] = collaboration
+    return event
+
+
 def _snapshot_node_inputs(context: NodeContext) -> dict:
    return {
        'variables': dict(context.variables),
@@ -691,6 +787,7 @@ class AIWorkflowService:
        all_logs = []
        total_tokens = 0
        total_steps = 0
+        branch_labels = _parallel_branch_labels(node_map, parallel_node_id)
        
        async def execute_branch(branch_id: str, branch_start_id: str, branch_context: NodeContext) -> Dict:
            """执行单个分支（异步）"""
@@ -723,17 +820,19 @@ class AIWorkflowService:
                result = await node_instance.execute_async(branch_context)
                elapsed = int((time.time() - start_time) * 1000)
                
-                branch_logs.append({
-                    'node_id': current_id,
-                    'node_type': node_type,
-                    'status': 'completed' if result.success else 'failed',
-                    'output': result.output,
-                    'error': result.error,
-                    'elapsed_time': elapsed,
-                    'tokens_used': result.tokens_used,
-                    'metadata': _node_result_metadata(result),
-                    'branch': branch_start_id,
-                })
+                branch_logs.append(_make_execution_log_entry(
+                    node_map,
+                    current_id,
+                    node_type,
+                    status='completed' if result.success else 'failed',
+                    output=result.output,
+                    error=result.error,
+                    elapsed_time=elapsed,
+                    tokens_used=result.tokens_used,
+                    metadata=_node_result_metadata(result),
+                    branch=branch_id,
+                    branch_label=branch_labels.get(branch_id, branch_id),
+                ))
                
                branch_tokens += result.tokens_used
                branch_steps += 1
@@ -881,6 +980,7 @@ class AIWorkflowService:
        all_logs = []
        total_tokens = 0
        total_steps = 0
+        branch_labels = _parallel_branch_labels(node_map, parallel_node_id)
        
        target_to_branch = {v: k for k, v in branch_mapping.items()}
        
@@ -921,45 +1021,49 @@ class AIWorkflowService:
                    continue
                
                # 发送节点开始事件
-                yield {
+                yield _merge_event_collaboration({
                    'type': 'node_start',
                    'node_id': current_id,
                    'node_type': node_type,
                    'node_label': node_label,
                    'branch_id': branch_id,
+                    'branch_label': branch_labels.get(branch_id, branch_id),
                    'timestamp': datetime.now().isoformat(),
                    'inputs': {
                        'variables': dict(branch_context.variables),
                        'previous_output': branch_context.previous_output,
                    },
-                }
+                }, node_map, current_id, node_type)
                
                # 执行节点（异步）
                start_time = time.time()
                result = await node_instance.execute_async(branch_context)
                elapsed = int((time.time() - start_time) * 1000)
                
-                branch_logs.append({
-                    'node_id': current_id,
-                    'node_type': node_type,
-                    'status': 'completed' if result.success else 'failed',
-                    'output': result.output,
-                    'error': result.error,
-                    'elapsed_time': elapsed,
-                    'tokens_used': result.tokens_used,
-                    'metadata': _node_result_metadata(result),
-                    'branch': branch_id,
-                })
+                branch_logs.append(_make_execution_log_entry(
+                    node_map,
+                    current_id,
+                    node_type,
+                    status='completed' if result.success else 'failed',
+                    output=result.output,
+                    error=result.error,
+                    elapsed_time=elapsed,
+                    tokens_used=result.tokens_used,
+                    metadata=_node_result_metadata(result),
+                    branch=branch_id,
+                    branch_label=branch_labels.get(branch_id, branch_id),
+                ))
                
                branch_tokens += result.tokens_used
                branch_steps += 1
                
                # 发送节点完成事件
-                yield _apply_node_result_warnings({
+                yield _merge_event_collaboration(_apply_node_result_warnings({
                    'type': 'node_complete',
                    'node_id': current_id,
                    'node_type': node_type,
                    'branch_id': branch_id,
+                    'branch_label': branch_labels.get(branch_id, branch_id),
                    'status': 'success' if result.success else 'failed',
                    'elapsed_time': elapsed,
                    'tokens_used': result.tokens_used,
@@ -968,7 +1072,7 @@ class AIWorkflowService:
                        'output': result.output,
                        'output_variables': result.metadata.get('frontend_output_variables', result.output_variables),
                    },
-                }, result)
+                }, result), node_map, current_id, node_type, result)
                
                if not result.success:
                    break
@@ -1205,7 +1309,7 @@ class AIWorkflowService:
                    continue
                
                # 发送节点开始事件
-                yield {
+                yield _merge_event_collaboration({
                    'type': 'node_start',
                    'node_id': current_id,
                    'node_type': node_type,
@@ -1219,7 +1323,7 @@ class AIWorkflowService:
                            '_loop_total': context.get_variable('_loop_total'),
                        }
                    },
-                }
+                }, node_map, current_id, node_type)
                
                # 执行节点（异步）
                start_time = time.time()
@@ -1228,23 +1332,24 @@ class AIWorkflowService:
                elapsed = int((time.time() - start_time) * 1000)
                logger.info(f'[Loop] Child node result: success={result.success}, output={result.output}, error={result.error}')
                
-                iteration_logs.append({
-                    'node_id': current_id,
-                    'node_type': node_type,
-                    'status': 'completed' if result.success else 'failed',
-                    'output': result.output,
-                    'error': result.error,
-                    'elapsed_time': elapsed,
-                    'tokens_used': result.tokens_used,
-                    'metadata': _node_result_metadata(result),
-                    'loop_iteration': iteration,
-                })
+                iteration_logs.append(_make_execution_log_entry(
+                    node_map,
+                    current_id,
+                    node_type,
+                    status='completed' if result.success else 'failed',
+                    output=result.output,
+                    error=result.error,
+                    elapsed_time=elapsed,
+                    tokens_used=result.tokens_used,
+                    metadata=_node_result_metadata(result),
+                    loop_iteration=iteration,
+                ))
                
                iteration_tokens += result.tokens_used
                iteration_steps += 1
                
                # 发送节点执行完成事件
-                yield _apply_node_result_warnings({
+                yield _merge_event_collaboration(_apply_node_result_warnings({
                    'type': 'node_complete',
                    'node_id': current_id,
                    'node_type': node_type,
@@ -1265,7 +1370,7 @@ class AIWorkflowService:
                            '_loop_total': context.get_variable('_loop_total'),
                        }
                    },
-                }, result)
+                }, result), node_map, current_id, node_type, result)
                
                # 发送节点产生的事件（如消息事件）
                if result.events:
@@ -1280,13 +1385,14 @@ class AIWorkflowService:
                # 检查是否需要等待用户输入（设计预览节点等）
                if result.waiting_for_input:
                    # 记录日志
-                    iteration_logs.append({
-                        'node_id': current_id,
-                        'node_type': node_type,
-                        'status': 'waiting',
-                        'elapsed_time': elapsed,
-                        'loop_iteration': iteration,
-                    })
+                    iteration_logs.append(_make_execution_log_entry(
+                        node_map,
+                        current_id,
+                        node_type,
+                        status='waiting',
+                        elapsed_time=elapsed,
+                        loop_iteration=iteration,
+                    ))
                    
                    # 更新运行记录为等待状态
                    run.status = 'waiting'
@@ -1304,13 +1410,13 @@ class AIWorkflowService:
                        await self._db.commit()
                    
                    # 发送等待事件
-                    yield {
+                    yield _merge_event_collaboration({
                        'type': 'waiting_input',
                        'node_id': current_id,
                        'node_type': node_type,
                        'config': result.waiting_config,
                        'loop_iteration': iteration,
-                    }
+                    }, node_map, current_id, node_type, result)
                    # 暂停循环执行，等待用户输入
                    return
                
@@ -1574,14 +1680,14 @@ class AIWorkflowService:
            }
            
            # 发送节点开始事件（包含时间戳和输入数据）
-            yield {
+            yield _merge_event_collaboration({
                'type': 'node_start',
                'node_id': current_node_id,
                'node_type': node_type,
                'node_label': node_label,
                'timestamp': datetime.now().isoformat(),
                'inputs': node_inputs,
-            }
+            }, node_map, current_node_id, node_type)
            
            # 结束节点
            if node_type == 'end':
@@ -1597,14 +1703,14 @@ class AIWorkflowService:
                        metadata=_node_result_metadata(result),
                        inputs=copy.deepcopy(node_inputs),
                    ))
-                    yield {
+                    yield _merge_event_collaboration({
                        'type': 'node_complete',
                        'node_id': current_node_id,
                        'node_type': node_type,
                        'status': 'success',
                        'elapsed_time': 0,
                        'tokens_used': 0,
-                    }
+                    }, node_map, current_node_id, node_type, result)
                    
                    # 保存结束节点输出和上下文变量到运行记录
                    run.execution_log = logs.copy()
@@ -1659,12 +1765,12 @@ class AIWorkflowService:
                        )
                        # 发送 LLM 流式内容事件
                        if chunk_event.content:
-                            yield {
+                            yield _merge_event_collaboration({
                                'type': 'llm_chunk',
                                'node_id': current_node_id,
                                'content': chunk_event.content,
                                'accumulated_content': chunk_event.accumulated_content,
-                            }
+                            }, node_map, current_node_id, node_type)
                except StopIteration as e:
                    # 生成器结束，获取返回值（NodeResult）
                    result = e.value
@@ -1682,18 +1788,22 @@ class AIWorkflowService:
                for event in result.events:
                    if event.get('type') == 'message':
                        # 消息事件直接发送为 answer 类型
-                        yield {
+                        answer_event = _merge_event_collaboration({
                            'type': 'answer',
                            'node_id': current_node_id,
                            'content': event.get('content', ''),
-                        }
+                        }, node_map, current_node_id, node_type, result)
+                        if event.get('from_subflow'):
+                            answer_event['from_subflow'] = True
+                            answer_event['subflow_name'] = event.get('subflow_name')
+                        yield answer_event
                    else:
                        # 其他事件保持 node_event 格式
-                        yield {
+                        yield _merge_event_collaboration({
                            'type': 'node_event',
                            'node_id': current_node_id,
                            'event': event,
-                        }
+                        }, node_map, current_node_id, node_type, result)
            
            # 检查是否需要等待用户输入（对话流节点）
            if result.waiting_for_input:
@@ -1723,12 +1833,12 @@ class AIWorkflowService:
                # 注意：save由调用方处理
                
                # 发送等待输入事件
-                yield {
+                yield _merge_event_collaboration({
                    'type': 'waiting_input',
                    'node_id': current_node_id,
                    'node_type': node_type,
                    'config': result.waiting_config,
-                }
+                }, node_map, current_node_id, node_type, result)
                
                # 暂停工作流执行，等待用户输入后续流
                return
@@ -1760,7 +1870,7 @@ class AIWorkflowService:
            # 注意：save由调用方处理
            
            # 发送节点完成事件（包含输出数据）
-            yield _apply_node_result_warnings({
+            yield _merge_event_collaboration(_apply_node_result_warnings({
                'type': 'node_complete',
                'node_id': current_node_id,
                'node_type': node_type,
@@ -1772,7 +1882,7 @@ class AIWorkflowService:
                    'output': result.output,
                    'output_variables': result.metadata.get('frontend_output_variables', result.output_variables),
                },
-            }, result)
+            }, result), node_map, current_node_id, node_type, result)
            
            if not result.success:
                raise ValueError(f'节点执行失败: {result.error}')
@@ -1791,11 +1901,15 @@ class AIWorkflowService:
            if node_type == 'parallel':
                # 并行节点：执行所有分支（流式）
                branch_ids = list(parallel_edge_map.get(current_node_id, {}).keys()) or edge_map.get(current_node_id, [])
-                yield {
+                branch_labels = _parallel_branch_labels(node_map, current_node_id)
+                yield _merge_event_collaboration({
                    'type': 'parallel_start',
                    'node_id': current_node_id,
                    'branches': branch_ids,
-                }
+                    'branch_count': len(branch_ids),
+                    'branch_labels': branch_labels,
+                    'collaboration_mode': 'parallel',
+                }, node_map, current_node_id, node_type)
                
                # 使用流式版本的并行分支执行（异步）
                async for event in self._execute_parallel_branches_stream(
@@ -1809,12 +1923,15 @@ class AIWorkflowService:
                        total_steps += parallel_results['total_steps']
                        logs.extend(parallel_results['logs'])
                        
-                        yield {
+                        yield _merge_event_collaboration({
                            'type': 'parallel_complete',
                            'node_id': current_node_id,
                            'branch_results': parallel_results['results'],
                            'total_tokens': parallel_results['total_tokens'],
-                        }
+                            'branch_count': len(parallel_results['results']),
+                            'branch_labels': branch_labels,
+                            'collaboration_mode': 'parallel',
+                        }, node_map, current_node_id, node_type)
                        
                        # 找到合并节点继续执行
                        current_node_id = parallel_results.get('merge_node_id')
@@ -2126,14 +2243,14 @@ class AIWorkflowService:
            }
            
            # 发送节点开始事件
-            node_start_event = {
+            node_start_event = _merge_event_collaboration({
                'type': 'node_start',
                'node_id': current_node_id,
                'node_type': node_type,
                'node_label': node_label,
                'timestamp': datetime.now().isoformat(),
                'inputs': node_inputs,
-            }
+            }, node_map, current_node_id, node_type)
            # 如果在循环中，添加迭代信息
            if loop_state:
                node_start_event['loop_iteration'] = loop_state.get('iteration', 0)
@@ -2152,14 +2269,14 @@ class AIWorkflowService:
                        output=result.output,
                        inputs=copy.deepcopy(node_inputs),
                    ))
-                    yield {
+                    yield _merge_event_collaboration({
                        'type': 'node_complete',
                        'node_id': current_node_id,
                        'node_type': node_type,
                        'status': 'success',
                        'elapsed_time': 0,
                        'outputs': {'output': result.output},
-                    }
+                    }, node_map, current_node_id, node_type, result)
                    
                    # 更新运行记录，保存结束节点的输出
                    run.status = 'completed'
@@ -2191,29 +2308,34 @@ class AIWorkflowService:
                for event in result.events:
                    if event.get('type') == 'message':
                        # 消息事件直接发送为 answer 类型
-                        yield {
+                        answer_event = _merge_event_collaboration({
                            'type': 'answer',
                            'node_id': current_node_id,
                            'content': event.get('content', ''),
-                        }
+                        }, node_map, current_node_id, node_type, result)
+                        if event.get('from_subflow'):
+                            answer_event['from_subflow'] = True
+                            answer_event['subflow_name'] = event.get('subflow_name')
+                        yield answer_event
                    else:
                        # 其他事件保持 node_event 格式
-                        yield {
+                        yield _merge_event_collaboration({
                            'type': 'node_event',
                            'node_id': current_node_id,
                            'event': event,
-                        }
+                        }, node_map, current_node_id, node_type, result)
            
            # 检查是否需要等待用户输入
            if result.waiting_for_input:
-                log_entry = {
-                    'node_id': current_node_id,
-                    'node_type': node_type,
-                    'status': 'waiting',
-                    'output': result.output,
-                    'elapsed_time': elapsed,
-                    'metadata': _node_result_metadata(result),
-                }
+                log_entry = _make_execution_log_entry(
+                    node_map,
+                    current_node_id,
+                    node_type,
+                    status='waiting',
+                    output=result.output,
+                    elapsed_time=elapsed,
+                    metadata=_node_result_metadata(result),
+                )
                logs.append(log_entry)
                
                run.status = 'waiting'
@@ -2226,12 +2348,12 @@ class AIWorkflowService:
                    run.waiting_config['_loop_state'] = loop_state
                # 注意：save由调用方处理
                
-                waiting_event = {
+                waiting_event = _merge_event_collaboration({
                    'type': 'waiting_input',
                    'node_id': current_node_id,
                    'node_type': node_type,
                    'config': result.waiting_config,
-                }
+                }, node_map, current_node_id, node_type, result)
                # 如果在循环中，添加迭代信息
                if loop_state:
                    waiting_event['loop_iteration'] = loop_state.get('iteration', 0)
@@ -2239,16 +2361,17 @@ class AIWorkflowService:
                return
            
            # 记录日志
-            log_entry = {
-                'node_id': current_node_id,
-                'node_type': node_type,
-                'status': 'completed' if result.success else 'failed',
-                'output': result.output,
-                'error': result.error,
-                'elapsed_time': elapsed,
-                'tokens_used': result.tokens_used,
-                'metadata': _node_result_metadata(result),
-            }
+            log_entry = _make_execution_log_entry(
+                node_map,
+                current_node_id,
+                node_type,
+                status='completed' if result.success else 'failed',
+                output=result.output,
+                error=result.error,
+                elapsed_time=elapsed,
+                tokens_used=result.tokens_used,
+                metadata=_node_result_metadata(result),
+            )
            logs.append(log_entry)
            
            total_tokens += result.tokens_used
@@ -2266,7 +2389,7 @@ class AIWorkflowService:
            # 注意：save由调用方处理
            
            # 发送节点完成事件
-            node_complete_event = _apply_node_result_warnings({
+            node_complete_event = _merge_event_collaboration(_apply_node_result_warnings({
                'type': 'node_complete',
                'node_id': current_node_id,
                'node_type': node_type,
@@ -2278,7 +2401,7 @@ class AIWorkflowService:
                    'output': result.output,
                    'output_variables': result.metadata.get('frontend_output_variables', result.output_variables),
                },
-            }, result)
+            }, result), node_map, current_node_id, node_type, result)
            # 如果在循环中，添加迭代信息
            if loop_state:
                node_complete_event['loop_iteration'] = loop_state.get('iteration', 0)