目录

赛题介绍Task1——跑通baseline结果

Task2——baseline代码解析部分核心代码

Task3——lora微调、vllm部署、多路投票loravllm部署多路投票

总结todo整理当前上分记录具体

后续上分思路上分之路

Datawhale AI 夏令营 第三期逻辑推理学习笔记

本期此项目手册：

https://datawhaler.feishu.cn/wiki/NOVDw5OtLiKJhlkbmoXc8nCinMf

赛题介绍

本次比赛提供基于自然语言的逻辑推理问题，涉及多样的场景，包括关系预测、数值计算、谜题等，期待选手通过分析推理数据，利用机器学习、深度学习算法或者大语言模型，建立预测模型。

初赛数据集为逻辑推理数据，其中训练集中包含500条训练数据，测试集中包含500条测试数据。每个问题包括若干子问题，每个子问题为单项选择题，选项不定（最多5个）。目标是为每个子问题选择一个正确答案。推理答案基于闭世界假设（closed-world assumption），即未观测事实或者无法推断的事实为假。

具体的，每条训练数据包含 <code>content, questions字段，其中content是题干，questions为具体的子问题。questions是一个子问题列表，每个子问题包括options和answer字段，其中options是一个列表，包含具体的选项，按照ABCDE顺序排列，answer是标准答案。

数据集格式如下:

{ 'id': 'round_train_data_001',

'problem': '有一个计算阶乘的递归程序。该程序根据给定的数值计算其阶乘。以下是其工作原理：\n\n当数字是0时，阶乘是1。\n对于任何大于0的数字，其阶乘是该数字乘以其前一个数字的阶乘。\n根据上述规则，回答以下选择题：',

'questions': [

{ 'question': '选择题 1：\n3的阶乘是多少？\n',

'options': ('3', '6', '9', '12'),

'answer': 'B'},

{ 'question': '选择题 2：\n8的阶乘是多少？\n',

'options': ('5040', '40320', '362880', '100000'),

'answer': 'B'},

{ 'question': '选择题 3：\n4的阶乘是多少？\n',

'options': ('16', '20', '24', '28'),

'answer': 'C'},

{ 'question': '选择题 4：\n3的阶乘是9吗？\n',

'options': ('是', '否'),

'answer': 'B'}

]

}

测试集中不带answer字段，待推理阶段输出测试结果为提交文件。

本次评估指标为所有子问题的回答准确率，每个子问题权重相同。满分为1。

学习计划

Task1——跑通baseline

调用大模型api，推理输出结果文件提交评分

Task1用的baseline代码讲解见task2。——数据处理、多线程调用api等。

采用qwen1.5-1.8b-chat。灵积平台有其他限时免费开放的模型，如qwen2-1.5b-instruct。

申领大模型key：https://dashscope.console.aliyun.com/apiKey

（开通 DashScope(阿里云灵积平台) 可以获赠一些其他模型的限时免费使用额度，大部分有效期为30天。)算力平台使用魔搭notebook。选择CPU环境，此task是调用大模型的api进行访问，利用多线程提高效率。

结果

不做修改直接运行baseline，由于网络或其他原因，部分题目丢失，其子问题都默认选A。分值只有0.3494。

听群内说72B大模型能到0.8+，而7B加微调也能到0.8。

Task2——baseline代码解析

对task1的baseline代码进行讲解https://datawhaler.feishu.cn/wiki/CvNRwdXDHimxJskZaArcvYqDnIc

整体代码介绍

整体代码主要包括答案生成和纠错与结果文件生成两个大模块。

答案生成部分包括大模型的处理函数、大模型返回结果抽取、多线程处理及答案生成的启动。

这里代码核心是大模型部分，动手能力强的小伙伴可以从这里入手开始自己的上分之路~为了保证整体代码性能使用多线程处理请求。

纠错与结果生成部分存在的目的是由于目前使用了api调用在线开源大模型，因为网络、模型能力等原因会导致有一些结果会出现缺失。（比如大模型回答时，没有明确给出ABCD的结果，而返回的空值。也有时因为网络retry模块机会使用结束后，依然没有提取到结果会跳过某个问题。）

部分核心代码

注意以下为ipynb的内容，按顺序执行，启动函数后，才逐步执行去重、纠错、补错、存储结果文件（略）。

——注意官方代码与后台评分机制契合，无需过多在意数据构成和保存方式。可自己加入一些策略上分，最后保存内容符合后台评分所需即可。

环境

<code>!pip install scipy openai tiktoken retry dashscope loguru

# 注意：这里需要填入阿里云灵积平台申请的key。

dashscope.api_key="sk-"code>

import包略……

api调用大模型

def call_qwen_api(MODEL_NAME, query):

# 这里采用dashscope的api调用模型推理，通过http传输的json封装返回结果

messages = [

{ 'role': 'user', 'content': query}]

response = dashscope.Generation.call(

MODEL_NAME,

messages=messages,

result_format='message', # set the result is message format.code>

)

if response.status_code == HTTPStatus.OK:

print(response)

return response['output']['choices'][0]['message']['content']

else:

print('Request id: %s, Status code: %s, error code: %s, error message: %s' % (

response.request_id, response.status_code,

response.code, response.message

))

raise Exception()

api_retry 这个函数是当大模型调用api时可能会导致出错中断的问题，为了保证每个问题都被大模型处理过，我们需要设置一个反复尝试的函数。# 最大尝试次数5次 # 再次尝试等待时间 60秒。如果出现错误我们存储到日志文件。

def api_retry(MODEL_NAME, query):

# 最大尝试次数

max_retries = 5

# 再次尝试等待时间

retry_delay = 60 # in seconds

attempts = 0

while attempts < max_retries:

try:

return call_qwen_api(MODEL_NAME, query)

except Exception as e:

attempts += 1

if attempts < max_retries:

logger.warning(f"Attempt { attempts} failed for text: { query}. Retrying in { retry_delay} seconds...")

time.sleep(retry_delay)

else:

logger.error(f"All { max_retries} attempts failed for text: { query}. Error: { e}")

raise

get_prompt prompt的模版函数，通过字符串处理的方式拼接完整的prompt（markdown格式的）

# 这里定义了prompt推理模版

def get_prompt(problem, question, options):

# 枚举生成对应格式的选项文本

options = '\n'.join(f"{ 'ABCDEFG'[i]}. { o}" for i, o in enumerate(options))

# prompt模板，字符串拼接，格式化

prompt = f"""你是一个逻辑推理专家，擅长解决逻辑推理问题。以下是一个逻辑推理的题目，形式为单项选择题。所有的问题都是（close-world assumption）闭世界假设，即未观测事实都为假。请逐步分析问题并在最后一行输出答案，最后一行的格式为"答案是：A"。题目如下：

### 题目:

{ problem}

### 问题:

{ question}

{ options}

"""

# print(prompt)

return prompt

抽取大模型回答答案

通过抽取函数可以将大语言模型生成的结果抽取成答案对应的选项，这里的匹配原则和prompt呼应。可以看到prompt要求【最后一行的格式为"答案是：A"】这样的规范，那么采用正则表达式re.compile方法匹配到答案对应的选项。当匹配为空时，默认选"A"。

# 这里使用extract抽取模获得抽取的结果

def extract(input_text):

ans_pattern = re.compile(r"答案是：(.)", re.S)

problems = ans_pattern.findall(input_text)

# print(problems)

if(problems == ''):

return 'A'

return problems[0]

多线程发送api调用

def process_datas(datas,MODEL_NAME):

results = []

# 定义线程池 选择16线程

with ThreadPoolExecutor(max_workers=16) as executor:

# 这里我们使用future_data 存储每个线程的数据

future_data = { }

# 这里的lens记录了调用api的次数，也就是我们每个问题背景下的所有子问题之和。

lens = 0

# 送入多线程任务

# 这里每个data下是一个问题背景，其中包含多个子问题。

for data in tqdm(datas, desc="Submitting tasks", total=len(datas)):code>

problem = data['problem']

# 这里面我们用enumerate方法每次循环得到问题的序号id和实际的问题。

for id,question in enumerate(data['questions']):

prompt = get_prompt(problem,

question['question'],

question['options'],

)

# 这里送入线程池等待处理，使用api_retry，向api_retry传入MODEL_NAME, prompt参数

# future是Futrue对象，代表异步任务

future = executor.submit(api_retry, MODEL_NAME, prompt)

# 每个线程我们存储对应的json问题数据以及问题序号id，这样我们就能定位出执行的是哪个子问题

future_data[future] = (data,id)

time.sleep(0.6) # 控制每0.6秒提交一个任务 防止接口超过并发数

lens += 1

# 处理多线程任务

for future in tqdm(as_completed(future_data), total=lens, desc="Processing tasks"):code>

# print('data',data)

# 取出每个线程中的字典数据及对应的问题id

data = future_data[future][0]

problem_id = future_data[future][1]

try:

# result()获取api运行结果，如果future未完成会阻塞

res = future.result()

# 抽取大语言模型返回结果

extract_response = extract(res)

# print('res',extract_response)

# 装入answer字段

data['questions'][problem_id]['answer'] = extract_response

# 在结果列表中新增数据字典

results.append(data)

# print('data',data)

except Exception as e:

logger.error(f"Failed to process text: { data}. Error: { e}")

return results

启动函数

读取数据集转为json，然后调用process_datas函数（根据数据构造prompt、多线程发起api请求、抽取回答最终返回list）。

回答去重与排序

可能是网络延迟与重试机制导致有重复。

将一个问题背景下的所有问题存入同一个字典，并按id序号排序。

def has_complete_answer(questions):

# 这里假设完整答案的判断逻辑是：每个question都有一个'answer'键

for question in questions:

if 'answer' not in question:

return False

return True

def filter_problems(data):

result = []

problem_set = set()

for item in data:

# print('处理的item' ,item)

problem = item['problem']

if problem in problem_set:

# 找到已存在的字典

for existing_item in result:

if existing_item['problem'] == problem:

# 如果当前字典有完整答案，替换已存在的字典

if has_complete_answer(item['questions']):

existing_item['questions'] = item['questions']

existing_item['id'] = item['id']

break

else:

# 如果当前字典有完整答案，添加到结果列表

if has_complete_answer(item['questions']):

result.append(item)

problem_set.add(problem)

return result

return_list = filter_problems(return_list)

# 排序工作 通过id字段后三位代表序号

sorted_data = sorted(return_list, key=lambda x: int(str(x['id'])[-3:]))

print(sorted_data)

纠错

找到丢失的问题。

def find_missing_ids(dict_list):

# 提取所有序号（后三位即可，一共500个）

extracted_ids = { int(d['id'][-3:]) for d in dict_list}

# 创建0-500的序号集合

all_ids = set(range(500))

# 找出缺失的序号

missing_ids = all_ids - extracted_ids

return sorted(missing_ids)

# 示例字典列表

dict_list = sorted_data

# 找出缺失的序号

missing_ids = find_missing_ids(dict_list)

print("缺失的序号:", missing_ids)

len(missing_ids)

补错

针对空缺的列表我们进行补错，让每个answer字段默认填充为A（就全蒙A，效果肯定好不了哪里去），也可考虑重新发起请求处理一遍。

data = []

with open('round1_test_data.jsonl') as reader:

for id,line in enumerate(reader):

if(id in missing_ids):

sample = json.loads(line)

for question in sample['questions']:

question['answer'] = 'A'

sorted_data.append(sample)

sorted_data = sorted(sorted_data, key=lambda x: int(str(x['id'])[-3:]))