95_跨任务提示：一个提示=无限可能！告别繁琐的单任务处理

跨任务提示优势分析
├── 经济价值
│   ├── token消耗减少30-60%
│   ├── API调用次数减少40-70%
│   └── 长期运营成本降低50%以上
├── 性能提升
│   ├── 端到端延迟减少40-60%
│   ├── 系统吞吐量提升80-120%
│   └── 资源利用率提高35-50%
├── 技术优势
│   ├── 上下文信息充分利用
│   ├── 任务间一致性保证
│   └── 应用架构简化
└── 用户体验
    ├── 响应速度更快
    ├── 交互流程更流畅
    └── 系统集成度更高

# 跨任务提示模式类型与特征
CROSS_TASK_PATTERNS = {
    'sequential': {
        '特征': '任务间存在明确依赖关系，后续任务需要前序任务的结果',
        '适用场景': '数据处理流水线、多步骤分析任务',
        '复杂度': '中等',
        '优化重点': '任务间信息传递的清晰度'
    },
    'parallel': {
        '特征': '任务间相互独立，可以同时处理',
        '适用场景': '多角度分析、多维度评价',
        '复杂度': '低',
        '优化重点': '任务边界的明确性'
    },
    'hierarchical': {
        '特征': '任务形成层次结构，高层任务依赖低层任务',
        '适用场景': '复杂问题分解、多粒度分析',
        '复杂度': '高',
        '优化重点': '层次关系的明确表示'
    },
    'conditional': {
        '特征': '任务执行依赖于条件判断，具有动态路由特性',
        '适用场景': '决策支持系统、智能客服',
        '复杂度': '很高',
        '优化重点': '条件判断的准确性与任务选择的合理性'
    }
}

{
  "task_results": [
    {
      "task_id": "task_1",
      "status": "completed",
      "result": "任务1的具体结果..."
    },
    {
      "task_id": "task_2",
      "status": "completed",
      "result": "任务2的具体结果..."
    }
  ],
  "summary": "总体执行情况总结..."
}

def create_sequential_analysis_prompt(text, tasks):
    """
    创建文本分析的串行式跨任务提示
    
    参数：
    - text: 待分析的文本内容
    - tasks: 任务列表，每个任务包含id、名称、描述和依赖关系
    
    返回：
    - 结构化的跨任务提示字符串
    """
    # 任务概览部分
    overview = f"""请按照指定顺序完成以下文本分析任务：
文本内容：{text}

任务列表：
"""
    
    # 任务定义部分
    task_definitions = []
    for i, task in enumerate(tasks, 1):
        dependencies = ", ".join(task['depends_on']) if task.get('depends_on') else "无"
        task_definitions.append(f"任务{i} ({task['id']}): {task['name']}\n描述：{task['description']}\n依赖任务：{dependencies}\n")
    
    # 格式要求部分
    format_instructions = """
请按照以下格式输出结果：
{
  "sequential_results": [
    {
      "task_id": "任务ID",
      "task_name": "任务名称",
      "result": "任务结果内容",
      "dependencies_used": ["使用的依赖任务ID列表"]
    },
    ...后续任务结果
  ],
  "overall_analysis": "基于所有任务结果的总体分析"
}
"""
    
    # 组合完整提示
    prompt = overview + "".join(task_definitions) + format_instructions
    return prompt

# 使用示例
analysis_tasks = [
    {
        'id': 'text_segmentation',
        'name': '文本分块',
        'description': '将文本分割成有意义的段落或块',
        'depends_on': []
    },
    {
        'id': 'key_extraction',
        'name': '关键词提取',
        'description': '提取文本中的关键概念和实体',
        'depends_on': ['text_segmentation']
    },
    {
        'id': 'sentiment_analysis',
        'name': '情感分析',
        'description': '分析文本的整体情感倾向',
        'depends_on': ['text_segmentation']
    },
    {
        'id': 'summary_generation',
        'name': '摘要生成',
        'description': '生成文本的简洁摘要，包含关键词和情感分析结果',
        'depends_on': ['key_extraction', 'sentiment_analysis']
    }
]

# 示例文本
input_text = "大语言模型在2025年取得了显著进展，特别是在多模态理解和跨任务处理能力方面。最新的GPT-4o和Claude 3 Opus模型展示了卓越的性能，能够同时处理文本、图像、音频等多种输入。然而，这些模型的计算资源需求仍然很高，限制了其在边缘设备上的应用。研究人员正在开发更高效的模型架构和压缩技术，以解决这一挑战。"

# 生成提示
sequential_prompt = create_sequential_analysis_prompt(input_text, analysis_tasks)
print(sequential_prompt)

class ParallelTaskPromptGenerator:
    def __init__(self, model_type="gpt-4o"):
        self.model_type = model_type
        self.task_template = """
任务 {task_id}: {task_name}
描述: {task_description}
输出要求: {output_requirements}
"""
    
    def generate_prompt(self, shared_input, tasks):
        """
        生成并行式跨任务提示
        
        参数：
        - shared_input: 所有任务共享的输入数据
        - tasks: 任务列表，每个任务包含id、名称、描述和输出要求
        
        返回：
        - 结构化的跨任务提示字符串
        """
        # 输入说明部分
        input_section = f"""请基于以下共享输入，并行完成指定的任务集：

共享输入：
{shared_input}

"""
        
        # 任务列表部分
        task_list = "并行任务列表：\n"
        for task in tasks:
            task_list += self.task_template.format(
                task_id=task['id'],
                task_name=task['name'],
                task_description=task['description'],
                output_requirements=task['output_requirements']
            ) + "\n"
        
        # 任务独立性说明
        independence_note = """
注意事项：
1. 所有任务都是相互独立的，可以并行处理
2. 请分别为每个任务提供完整且独立的结果
3. 确保结果的一致性和准确性

"""
        
        # 输出格式要求
        format_instructions = """
请按照以下JSON格式输出结果：
{
  "parallel_results": [
    {
      "task_id": "任务ID",
      "task_name": "任务名称",
      "result": {
        // 任务具体结果，格式根据任务要求
      }
    },
    // 其他任务结果
  ],
  "execution_summary": {
    "total_tasks": 总任务数,
    "completed_tasks": 已完成任务数,
    "shared_input_used": "共享输入使用情况描述"
  }
}
"""
        
        # 组合完整提示
        prompt = input_section + task_list + independence_note + format_instructions
        return prompt

# 使用示例
product_review = "这款智能手机有很多优点，包括出色的摄像头系统、流畅的性能和优秀的电池续航。屏幕显示效果也很好，色彩准确。然而，价格相对较高，而且重量略重，长时间手持不太舒适。另外，充电速度虽然支持快充，但与一些竞品相比还有提升空间。总体来说，这是一款高端旗舰手机，适合追求性能和摄影体验的用户。"

review_analysis_tasks = [
    {
        'id': 'sentiment_analysis',
        'name': '情感分析',
        'description': '分析产品评价的整体情感倾向，识别情感极性和强度',
        'output_requirements': '包含总体情感得分(1-10)、主要情感词列表和情感倾向总结'
    },
    {
        'id': 'feature_extraction',
        'name': '特性提取',
        'description': '提取评价中提到的产品特性和功能点',
        'output_requirements': '以列表形式返回所有提到的特性，每个特性包含名称和评价倾向'
    },
    {
        'id': 'recommendation_analysis',
        'name': '推荐分析',
        'description': '分析评价中的推荐意图和目标用户群体',
        'output_requirements': '包含推荐强度(1-10)、适合的用户类型和推荐理由'
    },
    {
        'id': 'price_value_analysis',
        'name': '性价比分析',
        'description': '分析评价中关于价格与价值的讨论',
        'output_requirements': '包含价格合理性评分(1-10)、价格与竞品比较和性价比总结'
    }
]

# 创建生成器并生成提示
generator = ParallelTaskPromptGenerator()
parallel_prompt = generator.generate_prompt(product_review, review_analysis_tasks)
print(parallel_prompt)

def create_hierarchical_prompt(main_task, task_hierarchy):
    """
    创建层次化跨任务提示
    
    参数：
    - main_task: 主任务描述
    - task_hierarchy: 任务层次结构，以字典形式表示
    
    返回：
    - 结构化的层次化跨任务提示
    """
    # 主任务说明
    main_section = f"""请按照以下层次结构完成复杂任务分析：

主任务：{main_task}

"""
    
    # 任务层次可视化
    hierarchy_visual = "任务层次结构：\n"
    hierarchy_visual += generate_hierarchy_visual(task_hierarchy, 0)
    hierarchy_visual += "\n"
    
    # 任务定义详情
    task_definitions = "详细任务定义：\n"
    task_definitions += generate_task_definitions(task_hierarchy, 0)
    task_definitions += "\n"
    
    # 处理流程指导
    process_guide = """
处理指南：
1. 从底层任务开始，逐步向上处理
2. 确保每个任务完成后再处理依赖它的上层任务
3. 上层任务应充分利用所有相关下层任务的结果
4. 在每个层次完成后进行总结，为上层任务提供清晰输入

"""
    
    # 输出格式要求
    format_instructions = """
请按照以下格式输出结果：
{
  "hierarchical_results": {
    "level_1": {
      "task_id": "任务ID",
      "task_name": "任务名称",
      "children": [
        {
          "task_id": "子任务ID",
          "task_name": "子任务名称",
          "result": "子任务结果",
          "children": [
            // 更深层次的子任务
          ]
        }
        // 其他同级子任务
      ],
      "result": "基于子任务结果的综合结果"
    }
  },
  "main_task_result": "最终主任务结果",
  "execution_path": "执行路径描述"
}
"""
    
    # 组合完整提示
    prompt = main_section + hierarchy_visual + task_definitions + process_guide + format_instructions
    return prompt

def generate_hierarchy_visual(hierarchy, level):
    """
    递归生成任务层次可视化
    """
    visual = ""
    indent = "  " * level
    
    if isinstance(hierarchy, dict):
        for task_id, task_info in hierarchy.items():
            task_name = task_info['name']
            visual += f"{indent}- {task_id}: {task_name}\n"
            if 'children' in task_info and task_info['children']:
                visual += generate_hierarchy_visual(task_info['children'], level + 1)
    
    return visual

def generate_task_definitions(hierarchy, level):
    """
    递归生成任务定义详情
    """
    definitions = ""
    indent = "  " * level
    
    if isinstance(hierarchy, dict):
        for task_id, task_info in hierarchy.items():
            task_name = task_info['name']
            task_desc = task_info['description']
            
            definitions += f"{indent}任务 {task_id}: {task_name}\n"
            definitions += f"{indent}  描述: {task_desc}\n"
            
            if 'children' in task_info and task_info['children']:
                definitions += f"{indent}  子任务: \n"
                definitions += generate_task_definitions(task_info['children'], level + 2)
            
            definitions += "\n"
    
    return definitions

# 使用示例
market_analysis_task = "对人工智能芯片市场进行全面分析，包括市场规模、主要玩家、技术趋势和投资机会"

market_analysis_hierarchy = {
    'market_size': {
        'name': '市场规模分析',
        'description': '评估AI芯片市场的总体规模、增长趋势和区域分布',
        'children': {
            'current_size': {
                'name': '当前市场规模',
                'description': '分析2024-2025年AI芯片市场的实际规模和主要细分市场占比'
            },
            'growth_forecast': {
                'name': '增长预测',
                'description': '预测未来3-5年AI芯片市场的增长率和驱动因素'
            },
            'regional_analysis': {
                'name': '区域分析',
                'description': '分析不同地区AI芯片市场的发展状况和特点'
            }
        }
    },
    'key_players': {
        'name': '主要厂商分析',
        'description': '识别并分析AI芯片市场的主要参与者及其竞争策略',
        'children': {
            'market_share': {
                'name': '市场份额分析',
                'description': '分析主要厂商的市场份额和排名变化'
            },
            'product_strategy': {
                'name': '产品策略',
                'description': '分析各厂商的产品定位和技术路线'
            },
            'competitive_advantage': {
                'name': '竞争优势',
                'description': '评估各厂商的核心竞争优势和劣势'
            }
        }
    },
    'tech_trends': {
        'name': '技术趋势分析',
        'description': '分析AI芯片领域的关键技术趋势和创新方向',
        'children': {
            'architecture_innovation': {
                'name': '架构创新',
                'description': '分析AI芯片架构的最新创新和发展方向'
            },
            'process_technology': {
                'name': '制程技术',
                'description': '评估先进制程技术在AI芯片中的应用状况'
            },
            'software_ecosystem': {
                'name': '软件生态',
                'description': '分析AI芯片配套软件和开发工具的发展情况'
            }
        }
    },
    'investment_opportunities': {
        'name': '投资机会分析',
        'description': '识别AI芯片领域的潜在投资机会和风险',
        'children': {
            'emerging_startups': {
                'name': '新兴创业公司',
                'description': '分析具有潜力的AI芯片创业公司和创新方向'
            },
            'investment_risk': {
                'name': '投资风险',
                'description': '评估AI芯片投资的主要风险因素'
            },
            'roi_forecast': {
                'name': '投资回报预测',
                'description': '预测不同细分领域的投资回报率'
            }
        }
    }
}

# 生成层次化提示
hierarchical_prompt = create_hierarchical_prompt(market_analysis_task, market_analysis_hierarchy)
print(hierarchical_prompt)

class ConditionalBranchPromptGenerator:
    def __init__(self):
        self.condition_template = """
条件 {condition_id}:
- 描述: {description}
- 判断标准: {criteria}
- 触发任务: {tasks}
"""
    
    def generate_prompt(self, input_content, conditions, default_path):
        """
        生成条件分支型跨任务提示
        
        参数：
        - input_content: 输入内容
        - conditions: 条件列表，每个条件包含id、描述、判断标准和触发任务
        - default_path: 默认任务路径
        
        返回：
        - 结构化的条件分支型跨任务提示
        """
        # 输入内容部分
        input_section = f"""
请分析以下输入内容，并根据条件判断规则动态选择适当的任务路径：

输入内容：
{input_content}

"""
        
        # 条件定义部分
        conditions_section = "条件判断规则：\n"
        for i, condition in enumerate(conditions, 1):
            tasks_str = ", ".join([task['id'] for task in condition['tasks']])
            conditions_section += self.condition_template.format(
                condition_id=condition['id'],
                description=condition['description'],
                criteria=condition['criteria'],
                tasks=tasks_str
            ) + "\n"
        
        # 默认路径部分
        default_tasks_str = ", ".join([task['id'] for task in default_path])
        default_section = f"默认路径：当没有条件满足时，执行以下任务：{default_tasks_str}\n\n"
        
        # 任务定义详情
        tasks_section = "任务详情定义：\n"
        
        # 收集所有任务
        all_tasks = {}
        for condition in conditions:
            for task in condition['tasks']:
                all_tasks[task['id']] = task
        
        # 添加默认任务
        for task in default_path:
            all_tasks[task['id']] = task
        
        # 生成任务定义
        for task_id, task_info in all_tasks.items():
            tasks_section += f"任务 {task_id}: {task_info['name']}\n"
            tasks_section += f"  描述: {task_info['description']}\n"
            tasks_section += f"  输出要求: {task_info['output_requirements']}\n\n"
        
        # 处理指南
        process_guide = """
处理步骤：
1. 首先仔细分析输入内容
2. 根据条件判断规则，确定满足哪些条件
3. 如果有多个条件满足，请选择最相关的一个
4. 执行对应条件触发的任务路径
5. 如果没有条件满足，执行默认路径
6. 在最终输出中说明条件判断过程和任务选择理由

"""
        
        # 输出格式要求
        format_instructions = """
请按照以下格式输出结果：
{
  "condition_analysis": {
    "input_category": "输入内容分类",
    "satisfied_conditions": ["满足的条件ID列表"],
    "selected_condition": "最终选择的条件ID",
    "selection_reason": "选择该条件的理由"
  },
  "executed_tasks": [
    {
      "task_id": "执行的任务ID",
      "task_name": "任务名称",
      "result": "任务执行结果"
    }
    // 按执行顺序排列的任务结果
  ],
  "task_sequence": "任务执行顺序描述",
  "final_answer": "基于所有执行任务的最终回答"
}
"""
    
        # 组合完整提示
        prompt = (input_section + conditions_section + default_section + 
                 tasks_section + process_guide + format_instructions)
        return prompt

# 使用示例
customer_inquiry = "我购买的智能手表上周突然无法充电，充电接口没有明显损坏，但连接充电器后完全没有反应。我已经尝试了不同的充电器和充电线，结果都是一样的。手表还在保修期内，我想知道应该如何处理这个问题？"

# 定义条件
inquiry_conditions = [
    {
        'id': 'hardware_issue',
        'description': '硬件故障相关问题',
        'criteria': '输入内容明确提及设备物理损坏、功能故障或硬件相关问题',
        'tasks': [
            {
                'id': 'troubleshooting',
                'name': '故障排查指南',
                'description': '提供详细的故障排查步骤和方法',
                'output_requirements': '包含分步骤的排查指南、可能的原因分析和每个步骤的预期结果'
            },
            {
                'id': 'warranty_info',
                'name': '保修信息查询',
                'description': '分析保修政策和用户权益',
                'output_requirements': '包含保修资格判断、保修期计算方法和保修范围内的处理流程'
            },
            {
                'id': 'repair_options',
                'name': '维修选项分析',
                'description': '提供各种维修渠道和选项的比较',
                'output_requirements': '包含官方维修、第三方维修、自行维修等选项的优缺点和推荐'
            }
        ]
    },
    {
        'id': 'software_issue',
        'description': '软件或使用相关问题',
        'criteria': '输入内容提及软件功能、设置问题、使用困难或更新相关问题',
        'tasks': [
            {
                'id': 'software_troubleshooting',
                'name': '软件故障排查',
                'description': '提供软件相关问题的解决步骤',
                'output_requirements': '包含重启、重置、更新等步骤的详细说明'
            },
            {
                'id': 'usage_guide',
                'name': '使用指南提供',
                'description': '提供相关功能的使用说明',
                'output_requirements': '包含功能说明、操作步骤和常见问题解答'
            }
        ]
    },
    {
        'id': 'purchase_refund',
        'description': '购买、退款或退货相关问题',
        'criteria': '输入内容提及购买流程、退款申请、退货政策或订单问题',
        'tasks': [
            {
                'id': 'refund_policy',
                'name': '退款政策解析',
                'description': '分析产品退款和退货政策',
                'output_requirements': '包含退款条件、申请流程和时间限制'
            },
            {
                'id': 'order_tracking',
                'name': '订单跟踪指南',
                'description': '提供订单状态查询和跟踪方法',
                'output_requirements': '包含查询步骤、联系方式和所需信息'
            }
        ]
    }
]

# 定义默认路径
default_task_path = [
    {
        'id': 'general_assistance',
        'name': '通用帮助',
        'description': '提供基本的客户支持指导',
        'output_requirements': '包含联系客服的方式、常见问题解答和后续步骤建议'
    },
    {
        'id': 'information_collection',
        'name': '信息收集建议',
        'description': '建议用户提供更详细的信息',
        'output_requirements': '包含需要补充的信息列表和如何提供这些信息的指导'
    }
]

# 生成条件分支提示
generator = ConditionalBranchPromptGenerator()
conditional_prompt = generator.generate_prompt(customer_inquiry, inquiry_conditions, default_task_path)
print(conditional_prompt)

class CrossTaskEvaluationFramework:
    def __init__(self, model_client, evaluation_metrics=None):
        self.model_client = model_client
        self.evaluation_metrics = evaluation_metrics or self._get_default_metrics()
        self.evaluation_results = []
    
    def _get_default_metrics(self):
        """
        获取默认评估指标
        """
        return {
            'efficiency': {
                'token_compression_ratio': {'weight': 0.3},
                'api_call_reduction': {'weight': 0.3},
                'time_savings': {'weight': 0.4}
            },
            'quality': {
                'task_accuracy': {'weight': 0.4},
                'result_consistency': {'weight': 0.3},
                'context_utilization': {'weight': 0.3}
            },
            'system': {
                'throughput_improvement': {'weight': 0.4},
                'resource_utilization': {'weight': 0.3},
                'error_handling': {'weight': 0.3}
            }
        }
    
    def evaluate(self, test_cases, runs_per_case=3):
        """
        评估跨任务提示的效果
        
        参数：
        - test_cases: 测试用例列表，每个测试用例包含原始任务提示和跨任务提示
        - runs_per_case: 每个测试用例运行的次数，用于减少随机性
        
        返回：
        - 综合评估结果
        """
        for i, test_case in enumerate(test_cases):
            case_results = []
            
            print(f"评估测试用例 {i+1}/{len(test_cases)}: {test_case['name']}")
            
            for run in range(runs_per_case):
                print(f"  运行 {run+1}/{runs_per_case}")
                
                # 评估单独任务执行
                individual_results = self._execute_individual_tasks(test_case['individual_tasks'])
                
                # 评估跨任务执行
                cross_task_result = self._execute_cross_task(test_case['cross_task_prompt'])
                
                # 计算评估指标
                metrics = self._calculate_metrics(individual_results, cross_task_result)
                
                # 评估质量（需要人工或额外模型评估）
                quality_scores = self._evaluate_quality(individual_results, cross_task_result, test_case['expected_outputs'])
                
                # 合并指标
                run_result = {
                    'test_case_id': test_case['id'],
                    'run_id': run + 1,
                    'efficiency_metrics': metrics['efficiency'],
                    'quality_metrics': quality_scores,
                    'system_metrics': metrics['system'],
                    'timestamp': datetime.now().isoformat()
                }
                
                case_results.append(run_result)
            
            # 计算测试用例的平均结果
            average_result = self._calculate_average_case_result(case_results)
            self.evaluation_results.append(average_result)
        
        # 生成综合报告
        comprehensive_report = self._generate_comprehensive_report()
        return comprehensive_report
    
    def _execute_individual_tasks(self, individual_tasks):
        """
        执行单独任务提示
        """
        results = []
        total_tokens = 0
        total_time = 0
        
        for task in individual_tasks:
            start_time = time.time()
            
            # 调用模型
            response = self.model_client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": task['prompt']}],
                temperature=0.2
            )
            
            end_time = time.time()
            processing_time = end_time - start_time
            
            # 计算token数量（这里使用简单估算，实际应使用tiktoken）
            prompt_tokens = len(task['prompt']) / 4  # 粗略估算
            completion_tokens = len(response.choices[0].message.content) / 4
            total_tokens += (prompt_tokens + completion_tokens)
            total_time += processing_time
            
            results.append({
                'task_id': task['id'],
                'task_name': task['name'],
                'prompt': task['prompt'],
                'response': response.choices[0].message.content,
                'prompt_tokens': prompt_tokens,
                'completion_tokens': completion_tokens,
                'processing_time': processing_time
            })
        
        return {
            'tasks': results,
            'total_tokens': total_tokens,
            'total_time': total_time,
            'api_calls': len(individual_tasks)
        }
    
    def _execute_cross_task(self, cross_task_prompt):
        """
        执行跨任务提示
        """
        start_time = time.time()
        
        # 调用模型
        response = self.model_client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": cross_task_prompt}],
            temperature=0.2
        )
        
        end_time = time.time()
        processing_time = end_time - start_time
        
        # 计算token数量（这里使用简单估算）
        prompt_tokens = len(cross_task_prompt) / 4
        completion_tokens = len(response.choices[0].message.content) / 4
        total_tokens = prompt_tokens + completion_tokens
        
        # 解析结果（假设返回JSON格式）
        try:
            parsed_result = json.loads(response.choices[0].message.content)
        except:
            parsed_result = {"error": "Failed to parse JSON"}
        
        return {
            'prompt': cross_task_prompt,
            'response': response.choices[0].message.content,
            'parsed_result': parsed_result,
            'prompt_tokens': prompt_tokens,
            'completion_tokens': completion_tokens,
            'total_tokens': total_tokens,
            'processing_time': processing_time,
            'api_calls': 1
        }
    
    def _calculate_metrics(self, individual_results, cross_task_result):
        """
        计算效率和系统指标
        """
        # 效率指标
        token_compression_ratio = (individual_results['total_tokens'] - 
                                 cross_task_result['total_tokens']) / 
                                 individual_results['total_tokens']
        
        api_call_reduction = (individual_results['api_calls'] - 
                            cross_task_result['api_calls']) / 
                            individual_results['api_calls']
        
        time_savings = (individual_results['total_time'] - 
                      cross_task_result['processing_time']) / 
                      individual_results['total_time']
        
        # 系统指标（这里简化计算）
        throughput_improvement = individual_results['api_calls'] / 1.0  # 调用次数减少倍数
        resource_utilization = token_compression_ratio  # 简化为token压缩率
        error_handling = 1.0 if 'error' not in cross_task_result['parsed_result'] else 0.0
        
        return {
            'efficiency': {
                'token_compression_ratio': max(0, token_compression_ratio),
                'api_call_reduction': api_call_reduction,
                'time_savings': max(0, time_savings)
            },
            'system': {
                'throughput_improvement': throughput_improvement,
                'resource_utilization': resource_utilization,
                'error_handling': error_handling
            }
        }
    
    def _evaluate_quality(self, individual_results, cross_task_result, expected_outputs):
        """
        评估质量指标
        """
        # 这里使用模型自评估，实际应用中可能需要人工评估或更复杂的方法
        
        # 准备评估提示
        evaluation_prompt = self._prepare_quality_evaluation_prompt(
            individual_results, cross_task_result, expected_outputs
        )
        
        # 调用评估模型
        eval_response = self.model_client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": evaluation_prompt}],
            temperature=0.1
        )
        
        # 解析评估结果
        try:
            eval_result = json.loads(eval_response.choices[0].message.content)
            return eval_result
        except:
            # 返回默认评估值
            return {
                'task_accuracy': 0.5,
                'result_consistency': 0.5,
                'context_utilization': 0.5
            }
    
    def _prepare_quality_evaluation_prompt(self, individual_results, cross_task_result, expected_outputs):
        """
        准备质量评估提示
        """
        # 构建详细的评估提示
        # ...
        return "评估提示内容"
    
    def _calculate_average_case_result(self, case_results):
        """
        计算测试用例的平均结果
        """
        # 实现平均计算逻辑
        # ...
        return average_result
    
    def _generate_comprehensive_report(self):
        """
        生成综合评估报告
        """
        # 实现报告生成逻辑
        # ...
        return report

# 使用示例
# 需要先初始化一个模型客户端，如OpenAI客户端
# framework = CrossTaskEvaluationFramework(model_client)
# results = framework.evaluate(test_cases)