diff --git a/backend-fastapi/ai_platform/services/agent_service.py b/backend-fastapi/ai_platform/services/agent_service.py index 46e1562..47249b4 100644 --- a/backend-fastapi/ai_platform/services/agent_service.py +++ b/backend-fastapi/ai_platform/services/agent_service.py @@ -26,6 +26,26 @@ class AgentService: def __init__(self, db: Optional[AsyncSession] = None): self._db = db self.llm_service = LLMService(db) + + def _agent_collaboration_metadata(self, agent: Agent) -> Dict[str, Any]: + metadata = { + "agent_id": str(agent.id), + "agent_code": agent.code, + "agent_name": agent.name, + "collaboration_role": agent.name or agent.code, + "collaboration_mode": agent.mode, + } + if agent.model_id: + metadata["model_id"] = str(agent.model_id) + return {key: value for key, value in metadata.items() if value} + + def _attach_agent_collaboration(self, event: Dict[str, Any], agent: Agent) -> Dict[str, Any]: + enriched = dict(event) + collaboration = dict(enriched.get("collaboration") or {}) + for key, value in self._agent_collaboration_metadata(agent).items(): + collaboration.setdefault(key, value) + enriched["collaboration"] = collaboration + return enriched async def chat( self, @@ -52,12 +72,15 @@ class AgentService: """ if agent.mode == 'autonomous': async for event in self._chat_autonomous(agent, conversation, user_message, attachments): - yield event + yield self._attach_agent_collaboration(event, agent) elif agent.mode == 'dialog_flow': async for event in self._chat_dialog_flow(agent, conversation, user_message, application_id, form_code, attachments): - yield event + yield self._attach_agent_collaboration(event, agent) else: - yield {'type': 'error', 'content': f'不支持的模式: {agent.mode}'} + yield self._attach_agent_collaboration( + {'type': 'error', 'content': f'不支持的模式: {agent.mode}'}, + agent, + ) async def _chat_autonomous( self, diff --git a/backend-fastapi/ai_platform/services/workflow_service.py b/backend-fastapi/ai_platform/services/workflow_service.py index f37694e..c6ee7f3 100644 --- a/backend-fastapi/ai_platform/services/workflow_service.py +++ b/backend-fastapi/ai_platform/services/workflow_service.py @@ -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) diff --git a/web/apps/web-ele/src/components/ChatBox/MessageBubble.vue b/web/apps/web-ele/src/components/ChatBox/MessageBubble.vue index 947dd66..3945a7b 100644 --- a/web/apps/web-ele/src/components/ChatBox/MessageBubble.vue +++ b/web/apps/web-ele/src/components/ChatBox/MessageBubble.vue @@ -192,6 +192,39 @@ const getStepColor = (step: ReasoningStep) => { return colors[step.type] || 'text-gray-500'; }; +const getStepMetaItems = (step: ReasoningStep) => { + const items: string[] = []; + if (step.agent_name || step.agent_code) { + items.push(`智能体: ${step.agent_name || step.agent_code}`); + } + if (step.branch_label || step.branch_id) { + items.push(`分支: ${step.branch_label || step.branch_id}`); + } + if (step.subflow_name) { + items.push(`子流程: ${step.subflow_name}`); + } + if (step.model || step.model_id) { + items.push(`模型: ${step.model || step.model_id}`); + } + return items; +}; + +const messageMetaItems = computed(() => { + if (isUser.value) return []; + + const items: string[] = []; + if (props.message.agent_name || props.message.agent_code) { + items.push(`智能体: ${props.message.agent_name || props.message.agent_code}`); + } + if (props.message.model_name || props.message.model_id) { + items.push(`模型: ${props.message.model_name || props.message.model_id}`); + } + if (props.message.collaboration_mode) { + items.push(`模式: ${props.message.collaboration_mode}`); + } + return items; +}); + // 格式化文件大小 const formatFileSize = (bytes?: number) => { if (!bytes) return ''; @@ -293,6 +326,16 @@ const formatVoiceDuration = (seconds: number) => { +
+