Datawhale AI夏令营第三期Task1学习笔记

Task 1 任务安排：

1. 运行官方提供的Baseline代码，提交结果以获得首个评分
2. 观看助教直播讲解，了解比赛规则及注意事项
3. 学习助教讲解的Baseline代码，尝试理解代码逻辑

前两项相对简单，这里重点分析第三项内容，深入讲解Baseline中的核心方法函数。

核心函数分析：批量处理数据

下面是用于批量处理题目数据的关键函数实现：

def batch_process(dataset, model_name):
    output = []
    task_queue = {}
    task_counter = 0
    
    with ThreadPoolExecutor(max_workers=16) as executor:
        # 提交所有任务
        for item in tqdm(dataset, desc="提交任务中", total=len(dataset)):
            problem_text = item['problem']
            for q_idx, question_item in enumerate(item['questions']):
                prompt_text = build_prompt(
                    problem_text,
                    question_item['question'],
                    question_item['options']
                )
                
                future_obj = executor.submit(call_api_with_retry, model_name, prompt_text)
                task_queue[future_obj] = (item, q_idx)
                time.sleep(0.6)
                task_counter += 1
        
        # 处理返回结果
        for completed_task in tqdm(as_completed(task_queue), total=task_counter, desc="处理结果中"):
            original_item = task_queue[completed_task][0]
            question_idx = task_queue[completed_task][1]
            
            try:
                api_response = completed_task.result()
                parsed_result = parse_answer(api_response)
                original_item['questions'][question_idx]['answer'] = parsed_result
                output.append(original_item)
                
            except Exception as err:
                logger.error(f"处理数据失败: {original_item}, 错误: {err}")
    
    return output

函数详解

函数参数

• dataset: 包含题目和选项信息的数据集列表
• model_name: 指定使用的模型标识符

返回值

返回处理完成的结果列表，包含每个题目的答案信息。

实现逻辑

1. 初始化阶段：
   • 创建空列表用于存储最终结果
   • 实例化ThreadPoolExecutor，设置16个工作线程
   • 创建字典映射Future对象与原始数据
   • 初始化计数器用于追踪任务总数
2. 任务提交阶段：
   • 使用tqdm显示任务提交进度
   • 遍历数据集中的每条记录
   • 对每条记录中的所有问题逐个处理
   • 构建包含问题和选项的提示文本
   • 通过executor.submit提交异步任务
   • 每次提交后休眠0.6秒，避免API调用过于频繁
3. 结果处理阶段：
   • 使用as_completed获取已完成的任务
   • 从Future对象中提取原始数据和问题索引
   • 调用result()获取API返回结果
   • 使用解析函数提取答案，更新原始数据
   • 将处理后的数据加入结果列表

技术要点

• 并发处理: ThreadPoolExecutor实现了多线程并发，大幅提升处理效率
• 异步执行: 通过submit方法实现任务的异步提交和执行
• 异常捕获: try-except结构确保单个任务失败不影响整体流程
• 进度监控: tqdm库实时显示任务提交和处理进度

使用前提

• 需要安装tqdm和concurrent.futures库
• 外部需定义build_prompt和call_api_with_retry函数
• 需配置好日志记录器logger
• parse_answer函数用于从API响应中提取答案

调用示例

import concurrent.futures
from tqdm import tqdm
import time
import logging

# 模拟API调用函数
def call_api_with_retry(model_name, prompt):
    time.sleep(1)
    return "答案: C"

# 模拟提示构建函数
def build_prompt(problem, question, options):
    return f"{problem}\n{question}\n{options}"

# 解析答案函数
def parse_answer(response):
    if "C" in response:
        return "C"
    return "A"

# 日志配置
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)

# 测试数据
test_data = [
    {
        "problem": "样本问题",
        "questions": [
            {"question": "题目1", "options": "A B C D"},
            {"question": "题目2", "options": "E F G H"}
        ]
    }
]

# 执行处理
results = batch_process(test_data, "gpt-model")

# 输出结果
for item in results:
    print(item)

以上示例演示了如何构建测试数据并调用批量处理函数，最终返回带有答案的处理结果。