【Datawhale AI 夏令营】Task2-精读baseline

发表于
【Datawhale AI 夏令营】Task1-精读baseline

一、提示工程(Prompt Engineering)

1.1 关键概念

  1. Prompt（提示）：给定的一段文本或指令，用于启动或引导AI模型响应。
  2. Prompt Design（提示设计）：设计良好的提示，目的是帮助模型更好理解输入的内容。
  3. Few-shot Learning（少样本学习）：提供给模型几个示例，让模型学习并应用于新的任务或数据点上。
  4. Zero-shot Learning（零样本学习）：不提供任何示例，仅通过明确的指令或描述来引导模型完成任务。
  5. In-context Learning（基于上下文的学习）：利用直接在提示中提供的上下文信息来进行学习和推理。
  6. Chain-of-thought Prompting（思维链提示）：通过在提示中加入中间步骤或思考过程来引导模型产生更复杂的推理和回答。

1.2 示例

1.2.1 输入问题

1
2
3
4
5
6
7
8
9
10
{
  'problem': 
        '有一群人和一些食物类型。下列是关于这些个体和食物的已知信息：\n\n1. 鸡肉是一种食物。\n2. 苹果是一种食物。\n3. 如果X吃了Y，且X活着，则Y是一种食物。\n4. Bill存活。\n5. Bill吃了花生。\n6. John吃所有食物。\n7. Sue吃所有Bill吃的食物。\n8. John喜欢所有食物。\n\n根据以上信息，回答以下选择题：', 
   'questions': [
        {
        'question': '选择题 1：\n谁喜欢吃花生？', 
        'options': ['Bill', 'Sue', 'John', 'None of the above']
        }
    ], 'id': 'round1_test_data_000'
}

1.2.2 转化

上面的问题字典转化为MD格式的prompt

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
你是一个逻辑推理专家，擅长解决逻辑推理问题。以下是一个逻辑推理的题目，形式为单项选择题。所有的问题都是（close-world assumption）闭世界假设，即未观测事实都为假。请逐步分析问题并在最后一行输出答案，最后一行的格式为"答案是：A"。题目如下：

###题目:
有一群人和一些食物类型。下列是关于这些个体和食物的已知信息：

鸡肉是一种食物。
苹果是一种食物。
如果X吃了Y，且X活着，则Y是一种食物。
Bill存活。
Bill吃了花生。
John吃所有食物。
Sue吃所有Bill吃的食物。
John喜欢所有食物。

根据以上信息，回答以下选择题：

###问题:
选择题 1：
谁喜欢吃花生？
A. Bill
B. Sue
C. John
D. None of the above

  1. 我们看到上述生成的prompt，首先告诉了模型，你要”作为“什么，然后”干“什么。(贫僧从东土大唐来，前往西天拜佛求经)

    • 作为什么：主要是让模型，把自己认为是那方面的专家。

    • 干什么：尽可能详细的告诉模型，你干什么事（分析问题、最后一行输入答案），有什么限制（例如上述题目是单选题、所有题目都是闭世界假设、最后一行的格式）

  2. “预防针”打完了，就要步入正题了。告诉大模型题目是什么，问题是什么，选项有哪些。

1.2.3 大模型响应

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
{
    "status_code": 200,
    "request_id": "18d43478-2aec-92be-8840-cd3bb08009cf",
    "code": "",
    "message": "",
    "output": {
        "text": null,
        "finish_reason": null,
        "choices": [
            {
                "finish_reason": "stop",
                "message": {
                    "role": "assistant",
                    "content": "我们可以通过分析给出的信息来解答这个问题。\n\n首先，我们知道Bill吃了花生（根据第5条信息），并且Bill还活着（根据第4条信息）。根据第3条规则，“如果X吃了Y，且X活着，则Y是一种食物”，我们可以推断出花生是一种食物。\n\n接下来，我们知道John吃所有食物（根据第6条信息），并且John喜欢所有食物（根据第8条信息）。既然花生是一种食物，那么John应该也吃花生，并且喜欢花生。\n\nSue吃所有Bill吃的食物（根据第7条信息），既然Bill吃了花生，那么Sue也应该吃花生。\n\n所以，喜欢吃花生的人有Bill、Sue和John。\n\n答案是：C. John\n\n但是，题目要求选择一个选项，而实际上Bill、Sue和John都喜欢吃花生。然而，在给出的选项中，只有John被明确列出作为喜欢吃花生的人（选项C）。因此，虽然Bill和Sue也喜欢吃花生，但根据题目要求，正确答案应当是包含在选项中的那个，即John。\n\n答案是：C. John"
                }
            }
        ]
    },
    "usage": {
        "input_tokens": 210,
        "output_tokens": 223,
        "total_tokens": 433
    }
}

我们可以看到，output.choices.message.content中的模型响应，首先分析了上述题目中的信息，在分析过程中，解答出了三个答案。

根据我们提示中说到的，“题目是单选题”，模型又继续推理出了更准确的一个答案。根据提示要求，将答案输出在了最后一行。

二、代码阅读

2.1 前言

baseline共分为了两部分。

第一部分：借助Qwen的API，进行问题的推理。

第二部分：因各种原因，导致Qwen没有推理出来的答案进行纠正和容错。

image-20240730225012884

偷个懒吧。直接从讲义中拷贝过来。

我目前是在职的Java开发，略懂点前端开发。python也能看懂，但仅停留在能看懂，上次接触python还是大二那年，过去也有个三四年了。

2.2 问题推理

2.2.1 call_qwen_api

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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',  # 将结果设置为消息格式。
    )
    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()

这段代码主要就是调用Qwen的api。

dashscope是阿里模型服务灵积的一个包吧。通过这个包，可以边界的调用，阿里云上的模型服务，进行推理等操作。其实说简单点，就是个SDK，阿里云将模型部署好，开发了个SDK，方便我们调用。具体的其他用法，可以参考官网的开发文档

我们继续往下看这段代码，如果调用返回状态为OK。那么将模型的对题目的响应返回出去。

如果状态不是OK，那么，就打印错误日志，抛出异常。

2.2.2 api_retry

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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

这段代码就是API的重试机制，简单的说，就是当API调用异常了，在5分钟内，每1分钟重新请求一次，共尝试5次。

再次过程中，捕获call_qwen_api方法中抛出的异常，存储到日志文件中。

2.2.3 get_prompt

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 这里定义了prompt推理模版

def get_prompt(problem, question, options):

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

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

### 题目:
{problem}

### 问题:
{question}
{options}
"""
    # print(prompt)
    return prompt

这段代码，就是真家伙了。将需要推理的题目，转换成提示词模板。

个人感觉，如果想提分，这里是可以进行优化的，具体优化方法，我不知道。因为刚开始接触大模型应用，不管是数据清洗、提示词工程、还是推理都不是很了解。如果各位有幸访问到我的博客，可以在评论区交流，我也会多多吸取大家的经验，十份感谢。

2.2.4 extract

1
2
3
4
5
6
7
8
9
10
# 这里使用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]

这段代码，主要就是从模型返回的响应中，获取答案。如果找不答案，那么就返回一个固定的A答案。算是容错的一种方式。

其实我觉得也可以随机返回一个答案，不过都是概率问题（猴子算法哈哈哈哈哈哈），对错就不好说了。

2.2.5 process_datas

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
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)):
            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 = 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"):
            # print('data',data)
            # 取出每个线程中的字典数据及对应的问题id
            data = future_data[future][0]
            problem_id = future_data[future][1]
            try:
                # 获取api运行结果
                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

这段代码，注释写的很清晰。就是创建线程池，并发处理。

我们调用的是Qwen的API，并发处理应该问题不大，毕竟阿里云是在国内云服务厂商也是大牛级别的存在，但是在自己机器上并发处理，不太清楚会不会把服务打死。

主要是我也没法尝试，在自己机器上会不会打死，因为公司发的这个ThinkPad，实在是让人抓马。攒钱，2026年顺便薅公司更新机器的羊毛，上MacBook Pro。

不扯淡了，这里我突然想起了一个问题，如果我们不是并发执行任务，而是想使用ChatGPT一样，先上来使用提示词告诉模型要做什么。然后一个问题一个问题的问，基于上下文的这种情况，会不会提高模型响应答案的准确率？

2.2.6 main

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def main(ifn, ofn):
    if os.path.exists(ofn):
        pass
    data = []
    # 按行读取数据
    with open(ifn) as reader:
        for line in reader:
            sample = json.loads(line)
            data.append(sample)
    datas = data
    # print(data)
    # 均匀地分成多个数据集
    return_list = process_datas(datas,MODEL_NAME)
    print(len(return_list))
    print("All tasks finished!")
    return return_list
    
if __name__ == '__main__':
# 这里给了一个抽取范例参考
    a = extract("""根据欧几里得算法，逐步解析计算两个数6和7的最大公约数（gcd）的步骤如下：

1. 判断6和7是否相等：不相等。
2. 判断6和7大小关系，7 > 6，所以用更大的数7减去较小的数6得到结果1。
3. 现在计算6和1的最大公约数。
4. 6 > 1，根据算法用更大的数6减去较小的数1得到结果5。
5. 再计算5和1的最大公约数。
6. 5 > 1，用5减去1得到结果4。
7. 再计算4和1的最大公约数。
8. 4 > 1，用4减去1得到结果3。
9. 再计算3和1的最大公约数。
10. 3 > 1，用3减去1得到结果2。
11. 再计算2和1的最大公约数。
12. 2 > 1，用2减去1得到结果1。
13. 最后计算1和1的最大公约数，两数相等，gcd即为这两个数，也就是1。

因此，6和7的最大公约数是1。

答案是：C.""")

    print(a)
    # 调用主函数
    return_list = main('round1_test_data.jsonl', 'upload.jsonl')

主函数，主要就是调用上述方法，让模型对题目进行推理。也没啥好讲的。

2.3 纠正和容错

2.3.1 has_complete_answer

1
2
3
4
5
6
def has_complete_answer(questions):
    # 这里假设完整答案的判断逻辑是：每个question都有一个'answer'键
    for question in questions:
        if 'answer' not in question:
            return False
    return True

检查答案是否完整，如果question有answer返回真，没有就返回假。

2.3.2 filter_problems

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
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)

问题过滤器，将所有的问题，存入一个字典，根据id的后三位，进行排序

2.3.3 find_missing_ids

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def find_missing_ids(dict_list):
    # 提取所有序号
    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)

这段代码，就是将没有返回响应的题目找出来。返回了一个缺失题目的id集合。

2.3.4 补错

1
2
3
4
5
6
7
8
9
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:]))

这段代码就是将2.3.3中缺失响应的题目的答案，固定填写为A。Datawhale的讲义中讲到，可以再推一遍，然后填写到这里。不过会加长运行时间。

2.3.5 输出结果

1
2
3
4
with open('upload.jsonl', 'w') as writer:
    for sample in sorted_data:
        writer.write(json.dumps(sample, ensure_ascii=False))
        writer.write('\n')

没啥可讲的，了解。

三、理解

通过代码的阅读，我们可以知道，Baseline到底带着我们干了什么。

主要就是生成提示词，调用模型，对问题进行推理，对模型返回的答案进行处理（记录、纠错、容错），输出结果。

周围好多人跟我说，你这不就是跟着Datawhale学了一下怎么调用API嘛。

非也！其实本Task，我起码了解的Prompt提示工程。在推理过程中，也产生了一些自己的想法和疑问。在专业助教老师没有讲解的情况下，作为一个成年人，一个“优秀”的新时代青年，一个“牛X” 的程序员。我们完全可以将想法和疑问，交给大模型，让大模型来对我们的想法和疑问进行分析和解答。

学会学习，不能读死书，掌握灵活的学习方法，先进的学习工具，才能让我们更快地进步。

声明如下：

本篇文章提示工程章节，参考了Datawhale助教老师讲义以及同义千问的解答。代码阅读部分，参考了Datawhale助教老师的讲义，根据自己的理解进行了梳理。