ML2025 Homework 2 使用大模型生成代码解决数据预测问题 Disease Prediction
·
任务简介 - 作为数据科学家的 AI 助手
- 使用 LLMs 自动编写代码
- 只需了解数据集，LLMs 就能为每项任务生成定制化解决方案
任务 Disease Prediction
给定美国某州过去 3 天的调查结果，进而预测第 3 天新增检测阳性病例的百分比。
要用llm生成的代码运行进行预测，不要修改llm生成的代码
数据集描述
代码结构
计划
我们将按照以下步骤进行操作：
1.加载必要的程序库和数据。
2.通过处理缺失值和特征缩放对数据进行预处理。
3.将数据集划分为训川练集和测试集。
4.在训练数据上训川练回归模型。
5.对测试数据进行预测。
6.将预测结果保存至提交文件。
开源模型排行榜
https://huggingface.co/spaces/open-llm-leaderboard/open_llm_leaderboard#/
提示－不同LLMs
更强的LLM可能有用，但不总是如此…….
我们如何知道哪个LLM更好呢？
模型大小？预训练语料库？开放LLM排行榜？
你可以从HuggingFace上选择模型。
禁止使用诸如GPT-4o、Gemini等闭源LLMAPl。
当然还有so many ways,这里我说下我所采用的方法
- 使用开源的大模型qwen-coder-plus-latest
- 优化各种提示词
- 增大steps数
- 增大num_drafts数
- 温度0.3
下面是核心代码
# openai模型api
import os
import openai
api_key=os.getenv('OPENAI_API_KEY') or 'your_api_key'
base_url=os.getenv('OPENAI_BASE_URL') or "your_base_url"
# model=os.getenv('OPENAI_MODEL') or "your_model"
model='qwen-coder-plus-latest'
# print(api_key,base_url,model)
client=openai.OpenAI(
api_key=api_key,
base_url=base_url,
)
def generate_response(prompt,system_prompt="",max_tokens=8192,temperature=0.1,stream=False):
response=client.chat.completions.create(
model=model,
messages=[
{
'role': 'system',
'content':system_prompt
},
{
'role':'user',
'content':prompt
}
]
,max_tokens=max_tokens,
temperature=temperature,
stream=stream
)
if stream:
# 处理流式输出
def generate():
for chunk in response:
if chunk.choices[0].delta.content is not None:
yield chunk.choices[0].delta.content
return generate()
else:
# 处理非流式输出
return response.choices[0].message.content
print("模型api调用成功！")
# # 非流式调用
# result = generate_response("你是谁")
# print(result)
# # 流式调用
# for chunk in generate_response(prompt="喜欢什么", stream=True):
# print(chunk, end="", flush=True)
import random
from typing import Any, Callable, cast
import re
import sys
import json
import humanize
from pydantic import BaseModel
ExecCallbackType = Callable[[str, bool], ExecutionResult]
class Agent:
def __init__(
self,
cfg,
journal: Journal,
):
super().__init__()
self.cfg = cfg
self.journal = journal
self.data_preview: str | None = None
def search_policy(self) -> Node | None:
"""Select a node to work on (or None to draft a new node)."""
search_cfg = self.cfg.agent.search
# initial drafting
if len(self.journal.draft_nodes) < search_cfg.num_drafts:
return None
# debugging
if random.random() < search_cfg.debug_prob:
# nodes that are buggy + leaf nodes + debug depth < max debug depth
debuggable_nodes = [
n
for n in self.journal.buggy_nodes
if n.is_leaf
]
if debuggable_nodes:
return random.choice(debuggable_nodes)
# back to drafting if no nodes to improve
good_nodes = self.journal.good_nodes
if not good_nodes:
return None
# greedy
greedy_node = self.journal.get_best_node()
return greedy_node
def plan_and_code_query(self, system_message, user_message, retries=3) -> tuple[str, str]:
"""Generate a natural language plan + code in the same LLM call and split them apart."""
completion_text = None
for _ in range(retries):
response = generate_response(
prompt=user_message,
system_prompt=system_message,
)
completion_text = response
code = extract_code(completion_text)
nl_text = extract_text_up_to_code(completion_text)
if code:
return nl_text, code
print("Plan + code extraction failed, retrying...")
print("Final plan + code extraction attempt failed, giving up...")
return "", completion_text
def _draft(self) -> Node:
# ================ TODO: ask LLM agents to come up with a solution and then implement ================
# system_prompt = "You are an AI agent."
# user_prompt = [
# "You have to come up with a solution for machine learning task and then implement this solution in Python."
# f"The task is to {str(self.cfg.task_goal)} ",
# f'All the provided input data is stored in "{self.cfg.data_dir}" directory.',
# f"{str(self.data_preview)}",
# 'You have to save the predictions result on testing set in "/content/submission.csv".',
# 'Note that the testing file DOES NOT have the target column.'
# ]
system_prompt = """You are an expert AI agent specializing in time series prediction and data analysis.
You are running on Ubuntu 22.04.5 LTS with Python 3.11.
Your task is to develop a solution that predicts testing probabilities based on survey data.
You should:
1. Analyze the data structure and features
2. Design an appropriate model architecture
3. Implement the solution in Python
4. Ensure the code is well-documented and follows best practices
5. Save predictions to the specified output file
Focus on creating a robust solution that minimizes Mean Squared Error (MSE)."""
user_prompt = [
"Task: Develop a machine learning model to predict testing probabilities.",
f"Goal: {str(self.cfg.task_goal)}",
f"Data Location: {self.cfg.data_dir}",
f"Data Overview:\n{str(self.data_preview)}",
"Requirements:",
"1. Save predictions to '/content/submission.csv'",
"2. Note that the testing file DOES NOT have the target column",
"3. Implement proper data preprocessing",
"4. Use appropriate model selection and validation",
"\nNeed to provide:",
"1. A detailed plan explaining your approach",
"2. Full Python implementation",
"\nNote:",
"1. Token limit is 8192.",
]
system_message = system_prompt
user_message = "\n".join(user_prompt)
print(f'user_message: {user_message}')
plan, code = self.plan_and_code_query(system_message=system_message, user_message=user_message)
return Node(plan=plan, code=code)
def _improve(self, parent_node: Node) -> Node:
# ================ TODO: ask LLM agent to improve drafts ================
# system_prompt = "You are an AI assistant."
# user_prompt = [
# f"Task description: {str(self.cfg.task_goal)} "
# f"Memory: {str(self.journal.generate_summary())} "
# f"Previous solution: Code: {str(wrap_code(parent_node.code))} "
# ]
system_prompt = """You are an expert AI agent tasked with improving an existing prediction model.
You are running on Ubuntu 22.04.5 LTS with Python 3.11.
Your goal is to enhance the model's performance by:
1. Analyzing the current implementation
2. Identifying potential improvements
3. Implementing optimizations
4. Maintaining or improving code quality
Focus on reducing the Mean Squared Error (MSE) while keeping the solution practical and efficient."""
user_prompt = [
"Task: Improve the existing prediction model",
f"Original Goal: {str(self.cfg.task_goal)}",
"Previous Solutions:",
f"{str(self.journal.generate_summary())}",
"Current Implementation:",
f"Code:\n{str(wrap_code(parent_node.code))}",
"\nNeed to provide:",
"1. Analysis of current implementation",
"2. Specific improvement plan",
"3. Enhanced implementation"
]
system_message = system_prompt
user_message = "\n".join(user_prompt)
print(f'user_message: {user_message}')
plan, code = self.plan_and_code_query(system_message=system_message, user_message=user_message)
return Node(plan=plan, code=code, parent=parent_node)
def _debug(self, parent_node: Node) -> Node:
# ================ TODO: ask LLM agent to debug ================
# system_prompt = "You are an AI agent."
# user_prompt = [
# f"Task description: {str(self.cfg.task_goal)}\n\n",
# f"Previous (buggy) implementation: {str(wrap_code(parent_node.code))}\n\n",
# f"Execution output: {str(wrap_code(parent_node.term_out, lang=''))}\n\n",
# str(self.data_preview)
# ]
system_prompt = """You are an expert AI agent debugger specializing in machine learning code.
You are running on Ubuntu 22.04.5 LTS with Python 3.11.
Your task is to:
1. Analyze the error or issue in the current implementation
2. Identify the root cause
3. Propose and implement a fix
4. Ensure the solution maintains good performance
Focus on creating a robust fix that resolves the issue while maintaining or improving the model's performance."""
user_prompt = [
"Task: Debug and fix the prediction model",
f"Original Goal: {str(self.cfg.task_goal)}",
"Current Implementation:",
f"Code:\n{str(wrap_code(parent_node.code))}",
"Execution Output:",
f"{str(wrap_code(parent_node.term_out, lang=''))}",
"Data Context:",
f"{str(self.data_preview)}",
"\nNeed to provide:",
"1. Analysis of the error/issue",
"2. Root cause identification",
"3. Proposed fix",
"4. Corrected implementation"
]
system_message = system_prompt
user_message = "\n".join(user_prompt)
print(f'user_message: {user_message}')
plan, code = self.plan_and_code_query(system_message=system_message, user_message=user_message)
return Node(plan=plan, code=code, parent=parent_node)
def update_data_preview(
self,
):
self.data_preview = data_preview_generate(cfg.data_dir)
def step(self, exec_callback: ExecCallbackType):
if not self.journal.nodes or self.data_preview is None:
self.update_data_preview()
parent_node = self.search_policy()
if parent_node is None:
result_node = self._draft()
elif parent_node.is_buggy:
result_node = self._debug(parent_node)
else:
result_node = self._improve(parent_node)
self.parse_exec_result(
node=result_node,
exec_result=exec_callback(result_node.code, True),
)
self.journal.append(result_node)
def parse_exec_result(self, node: Node, exec_result: ExecutionResult):
node.absorb_exec_result(exec_result)
system_prompt = "You are an AI assistant. You are running on Ubuntu 22.04.5 LTS with Python 3.11."
# ================ TODO: ask LLM agent to extract evaluation result from the execution output. ================
# save log file
# user_prompt = f"""
# The task is:
# {self.cfg.task_goal}
# The code implementation is:
# {wrap_code(node.code)}
# The execution output is:
# {wrap_code(node.term_out, lang="")}
# """
user_prompt = f"""
Task: Evaluate the prediction model implementation
Original Goal:
{self.cfg.task_goal}
Implementation:
{wrap_code(node.code)}
Execution Output:
{wrap_code(node.term_out, lang="")}
Need to provide a structured analysis including:
1. Execution Status (Success/Error)
2. Performance Metrics (especially MSE)
3. Issues or Concerns (if any)
4. Overall Assessment"""
system_message = system_prompt
# user_message = " ".join(user_prompt)
user_message = user_prompt
print(f'user_message: {user_message}')
response = generate_response(
prompt=user_message,
system_prompt=system_message,
)
# ================ TODO: evaluation ================
# you can force the LLM to structure the output to extract the metric
# reference: https://python.useinstructor.com/integrations/llama-cpp-python/#llama-cpp-python
# node.analysis = response.summary
# node.is_buggy = (
# response.is_buggy
# or node.exc_type is not None
# or response.metric is None
# )
try:
parsed_response = json.loads(response)
node.analysis = parsed_response.get("assessment", "")
node.metric = parsed_response["mse"] if parsed_response.get("mse") is not None else 0.0
node.is_buggy = (
parsed_response.get("execution_status", "Error") == "Error"
or node.exc_type is not None
or node.metric is None
)
except Exception as e:
print("Failed to parse evaluation response:", e)
node.analysis = "Failed to parse evaluation response."
node.is_buggy = False
node.metric = 0.0
# ================ TODO: config ================
config = {
# experiment configurations
"exp_name": "ML2025_HW2",
"data_dir": Path("./ML2025Spring-hw2-public").resolve(),
# the description of the task
"task_goal": "Given the survey results from the past two days in a specific state in the U.S.,\
predict the probability of testing positive on day 3. \
The evaluation metric is Mean Squared Error (MSE).",
"agent": {
# the number of iterations
"steps": 5,
"search": {
# decide whether to debug or improve
"debug_prob": 0.5,
# the number of draft generated before improving/debugging
"num_drafts": 5,
},
},
}
cfg = Config(config)
完整代码
https://github.com/hllqkb/ML2025Spring_HUNG_YI_LEE_Machine_Learning_Homework/blob/master/ML2025Spring-hw2/ML2025Spring_hw2_bossline.ipynb
根据下面的Baseline表
前往比赛提交并且查看我们的成绩
比赛链接
输出结果0.947
‼️ 目前仍难以获得可复现且稳定的结果。 ‼️
当然你想拿高分的话可以偷偷作弊一下，直接自己写一个，肯定比ai写的好
0.8300
best.py完整代码
# 引入必要模块
# 数据操作
import pandas as pd
import numpy as np
import math
import csv
import random
import os
from monai.utils import set_determinism
from torch.utils.data import Dataset, DataLoader
# 进度条
from tqdm import tqdm
# 特征选择
from sklearn.feature_selection import SelectKBest, f_regression
# pytorch
import torch
import torch.nn as nn
# 绘制图像
from torch.utils.tensorboard import SummaryWriter
# set random seed
def split_valid_set(data_set, valid_ratio, seed):
np.random.seed(seed)
indices = np.random.permutation(len(data_set))
valid_size = int(len(data_set) * valid_ratio)
valid_idx = indices[:valid_size]
train_idx = indices[valid_size:]
return data_set[train_idx], data_set[valid_idx]
def select_features(train_data,valid_data,test_data,select_all=True):
# choose label
y_train=train_data[:,-1]
y_valid=valid_data[:,-1]
# choose all features except label
raw_x_train=train_data[:,:-1]
raw_x_valid=valid_data[:,:-1]
raw_x_test=test_data
if select_all:
feat_idx=list(range(raw_x_train.shape[1]))
all_to_remove = {0}
feat_idx = [x for x in feat_idx if x not in all_to_remove]
else:
# Feature selection
feat_idx = [34, 35, 36, 43, 46, 47, 51, 52, 53,
54, 61, 64, 65, 69, 70, 71, 72, 79, 82, 83]
return raw_x_train[:,feat_idx],raw_x_valid[:,feat_idx],raw_x_test[:,feat_idx],y_train,y_valid
class COVID19Dataset(Dataset):
def __init__(self, features,target=None):
if target is None:
# predict
self.target=target
else:
# train
self.target = torch.FloatTensor(target)
self.features = torch.FloatTensor(features)
def __getitem__(self,idx):
# idx means index of sample
if self.target is None:
return self.features[idx]
else:
return self.features[idx],self.target[idx]
def __len__(self):
return len(self.features)
# define model
class MyModel(nn.Module):
def __init__(self, input_dim):
super(MyModel, self).__init__()
hidden_dims = config['layer'] # 如 [128, 64, 32]
layers = []
dims = [input_dim] + hidden_dims
for in_dim, out_dim in zip(dims[:-1], dims[1:]):
layers.append(nn.Linear(in_dim, out_dim))
layers.append(nn.GELU()) # 更先进的激活函数
# 可选：加入 Dropout 或 BatchNorm 以提升泛化能力
# layers.append(nn.Dropout(0.2))
layers.append(nn.Linear(dims[-1], 1)) # 输出层：无激活
self.layers = nn.Sequential(*layers)
def forward(self, x):
x = self.layers(x)
return x.squeeze(1) # 输出 (B)
# Set CFG
device=torch.device('cuda' if torch.cuda.is_available() else 'cpu')
config={
'layer': [128, 64, 32],
'k':20,
'seed': 3407,
'select_all': False,
'normalize': True,
'batch_size': 128,
'lr': 0.00035099698616997447,
'n_epochs': 20000,
'valid_ratio': 0.2,
'early_stop': 400,
'save_path': './models/model.ckpt',
'beta1': 0.95, # β₁
'eps': 1e-7, # ϵ
}
# train
def train(train_loader,valid_loader,model,config,device):
criterion=nn.MSELoss()
optimizer = torch.optim.Adam(
model.parameters(),
lr=config['lr'],
betas=(config['beta1'], 0.999), # 调整 β₁
eps=config['eps'], # 调整 ϵ
# weight_decay=0 # 可选：是否加入 L2 正则化
)
writer=SummaryWriter()
best_loss=math.inf
early_stop_count=0
step=0
for epoch in range(config['n_epochs']):
loss_record=[]
model.train()
train_bar=tqdm(train_loader)
for x,y in train_bar:
x,y=x.to(device),y.to(device)
optimizer.zero_grad()
outputs=model(x)
loss=criterion(outputs,y)
loss.backward()
optimizer.step()
loss_record.append(loss.detach().item())
step+=1
# validate
model.eval()
valid_loss_record=[]
valid_bar=tqdm(valid_loader)
for x,y in valid_bar:
x,y=x.to(device),y.to(device)
with torch.no_grad():
outputs=model(x)
loss=criterion(outputs,y)
valid_loss_record.append(loss.detach().item())
valid_loss=np.mean(valid_loss_record)
writer.add_scalar('valid_loss',valid_loss,step)
# save best model
if valid_loss<best_loss:
best_loss=valid_loss
torch.save(model.state_dict(),config['save_path'])
early_stop_count=0
print('Saving model with loss {:.3f}...'.format(best_loss))
else:
early_stop_count+=1
if early_stop_count>=config['early_stop']:
return best_loss
return best_loss
# set same seed
set_determinism(seed=config['seed'])
# 读取数据（保留 DataFrame 或 NumPy）
train_data = pd.read_csv('./ML2025Spring-hw2-public/train.csv').values
test_data = pd.read_csv('./ML2025Spring-hw2-public/test.csv').values
# 切分数据（不转成 Subset）
train_data, valid_data = split_valid_set(train_data, config['valid_ratio'], config['seed'])
# 检查标签是否有 NaN
if np.isnan(train_data[:, -1]).sum() > 0 or np.isnan(valid_data[:, -1]).sum() > 0:
raise ValueError("Labels contain NaN values. Please check your data.")
# Normalization with Min-Max scaling 可以加一个点
if config['normalize']:
train_min = np.min(train_data[:, 35:-1], axis=0) # 计算每列最小值
train_max = np.max(train_data[:, 35:-1], axis=0) # 计算每列最大值
# 防止除以0的情况，可以加一个很小的数
epsilon = 1e-8
train_range = train_max - train_min + epsilon
# 归一化训练集
train_data[:, 35:-1] = (train_data[:, 35:-1] - train_min) / train_range
# 用相同的min和range归一化验证集
valid_data[:, 35:-1] = (valid_data[:, 35:-1] - train_min) / train_range
# 用相同的min和range归一化测试集
test_data[:, 35:] = (test_data[:, 35:] - train_min) / train_range
x_train,x_valid,x_test,y_train,y_valid=select_features(train_data,valid_data,test_data,config['select_all'])
# convert to tensor
train_dataset=COVID19Dataset(x_train,y_train)
valid_dataset=COVID19Dataset(x_valid,y_valid)
test_dataset=COVID19Dataset(x_test)
# pin memory能保存数据到GPU的高速缓存中，能加快数据加载速度,shuffle=True表示每个epoch打乱数据
train_loader=DataLoader(train_dataset,batch_size=config['batch_size'],shuffle=True,pin_memory=True)
valid_loader=DataLoader(valid_dataset,batch_size=config['batch_size'],shuffle=True,pin_memory=True)
test_loader=DataLoader(test_dataset,batch_size=config['batch_size'],shuffle=False,pin_memory=True)
# create model
model=MyModel(input_dim=x_train.shape[1]).to(device)
# train
print('Training model...')
best_loss=train(train_loader,valid_loader,model,config,device)
print(f'Best validation loss: {best_loss:.3f}')
def predict(test_loader,model,device):
model.eval()
preds=[]
with torch.no_grad():
for x in tqdm(test_loader):
x=x.to(device)
outputs=model(x)
# 把数据从GPU转移到CPU，才能转换为numpy数组
preds.append(outputs.detach().cpu())
preds = torch.cat(preds, dim=0).numpy()
return preds
# predict
predictions=predict(test_loader,model,device)
# save predictions
def save_pred(preds, file):
''' Save predictions to specified file '''
with open(file, 'w') as fp:
writer = csv.writer(fp)
writer.writerow(['id', 'tested_positive'])
for i, p in enumerate(preds):
writer.writerow([i, p])
save_pred(predictions, './predictions.csv')
print('Predictions saved in ./predictions.csv')
代码库：https://github.com/hllqkb/ML2025Spring_HUNG_YI_LEE_Machine_Learning_Homework
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
21
0- 0
已为社区贡献2条内容
所有评论(0)