diff --git a/backend/AgentCoord/Export/__init__.py b/backend/AgentCoord/Export/__init__.py index 4e87363..ec76cf9 100644 --- a/backend/AgentCoord/Export/__init__.py +++ b/backend/AgentCoord/Export/__init__.py @@ -4,9 +4,16 @@ """ import json +import os from typing import Dict, Any, Optional from datetime import datetime +# 导入 LLM 导出器 +from .docx_llm import DocxLLMExporter +from .xlsx_llm import XlsxLLMExporter +from .infographic_llm import InfographicLLMExporter +from .markdown_llm import MarkdownLLMExporter + class BaseExporter: """导出器基类""" @@ -22,6 +29,11 @@ class BaseExporter: class MarkdownExporter(BaseExporter): """Markdown 导出器""" + # 需要过滤掉的任务大纲字段 + OUTLINE_FIELDS_TO_SKIP = { + 'Collaboration_Brief_FrontEnd', 'data', 'color', 'template', 'Collaboration_Brief' + } + def generate(self, file_path: str) -> bool: """生成 Markdown 文件""" try: @@ -43,23 +55,11 @@ class MarkdownExporter(BaseExporter): content_lines.append("## 任务大纲\n") content_lines.append(self._format_outline(task_outline)) - # 执行结果 - result = self.task_data.get('result') - if result: - content_lines.append("## 执行结果\n") - if isinstance(result, list): - for idx, item in enumerate(result, 1): - content_lines.append(f"### 步骤 {idx}\n") - content_lines.append(f"{json.dumps(item, ensure_ascii=False, indent=2)}\n") - else: - content_lines.append(f"{json.dumps(result, ensure_ascii=False, indent=2)}\n") - - # 参与智能体 - 从 task_outline 的 Collaboration Process 中提取 + # 参与智能体 task_outline = self.task_data.get('task_outline') if task_outline and isinstance(task_outline, dict): collaboration_process = task_outline.get('Collaboration Process', []) if collaboration_process and isinstance(collaboration_process, list): - # 收集所有参与步骤的智能体 all_agents = set() for step in collaboration_process: if isinstance(step, dict): @@ -69,10 +69,33 @@ class MarkdownExporter(BaseExporter): if agent: all_agents.add(agent) if all_agents: - content_lines.append("## 参与智能体\n") + content_lines.append("\n## 参与智能体\n") for agent_name in sorted(all_agents): content_lines.append(f"- {agent_name}") + # 智能体评分信息 + agent_scores = self.task_data.get('agent_scores') + if agent_scores: + content_lines.append("\n## 智能体评分\n") + content_lines.append(self._format_agent_scores(agent_scores)) + + # 执行结果 - 合并 rehearsal_log 和 result,取最完整的数据 + rehearsal_log = self.task_data.get('rehearsal_log') + result = self.task_data.get('result') + + # 优先使用 rehearsal_log,如果某个步骤数据不完整,用 result 补充 + if rehearsal_log: + content_lines.append("\n## 执行结果\n") + content_lines.append(self._format_with_fallback(rehearsal_log, result)) + elif result: + content_lines.append("\n## 执行结果\n") + if isinstance(result, list): + for idx, item in enumerate(result, 1): + content_lines.append(f"### 步骤 {idx}\n") + content_lines.append(self._format_result_item(item)) + else: + content_lines.append(self._format_result_item(result)) + # 写入文件 with open(file_path, 'w', encoding='utf-8') as f: f.write('\n'.join(content_lines)) @@ -80,8 +103,242 @@ class MarkdownExporter(BaseExporter): return True except Exception as e: print(f"Markdown 导出失败: {e}") + import traceback + traceback.print_exc() return False + def _format_result_item(self, item: Any, fallback: dict = None) -> str: + """格式化执行结果项 + Args: + item: 主数据字典 + fallback: 可选的备用数据字典,用于补充 item 中缺失的字段 + """ + lines = [] + + if not isinstance(item, dict): + lines.append(str(item)) + return '\n'.join(lines) + + # 如果有 fallback,用 fallback 补充缺失的字段 + if fallback: + data = { + 'OutputName': item.get('OutputName', fallback.get('OutputName', '')), + 'NodeId': item.get('NodeId', fallback.get('NodeId', '')), + 'TaskContent': item.get('TaskContent', fallback.get('TaskContent', '')), + 'InputName_List': item.get('InputName_List', fallback.get('InputName_List', [])), + 'inputObject_Record': item.get('inputObject_Record', fallback.get('inputObject_Record', [])), + 'ActionHistory': item.get('ActionHistory', fallback.get('ActionHistory', [])), + 'content': item.get('content', fallback.get('content', '')), + } + else: + data = item + + # 提取关键字段 + output_name = data.get('OutputName', '') + node_id = data.get('NodeId', '') + task_content = data.get('TaskContent', '') + action_history = data.get('ActionHistory', []) + content = data.get('content', '') + + # 输出产物 + if output_name: + lines.append(f"**输出产物**: {output_name}") + if node_id and node_id != output_name: + lines.append(f"**步骤名称**: {node_id}") + + # 产物级:产物执行结果 - content + # content 是产物级别的,独立显示 + if content: + if isinstance(content, str): + decoded_content = content.replace('\\n', '\n').replace('\\t', '\t').replace('\\r', '\r') + lines.append(f"\n**产物执行结果**:") + for line in decoded_content.split('\n'): + lines.append(line) + else: + lines.append(f"\n**产物执行结果**: {content}") + # 步骤级:动作执行过程 - ActionHistory + # 只有当没有 content 时才显示执行过程 + elif action_history and isinstance(action_history, list) and len(action_history) > 0: + lines.append("\n### 执行过程\n") + for action_idx, action in enumerate(action_history, 1): + if not isinstance(action, dict): + continue + + agent_name = action.get('AgentName', '未知智能体') + action_type = action.get('ActionType', '') + description = action.get('Description', '') + action_result = action.get('Action_Result', '') + + lines.append(f"#### 动作 {action_idx}: {agent_name} ({action_type})") + if description: + lines.append(f"**任务描述**: {description}") + + if action_result: + if isinstance(action_result, str): + decoded_result = action_result.replace('\\n', '\n').replace('\\t', '\t').replace('\\r', '\r') + lines.append(f"\n**执行结果**:\n{decoded_result}") + else: + lines.append(f"\n**执行结果**: {action_result}") + lines.append("") + + # 任务内容 + if task_content: + lines.append(f"\n**任务内容**: {task_content}") + + # 输入产物 - 包含详细内容 + input_name_list = data.get('InputName_List', []) + input_object_record = data.get('inputObject_Record', []) + if input_name_list: + lines.append(f"\n**输入产物**:") + for inp_name in input_name_list: + lines.append(f" - {inp_name}") + # 查找对应的详细内容 + if isinstance(input_object_record, list): + for record in input_object_record: + if isinstance(record, dict) and inp_name in record: + inp_content = record[inp_name] + if isinstance(inp_content, str): + decoded_inp = inp_content.replace('\\n', '\n').replace('\\t', '\t').replace('\\r', '\r') + lines.append(f" 内容:") + for line in decoded_inp.split('\n'): + lines.append(f" {line}") + break + + lines.append("") + return '\n'.join(lines) + + def _should_skip_field(self, key: str) -> bool: + """判断是否应该跳过某个字段""" + key_lower = key.lower() + skip_patterns = ['brief', 'data', 'color', 'template', '_front'] + return any(pattern in key_lower for pattern in skip_patterns) + + def _format_agent_scores(self, agent_scores: Any) -> str: + """格式化智能体评分信息""" + lines = [] + + if not agent_scores: + return "无评分信息" + + if isinstance(agent_scores, list): + for item in agent_scores: + if isinstance(item, dict): + aspect_list = item.get('aspectList', []) + agent_scores_dict = item.get('agentScores', {}) + + if aspect_list: + lines.append(f"\n**评估维度**: {', '.join(aspect_list)}\n") + + if isinstance(agent_scores_dict, dict): + for agent_name, scores in agent_scores_dict.items(): + lines.append(f"\n### {agent_name}") + if isinstance(scores, dict): + for aspect, score_info in scores.items(): + if isinstance(score_info, dict): + score = score_info.get('score', '') + reason = score_info.get('reason', '') + lines.append(f"- **{aspect}**:") + if score: + lines.append(f" - 评分: {score}") + if reason: + lines.append(f" - 理由: {reason}") + else: + lines.append(f"- **{aspect}**: {score_info}") + else: + lines.append(f"- {scores}") + elif isinstance(agent_scores, dict): + for agent_name, score in agent_scores.items(): + if isinstance(score, dict): + lines.append(f"\n### {agent_name}") + for k, v in score.items(): + lines.append(f"- **{k}**: {v}") + else: + lines.append(f"\n### {agent_name}: {score}") + else: + lines.append(str(agent_scores)) + + return '\n'.join(lines) if lines else "无评分信息" + + def _format_with_fallback(self, rehearsal_log: Any, result: Any) -> str: + """合并 rehearsal_log 和 result 的数据,取最完整的""" + lines = [] + + if not rehearsal_log and not result: + return "无执行日志" + + # 将 result 转换为字典方便查找 + # 支持多种 key: NodeId, OutputName, content + result_dict = {} + if result and isinstance(result, list): + for item in result: + if isinstance(item, dict): + # 用多个字段作为 key + keys = [] + if item.get('NodeId'): + keys.append(item['NodeId']) + if item.get('OutputName'): + keys.append(item['OutputName']) + for key in keys: + result_dict[key] = item + + if rehearsal_log and isinstance(rehearsal_log, list): + for idx, entry in enumerate(rehearsal_log, 1): + if isinstance(entry, dict): + # 用 NodeId 或 OutputName 作为 key + step_key = entry.get('NodeId', entry.get('OutputName', '')) + + lines.append(f"### 步骤 {idx}\n") + + # 如果 rehearsal_log 数据不完整,尝试从 result 补充 + found_in_result = False + if step_key and step_key in result_dict: + result_item = result_dict[step_key] + lines.append(self._format_result_item(entry, result_item)) + found_in_result = True + elif step_key: + # 尝试模糊匹配 - result 中的 NodeId 可能包含在 key 中 + for k, v in result_dict.items(): + if k in step_key or step_key in k: + lines.append(self._format_result_item(entry, v)) + found_in_result = True + break + + if not found_in_result: + lines.append(self._format_result_item(entry)) + lines.append("") + elif rehearsal_log and isinstance(rehearsal_log, dict): + lines.append(self._format_result_item(rehearsal_log)) + else: + if result and isinstance(result, list): + for idx, item in enumerate(result, 1): + lines.append(f"### 步骤 {idx}\n") + lines.append(self._format_result_item(item)) + lines.append("") + elif result: + lines.append(self._format_result_item(result)) + + return '\n'.join(lines) + + def _format_rehearsal_log(self, rehearsal_log: Any) -> str: + """格式化详细执行日志 - 使用和 result 相同的格式""" + lines = [] + + if not rehearsal_log: + return "无执行日志" + + if isinstance(rehearsal_log, list): + for idx, entry in enumerate(rehearsal_log, 1): + lines.append(f"### 步骤 {idx}\n") + lines.append(self._format_result_item(entry)) + lines.append("") + elif isinstance(rehearsal_log, dict): + # 如果是单个步骤 + lines.append(self._format_result_item(rehearsal_log)) + else: + lines.append(str(rehearsal_log)) + + return '\n'.join(lines) + def _format_outline(self, outline: Any, level: int = 2) -> str: """格式化大纲内容""" lines = [] @@ -89,14 +346,24 @@ class MarkdownExporter(BaseExporter): if isinstance(outline, dict): for key, value in outline.items(): + # 跳过不需要的字段 + if self._should_skip_field(key): + continue + # 如果值是简单类型,直接显示 if isinstance(value, (str, int, float, bool)) or value is None: lines.append(f"**{key}**: {value}") # 如果值是列表或字典,递归处理 elif isinstance(value, list): - lines.append(f"**{key}**:") - lines.append(self._format_outline(value, level + 1)) + if len(value) == 0: + lines.append(f"**{key}**: (无)") + else: + lines.append(f"**{key}**:") + lines.append(self._format_outline(value, level + 1)) elif isinstance(value, dict): + # 检查是否整个字典都是要跳过的字段 + if all(self._should_skip_field(k) for k in value.keys()): + continue lines.append(f"**{key}**:") lines.append(self._format_outline(value, level + 1)) else: @@ -104,20 +371,43 @@ class MarkdownExporter(BaseExporter): elif isinstance(outline, list): for idx, item in enumerate(outline, 1): if isinstance(item, dict): - # 列表中的每个字典作为一个整体项 - lines.append(f"{prefix} 步骤 {idx}") - for key, value in item.items(): - if isinstance(value, (str, int, float, bool)) or value is None: - lines.append(f" - **{key}**: {value}") - elif isinstance(value, list): - lines.append(f" - **{key}**:") - for v in value: - lines.append(f" - {v}") - elif isinstance(value, dict): - lines.append(f" - **{key}**:") - lines.append(self._format_outline(value, level + 2)) - else: - lines.append(f" - **{key}**: {value}") + # 检查是否是 Collaboration Process 中的步骤 + step_name = item.get('StepName', f'步骤 {idx}') + task_content = item.get('TaskContent', '') + agent_selection = item.get('AgentSelection', []) + output_object = item.get('OutputObject', '') + input_object_list = item.get('InputObject_List', []) + + # 步骤标题 + lines.append(f"{prefix} {step_name}") + + # 任务内容 + if task_content: + lines.append(f" - **任务内容**: {task_content}") + + # 输入产物 + if input_object_list: + lines.append(f" - **输入产物**: {', '.join(input_object_list)}") + + # 输出产物 + if output_object: + lines.append(f" - **输出产物**: {output_object}") + + # 参与智能体 + if agent_selection: + lines.append(f" - **参与智能体**: {', '.join(agent_selection)}") + + # 任务流程 - 显示所有动作 + task_process = item.get('TaskProcess', []) + if task_process and isinstance(task_process, list): + lines.append(f" - **执行流程**: 共 {len(task_process)} 个动作") + for proc_idx, proc in enumerate(task_process, 1): + if isinstance(proc, dict): + proc_agent = proc.get('AgentName', '') + proc_type = proc.get('ActionType', '') + proc_desc = proc.get('Description', '') + lines.append(f" {proc_idx}. {proc_agent} ({proc_type}): {proc_desc}") + lines.append("") else: lines.append(f"- {item}") @@ -163,16 +453,15 @@ class DocxExporter(BaseExporter): if isinstance(result, list): for idx, item in enumerate(result, 1): doc.add_heading(f'步骤 {idx}', level=2) - doc.add_paragraph(json.dumps(item, ensure_ascii=False, indent=2)) + self._add_result_item_to_doc(doc, item) else: - doc.add_paragraph(json.dumps(result, ensure_ascii=False, indent=2)) + self._add_result_item_to_doc(doc, result) # 参与智能体 - 从 task_outline 的 Collaboration Process 中提取 task_outline = self.task_data.get('task_outline') if task_outline and isinstance(task_outline, dict): collaboration_process = task_outline.get('Collaboration Process', []) if collaboration_process and isinstance(collaboration_process, list): - # 收集所有参与步骤的智能体 all_agents = set() for step in collaboration_process: if isinstance(step, dict): @@ -196,20 +485,82 @@ class DocxExporter(BaseExporter): return False except Exception as e: print(f"Word 导出失败: {e}") + import traceback + traceback.print_exc() return False + def _add_result_item_to_doc(self, doc, item: Any): + """格式化执行结果项并添加到 Word 文档""" + if not isinstance(item, dict): + doc.add_paragraph(str(item)) + return + + output_name = item.get('OutputName', '') + node_id = item.get('NodeId', '') + task_content = item.get('TaskContent', '') + + if output_name: + doc.add_paragraph(f"输出产物: {output_name}") + if node_id and node_id != output_name: + doc.add_paragraph(f"步骤名称: {node_id}") + + if task_content: + doc.add_paragraph(f"\n任务内容: {task_content}") + + input_name_list = item.get('InputName_List', []) + if input_name_list: + doc.add_paragraph(f"\n输入产物:") + for inp in input_name_list: + doc.add_paragraph(f" - {inp}") + + action_history = item.get('ActionHistory', []) + if action_history and isinstance(action_history, list): + doc.add_heading('执行过程', level=3) + for action_idx, action in enumerate(action_history, 1): + if not isinstance(action, dict): + continue + + agent_name = action.get('AgentName', '未知智能体') + action_type = action.get('ActionType', '') + description = action.get('Description', '') + action_result = action.get('Action_Result', '') + + doc.add_paragraph(f"动作 {action_idx}: {agent_name} ({action_type})") + if description: + doc.add_paragraph(f" 任务描述: {description}") + + if action_result: + if isinstance(action_result, str): + decoded_result = action_result.replace('\\n', '\n').replace('\\t', '\t').replace('\\r', '\r') + doc.add_paragraph(f"\n执行结果:\n{decoded_result}") + else: + doc.add_paragraph(f"\n执行结果: {action_result}") + doc.add_paragraph("") + + def _should_skip_field(self, key: str) -> bool: + """判断是否应该跳过某个字段""" + key_lower = key.lower() + skip_patterns = ['brief', 'data', 'color', 'template', '_front'] + return any(pattern in key_lower for pattern in skip_patterns) + def _add_outline_to_doc(self, doc, outline: Any, level: int = 2): """递归添加大纲到文档""" if isinstance(outline, dict): for key, value in outline.items(): - # 如果值是简单类型,直接显示为段落 + if self._should_skip_field(key): + continue + if isinstance(value, (str, int, float, bool)) or value is None: doc.add_paragraph(f"**{key}**: {value}") - # 如果值是列表或字典,递归处理 elif isinstance(value, list): - doc.add_paragraph(f"**{key}**:") - self._add_outline_to_doc(doc, value, level + 1) + if len(value) == 0: + doc.add_paragraph(f"**{key}**: (无)") + else: + doc.add_paragraph(f"**{key}**:") + self._add_outline_to_doc(doc, value, level + 1) elif isinstance(value, dict): + if all(self._should_skip_field(k) for k in value.keys()): + continue doc.add_paragraph(f"**{key}**:") self._add_outline_to_doc(doc, value, level + 1) else: @@ -217,20 +568,37 @@ class DocxExporter(BaseExporter): elif isinstance(outline, list): for idx, item in enumerate(outline, 1): if isinstance(item, dict): - # 列表中的每个字典作为一个整体项 - doc.add_heading(f"步骤 {idx}", level=min(level, 3)) - for key, value in item.items(): - if isinstance(value, (str, int, float, bool)) or value is None: - doc.add_paragraph(f" - **{key}**: {value}") - elif isinstance(value, list): - doc.add_paragraph(f" - **{key}**:") - for v in value: - doc.add_paragraph(f" - {v}") - elif isinstance(value, dict): - doc.add_paragraph(f" - **{key}**:") - self._add_outline_to_doc(doc, value, level + 2) - else: - doc.add_paragraph(f" - **{key}**: {value}") + step_name = item.get('StepName', f'步骤 {idx}') + task_content = item.get('TaskContent', '') + agent_selection = item.get('AgentSelection', []) + output_object = item.get('OutputObject', '') + input_object_list = item.get('InputObject_List', []) + + doc.add_heading(step_name, level=min(level, 3)) + + if task_content: + doc.add_paragraph(f" 任务内容: {task_content}") + + if input_object_list: + doc.add_paragraph(f" 输入产物: {', '.join(input_object_list)}") + + if output_object: + doc.add_paragraph(f" 输出产物: {output_object}") + + if agent_selection: + doc.add_paragraph(f" 参与智能体: {', '.join(agent_selection)}") + + task_process = item.get('TaskProcess', []) + if task_process and isinstance(task_process, list): + doc.add_paragraph(f" 执行流程: 共 {len(task_process)} 个动作") + for proc_idx, proc in enumerate(task_process, 1): + if isinstance(proc, dict): + proc_agent = proc.get('AgentName', '') + proc_type = proc.get('ActionType', '') + proc_desc = proc.get('Description', '') + doc.add_paragraph(f" {proc_idx}. {proc_agent} ({proc_type}): {proc_desc}") + + doc.add_paragraph("") else: doc.add_paragraph(str(item)) else: @@ -737,8 +1105,9 @@ class InfographicExporter(BaseExporter): task_content = self.task_data.get('task_content', '') task_outline = self.task_data.get('task_outline') result = self.task_data.get('result') + rehearsal_log = self.task_data.get('rehearsal_log') + agent_scores = self.task_data.get('agent_scores') - # 从 task_outline 中提取参与智能体 all_agents = [] if task_outline and isinstance(task_outline, dict): collaboration_process = task_outline.get('Collaboration Process', []) @@ -753,13 +1122,104 @@ class InfographicExporter(BaseExporter): agents_set.add(agent) all_agents = sorted(agents_set) - # 统计执行步骤数 step_count = 0 if task_outline and isinstance(task_outline, dict): collaboration_process = task_outline.get('Collaboration Process', []) if isinstance(collaboration_process, list): step_count = len(collaboration_process) + infographic_data = None + try: + llm_exporter = InfographicLLMExporter() + infographic_data = llm_exporter.generate(self.task_data) + except Exception as e: + print(f"调用 LLM 生成信息图内容失败: {e}") + + summary = infographic_data.get('summary', '') if infographic_data else '' + highlights = infographic_data.get('highlights', []) if infographic_data else [] + statistics = infographic_data.get('statistics', {}) if infographic_data else {} + key_insights = infographic_data.get('key_insights', []) if infographic_data else [] + timeline = infographic_data.get('timeline', []) if infographic_data else [] + agent_performance = infographic_data.get('agent_performance', []) if infographic_data else [] + + llm_step_count = statistics.get('total_steps', step_count) + llm_agent_count = statistics.get('agent_count', len(all_agents)) + completion_rate = statistics.get('completion_rate', 0) + quality_score = statistics.get('quality_score', 0) + + summary_html = f''' +
{summary or '无'}
+{h}
无
'} +{insight}
无
'} +无
' + + agent_performance_html = f''' +无
' + + timeline_html = f''' +