这是基于第二届世界科学智能大赛 逻辑推理赛道：复杂推理能力评估(点击跳转赛事官网)的一篇从0基础入门AI的实践课，适合于零基础小白，本篇文章是我自己的学习笔记，供大家参考
1.大语言模型介绍
大语言模型的概念
大语言模型（英文：Large Language Model，缩写LLM），是一种人工智能模型，旨在理解和生成人类语言。
我们可以想象一下，LLM就像一个图书馆的守护者，它阅读了数不尽的书籍，从古老的传说到现代的科学论文，它吸收了无数知识，并能够根据这些知识回答各种问题。
通常，大语言模型 (LLM) 指包含数十亿（Billion或更多）参数的语言模型，这些模型在大量的文本数据上进行训练，例如国外的有GPT-4、PaLM 、Galactica 和 LLaMA 等，国内的有Kimi、ChatGLM、文心一言、通义千问、讯飞星火等
大语言模型的能力和特点
1.大语言模型的能力
大语言模型（LLM）与以前的预训练语言模型（PLM）的主要区别在于其涌现能力。我们可以将PLM比作单车道道路，虽然可以行驶，但能力有限；LLM则像高速公路，能够处理更多的信息流，支持更复杂的任务并行处理。这种能力在小型模型中不明显，但在大型模型中显著。例如：
- 上下文学习：首次由GPT-3引入，允许模型在提供自然语言指令或多个任务示例的情况下，通过理解上下文并生成相应输出来执行任务。
- 指令遵循：通过指令微调，LLM可以根据任务指令执行未见过的任务，展示出强大的泛化能力。
- 逐步推理：通过"思维链（Chain of Thought, CoT）"策略，它使得LLM能够进行逐步推理，解决多步逻辑问题，例如数学问题，逻辑谜题，法律案例分析等。
2.大预言模型的特点
- 巨大的规模：我们想象LLM是一个庞大的图书馆，里面收藏了数十亿甚至数千亿本书，每本书代表一个参数。这个图书馆如此之大，以至于它能够涵盖人类语言的每一个细微差别和复杂结构。
- 预训练和微调：LLM的预训练阶段就像是在一个大熔炉中锻造一把剑，通过大量的文本数据，它学会了语言的基本规则和模式。微调阶段则像是对这把剑进行精细的打磨，使其能够适应特定的任务，比如在法律或医学领域中使用。
- 上下文感知：LLM的上下文感知能力就像是侦探在解决案件，它能够理解每个线索（文本片段）之间的联系，并根据这些线索推断出整体情况。
- 多语言支持：LLM的多语言能力就像是拥有多国护照的旅行者，能够自由地穿梭于不同的文化和语言之间，促进跨文化交流。
- 多模态支持：LLM的多模态能力可以比作一个能够同时欣赏音乐、绘画和诗歌的艺术家，它能够理解和生成结合了文本、图像和语音的复杂内容。
- 涌现能力：LLM的涌现能力就像是深海中的暗流，随着模型规模的增大，它能够展现出意想不到的能力和性能，解决更加复杂的任务。
- 多领域应用：LLM的多领域应用能力就像是瑞士军刀，具有多种工具和功能，能够在文本生成、翻译、检索等不同领域中发挥作用。
- 伦理和风险问题：LLM的伦理和风险问题可以比作一把双刃剑，它拥有巨大的潜力，但同时也需要我们谨慎地处理，以避免产生有害内容或侵犯隐私。
2.大语言模型解题方案介绍
大模型推理
推理能力，简而言之，就是机器理解基本逻辑规则，并运用这些规则来分析和解决问题。它类似于一个逻辑拼图，机器需要识别每一块拼图的位置，然后根据这些信息拼凑出完整的图像。例如，如果我们知道“所有狗都喜欢吃肉”，并且我们还知道“汤姆是一只狗”，那么通过逻辑推理，我们可以得出结论：“狗喜欢吃肉”。
尽管现代的人工智能模型，如大型语言模型（LLM），已经开始展现出这种推理能力，但它们是如何学会这一技能的，背后的技术细节却相当复杂。全球的科学家们正在不懈地研究这一问题，他们试图揭开让AI模型变得更智能的秘密。许多研究机构和公司都在这一领域投入了巨大的资源，因为掌握这一技术将可能引领人工智能的未来潮流。
推理能力的培养和发展，就像是在智慧的海洋中航行，需要不断地探索和发现。每一项技术进步，都是向着更深层次智能迈进的一步。而在这个过程中，我们不仅需要技术的智慧，还需要对伦理和责任的深思熟虑，确保人工智能的发展能够造福人类社会。
大模型推理实现最常用方法——提示工程(Prompt Engineering)
提示工程可以被看作是一门指导艺术，它是一门较新的学科，关注提示词开发和优化，帮助用户将大语言模型（Large Language Model, LLM）用于各场景和研究领域，它教会我们如何与LLM进行有效的对话，使其能够理解我们的需求并提供恰当的响应。
输入问题：
这里选取了Datawhale AI夏令营中train的第一个问题。可以看到有问题背景、提问、以及选项这样结构的数据字典。
{'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'}
转化后的prompt：
通过对数据进行了精心整理和格式化，转换成了Markdown格式的提示。这种格式使得大型语言模型能够轻松理解信息，并进行有效的深入分析。简而言之，这是我们确保语言模型能够准确捕捉并分析问题核心的关键步骤。
你是一个逻辑推理专家，擅长解决逻辑推理问题。以下是一个逻辑推理的题目，形式为单项选择题。所有的问题都是（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
大模型回答：
为了便于大家操作，暂时使用了未经微调的开源模型Qwen2-1.5b-instruct的API。（虽然比赛要求最终提交时必须使用大型模型并上传到环境，但大家现在可以在成本较低的API上先行测试。待测试完成后，再将环境打包成Docker文件进行上传。）
通过观察到，大型模型通过
output['choices']['message']['content']
字段返回了其推理的结果。通过这种方式，我们让大型模型成功完成了我们的推理任务。
{
"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
}
}
3.代码介绍
整体代码包括两个大模块分别是答案生成和纠错与结果文件生成。
在系统中，答案生成是一个关键环节，它涉及以下几个步骤：
- 大模型处理：编写了专门的函数来与大型语言模型进行交互，确保问题能够被正确处理。
- 结果提取：一旦模型给出了回答，代码会从中提取出有用的答案部分。
- 多线程应用：为了提升效率，采用了多线程技术，这样可以同时处理多个问题，加快整体的响应速度。
- 答案生成启动：有一个启动机制，用来触发整个答案生成流程。
对于那些喜欢动手实践的朋友，大模型的处理部分是一个很好的起点，你们可以在这里开始自己的探索和优化之旅。
文档里面还特别设计了一个纠错和结果生成的机制，以应对使用在线API调用大型模型时可能遇到的问题。由于网络不稳定或模型本身的限制，有时我们可能得不到完整的答案（例如，模型没有给出明确的选项ABCD，而是返回了空值）。此外，如果网络重试机制耗尽了所有机会，但依然无法获取到答案，系统会智能地跳过这个问题，以避免整个流程的延误。
3.1 环境配置
Datawhale官方有提供详细的速通文档：从零入门 AI 逻辑推理-飞书云文档
step1:安装需要的库
pip install scipy openai tiktoken retry dashscope loguru
step2:配置API
# 注意：这里需要填入你的key~ 咱们在第二步申请的。
dashscope.api_key="sk-"
step3:导入需要的环境
import json
import os
from pprint import pprint
import re
from tqdm import tqdm
import random
import uuid
import openai
import tiktoken
import json
import numpy as np
import requests
from retry import retry
from scipy import sparse
#from rank_bm25 import BM25Okapi
#import jieba
from http import HTTPStatus
import dashscope
from concurrent.futures import ThreadPoolExecutor, as_completed
from loguru import logger
import json
import time
from tqdm import tqdm
logger.remove() # 移除默认的控制台输出
logger.add("logs/app_{time:YYYY-MM-DD}.log", level="INFO", rotation="00:00", retention="10 days", compression="zip")
MODEL_NAME = 'qwen2-7b-instruct'
3.2 答案生成部分
step1:大模型部分
**call_qwen_api **这个函数目的是通过输入模型名称、prompt，完成大模型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', # 设置返回结果的格式为消息格式。
)
if response.status_code == HTTPStatus.OK:
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 request failed with status code: {}".format(response.status_code))
**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
# 这里定义了prompt推理模版
def get_prompt(problem, question, options):
options = '\n'.join(f"{'ABCDEFG'[i]}. {o}" for i, o in enumerate(options))
prompt = f"""# Role：逻辑推理专家
## Background：
在逻辑推理问题中，通常存在一组给定的事实和规则，需要通过这些信息来推导出正确答案。
## Attention：
逻辑推理是一项需要高度集中注意力和细致分析的任务，每一个细节都可能影响最终的结论。
## Profile：
- Author: pp
- Version: 2.1
- Language: 中文
- Description: 作为逻辑推理专家，我具备分析和解决复杂逻辑问题的能力。
### Skills:
- 能够理解和分析给定的逻辑规则和事实。
- 能够运用逻辑推理技巧，如演绎推理和归纳推理。
- 能够识别和排除与问题无关的信息。
- 能够从多个角度审视问题，寻找可能的解决方案。
## Goals:
- 准确理解题目中的逻辑规则和事实。
- 通过逻辑分析，排除不可能的选项。
- 确定最符合逻辑的选项作为答案。
## Constrains:
- 遵循闭世界假设，未观测到的信息视为假。
- 避免主观臆断，所有推理必须基于题目提供的信息。
- 保持客观和逻辑性，不引入外部信息。
## Workflow:
1. 首先，仔细阅读题目，理解所有给定的事实和规则。
2. 然后，分析问题，确定需要解决的关键点。
3. 接着，根据逻辑规则，逐一检查每个选项。
4. 排除那些与题目事实相矛盾或逻辑上不可能的选项。
5. 最后，选择最符合逻辑的选项作为答案。
## OutputFormat:
- 答案格式应为："答案是：[选项]"。
## Suggestions:
- 仔细审题，不要急于下结论。
- 尝试从不同角度审视问题，避免思维定势。
- 记录下每个选项的分析过程，以便于复查。
- 如果遇到难题，不要放弃，尝试重新审视问题和选项。
## Initialization
作为逻辑推理专家，我将遵循上述约束，使用中文与用户交流，并在分析完成后介绍我的工作流程。
"""
# print(prompt)
return prompt
step2: 结果提取
通过抽取函数可以将大语言模型生成的结果抽取成答案对应的选项，这里的匹配原则和prompt呼应。我们可以看到prompt要求【最后一行的格式为"答案是：A"】这样的规范，那么我们采用正则表达式re.compile方法匹配到答案对应的选项。当我们匹配为空时，我们默认选"A"。
# 抽取答案的函数
def extract_answer(text):
pattern = re.compile(r'答案是：([A-Z])')
match = pattern.search(text)
if match:
return match.group(1)
else:
return "A"
# 进一步处理抽取结果的函数
def process_answer(answer):
# 这里可以添加任何基于抽取结果的逻辑
print(f"处理后的答案为: {answer}")
# 主函数，整合抽取和处理逻辑
def main_extract_and_process(text):
answer = extract_answer(text) # 调用抽取函数
process_answer(answer) # 使用抽取的结果进行进一步处理
# 示例文本
text = """
这是一个示例文本，其中包含了一些信息。
最后一行是答案的指示：答案是：C
"""
# 调用主函数
main_extract_and_process(text)
step3: 多线程应用
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
step4: 启动函数
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')
3.3 纠错与结果文件生成
step1: 去重与排序
将一个问题背景下的所有问题存入同一个字典，并按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)
step2: 纠错
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)
step3：补错
针对空缺的列表我们进行补错，让每个answer字段默认填充为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:]))
step4：存储文件
with open('upload.jsonl', 'w') as writer:
for sample in sorted_data:
writer.write(json.dumps(sample, ensure_ascii=False))
writer.write('\n')
版权归原作者 柒小毓 所有， 如有侵权，请联系我们删除。