阅读量:0
Task 2 大语言模型Baseline精读
1.环境配置
2.答案生成
2.1大模型部分
2.2抽取函数
re.compile(r"答案是: (.)", re.S)：这一行代码使用re.compile编译了一个正则表达式模式。re.compile用于编译正则表达式模式，从而创建一个模式对象。这个模式对象可以反复使用，以匹配输入文本中的特定格式的字符串。通过这种方式，代码可以高效地提取符合特定格式的答案字符。r"答案是: (.)"：这是正则表达式模式，表示我们希望匹配一个以“答案是: ”开头，后面跟着一个字符（用括号捕获），即.。在正则表达式中，.表示匹配除换行符外的任何单个字符。re.S：这是一个标志，表示正则表达式在匹配时将点号.扩展为可以匹配所有字符，包括换行符。re.S也称为re.DOTALL。
ans_pattern：这是编译后的模式对象，可以用于后续的匹配操作。ans_pattern.findall(input_text)：使用编译后的模式对象ans_pattern在输入文本input_text中查找所有匹配的子串。findall方法返回所有非重叠匹配的列表，每个匹配项都是捕获组中的内容。problems：这是一个列表，包含所有匹配的答案字符。if(problems == '')：检查problems列表是否为空。如果是空的，意味着没有找到匹配的答案字符。return 'A'：如果没有找到匹配的答案字符，默认返回'A'。return problems[0]：否则，返回匹配列表中的第一个字符，即找到的答案。在这段代码
2.3多线程处理
2.4启动函数
### 代码功能概述
函数 `process_datas` 使用多线程处理一组包含问题背景及其子问题的数据，通过调用 API 获取每个问题的答案，并将结果整合到原始数据中返回。
### 示例数据
```python
datas = [
{
"problem": "在计算机科学中，Python是什么？",
"questions": [
{
"question": "Python是一种什么类型的编程语言？",
"options": ["A. 低级语言", "B. 高级语言", "C. 机器语言", "D. 汇编语言"]
},
{
"question": "Python的创始人是谁？",
"options": ["A. Dennis Ritchie", "B. James Gosling", "C. Guido van Rossum", "D. Bjarne Stroustrup"]
}
]
}
]
MODEL_NAME = "example_model"
```
### 主要步骤
1. **导入必要模块**
```python
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
from tqdm import tqdm
import re
```
2. **定义辅助函数**
- `get_prompt`: 生成 API 调用的提示字符串。
- `api_retry`: 模拟 API 调用，返回结果字符串。
- `extract`: 使用正则表达式提取 API 返回结果中的答案。
```python
def get_prompt(problem, question, options):
return f"{problem}\n{question}\n选项: {', '.join(options)}"
def api_retry(MODEL_NAME, prompt):
time.sleep(random.uniform(0.1, 0.5)) # 模拟API调用延迟
return f"答案是: B" # 模拟返回结果
def extract(response):
match = re.search(r"答案是: (.)", response)
if match:
return match.group(1)
return "A"
```
3. **定义主函数 `process_datas`**
```python
def process_datas(datas, MODEL_NAME):
results = []
with ThreadPoolExecutor(max_workers=16) as executor:
future_data = {}
lens = 0
for data in tqdm(datas, desc="Submitting tasks", total=len(datas)):
problem = data['problem']
for id, question in enumerate(data['questions']):
prompt = get_prompt(problem, question['question'], question['options'])
future = executor.submit(api_retry, MODEL_NAME, prompt)
future_data[future] = (data, id)
time.sleep(0.6)
lens += 1
for future in tqdm(as_completed(future_data), total=lens, desc="Processing tasks"):
data = future_data[future][0]
problem_id = future_data[future][1]
try:
res = future.result()
extract_response = extract(res)
data['questions'][problem_id]['answer'] = extract_response
results.append(data)
except Exception as e:
print(f"Failed to process text: {data}. Error: {e}")
return results
```
### 运行示例
```python
results = process_datas(datas, MODEL_NAME)
for result in results:
print(result)
```
### 预期输出
```python
{
"problem": "在计算机科学中，Python是什么？",
"questions": [
{
"question": "Python是一种什么类型的编程语言？",
"options": ["A. 低级语言", "B. 高级语言", "C. 机器语言", "D. 汇编语言"],
"answer": "B"
},
{
"question": "Python的创始人是谁？",
"options": ["A. Dennis Ritchie", "B. James Gosling", "C. Guido van Rossum", "D. Bjarne Stroustrup"],
"answer": "B"
}
]
}
```
### 总结
1. **多线程处理**：使用 `ThreadPoolExecutor` 提高处理效率。
2. **任务提交与等待**：通过 `executor.submit` 提交任务，使用 `as_completed` 等待任务完成。
3. **API 调用与结果提取**：模拟 API 调用，通过正则表达式提取答案。
4. **错误处理**：捕获并记录处理中的异常。
这个例子展示了如何高效地并行处理大量数据，并从外部 API 获取结果后进行处理。
