将您自己的 LLM 集成到 EA 中（第 5 部分）：使用 LLM 开发和测试交易策略（三） 适配器微调·进阶篇
「Transformer 输出后的适配器叠加与损失计算」
这段逻辑出现在自回归模型前向传播尾部，先取 transformer 主干的第 0 个输出作为隐藏状态，再逐个叠加外部适配器模块。
代码中 hidden_states = transformer_outputs[0] 拿到序列表征，随后 for i, adapter in enumerate(self.adapters): hidden_states = hidden_states + adapter(hidden_states) 把多个 adapter 的输出以残差方式并入，适配层数由 self.adapters 长度决定，常见配置为 2~4 个。
语言模型头 lm_head 将隐藏态映射为词表对数 lm_logits；当传入 labels 时，做标准下移位交叉熵：预测位错开一位，使 token < n 预测下一 token，损失由 nn.CrossEntropyLoss() 在展平后的张量上计算。
若 return_dict=False 则返回元组，否则封装为 CausalLMOutputWithCrossAttentions，携带 loss、logits 及注意力权重，方便在 MT5 的 Python 侧做 AIGC 信号提取时直接取 cross_attentions 观察价格文本对齐强度。
position_ids=position_ids, head_mask=head_mask, inputs_embeds=inputs_embeds, encoder_hidden_states=encoder_hidden_states, encoder_attention_mask=encoder_attention_mask, use_cache=use_cache, output_attentions=output_attentions, output_hidden_states=output_hidden_states, return_dict=return_dict, ) hidden_states = transformer_outputs[class="num">0] # Apply adapters for i, adapter in enumerate(self.adapters): hidden_states = hidden_states + adapter(hidden_states) lm_logits = self.lm_head(hidden_states) loss = None if labels is not None: # Shift so that tokens < n predict the next token shift_logits = lm_logits[..., :-class="num">1, :].contiguous() shift_labels = labels[..., class="num">1:].contiguous() # Flatten the tokens loss_fct = nn.CrossEntropyLoss() loss = loss_fct(shift_logits.view(-class="num">1, shift_logits.size(-class="num">1)), shift_labels.view(-class="num">1)) if not return_dict: output = (lm_logits,) + transformer_outputs[class="num">1:] class="kw">return ((loss,) + output) if loss is not None else output class="kw">return modeling_outputs.CausalLMOutputWithCrossAttentions( loss=loss, logits=lm_logits, past_key_values=transformer_outputs.past_key_values, hidden_states=transformer_outputs.hidden_states, attentions=transformer_outputs.attentions, cross_attentions=transformer_outputs.cross_attentions, )
把 Adapter 塞进 GPT-2 跑通微调
在自建 Adapter 类和 GPT2LMHeadModelWithAdapters 之后，真正动手只需把模块引入训练脚本，用 TextDataset 吃 train.txt，block_size 必须锁死为 60——这个值不是拍脑袋定的，要和预处理后样本长度对齐，否则批次拼接会错位。 数据收集器设 mlm=False，意味着走因果语言建模而非掩码任务，和 GPT-2 自回归本性一致。训练参数里 num_train_epochs=3、per_device_train_batch_size=32、save_strategy='no'，即三轮迭代、单卡 32 条样本、不落检查点，跑完直接 save_model 到 gpt2_Adapter-tuning 文件夹。 加载微调模型做推理时，必须用重写的 GPT2LMHeadModelWithAdapters.from_pretrained 而非原类，否则 Adapter 权重不会被读取；转 eval() 后拿 df 末行前 19 列拼 prompt 喂 generate，do_sample=True 看输出是否连贯，就能判断微调是否生效。 外汇与贵金属相关的文本生成实验同样面临模型过拟合与样本偏差的高风险，上述流程仅验证技术链路，不暗示任何实盘信号价值。
class="kw">import pandas as pd from transformers class="kw">import GPT2LMHeadModel, GPT2Tokenizer from transformers class="kw">import TextDataset, DataCollatorForLanguageModeling from transformers class="kw">import Trainer, TrainingArguments, modeling_outputs class="kw">import torch from torch class="kw">import nn class="kw">import torch.nn.functional as F dvc = &class="macro">#x27;cuda&class="macro">#x27; if torch.cuda.is_available() else &class="macro">#x27;cpu&class="macro">#x27; print(dvc) model_name_or_path = &class="macro">#x27;gpt2&class="macro">#x27; Tuned_model = "gpt2_Adapter-tuning" df = pd.read_csv(&class="macro">#x27;llm_data.csv&class="macro">#x27;) tokenizer = GPT2Tokenizer.from_pretrained(model_name_or_path) train_dataset = TextDataset(tokenizer=tokenizer, file_path="train.txt", block_size=class="num">60) data_collator = DataCollatorForLanguageModeling(tokenizer=tokenizer, mlm=False) training_args = TrainingArguments(output_dir=Tuned_model, overwrite_output_dir=True, num_train_epochs=class="num">3, per_device_train_batch_size=class="num">32, save_strategy=&class="macro">#x27;no&class="macro">#x27;, ) model = GPT2LMHeadModelWithAdapters.from_pretrained(model_name_or_path) trainer = Trainer(model=model, args=training_args, data_collator=data_collator, train_dataset=train_dataset,) trainer.train() trainer.save_model(Tuned_model) model = GPT2LMHeadModelWithAdapters.from_pretrained(Tuned_model) model.to(dvc) model.eval() prompt = &class="macro">#x27; &class="macro">#x27;.join(map(str, df.iloc[:, class="num">1:class="num">20].values[-class="num">1])) generated = tokenizer.decode(model.generate(tokenizer.encode(prompt, return_tensors=&class="macro">#x27;pt&class="macro">#x27;).to(dvc), do_sample=True,
◍ 给 GPT2 挂 Adapter 做轻量微调
想在 GPT2 上做领域适配又不重训全模型，Adapter 是务实选择：在每一层 Transformer 后插一个小瓶颈网络，只训这部分参数，显存和时长都压得下来。 下面这段代码定义了 Adapter 模块，降维到 64 维再升回原维度， dropout 设 0.1，权重用均值 0、标准差 0.02 的正态初始化，偏置归零。 随后把 Adapter 以 ModuleList 形式挂到 GPT2 的每个层上（config.n_layer 个）， forward 时先走原 transformer，再叠加 adapter 输出，实现参数高效微调。 实盘接小布盯盘的文本生成任务时，这种方案可能比全量微调更稳，尤其在你只有几千条行情点评语料的情况下。外汇与贵金属波动剧烈，任何模型输出都只是概率参考，高风险自负。
class Adapter(nn.Module): def __init__(self, in_features, bottleneck_features=class="num">64): super(Adapter, self).__init__() self.down_project = nn.Linear(in_features, bottleneck_features) self.up_project = nn.Linear(bottleneck_features, in_features) self.dropout = nn.Dropout(class="num">0.1) self.init_weights() def init_weights(self): nn.init.normal_(self.down_project.weight, mean=class="num">0.0, std=class="num">0.02) nn.init.constant_(self.down_project.bias, class="num">0) nn.init.normal_(self.up_project.weight, mean=class="num">0.0, std=class="num">0.02) nn.init.constant_(self.up_project.bias, class="num">0) def forward(self, hidden_states): hidden_states = self.down_project(hidden_states) hidden_states = F.relu(hidden_states) hidden_states = self.dropout(hidden_states) hidden_states = self.up_project(hidden_states) hidden_states = self.dropout(hidden_states) class="kw">return hidden_states class GPT2LMHeadModelWithAdapters(GPT2LMHeadModel): def __init__(self, config): super().__init__(config) self.adapters = nn.ModuleList([Adapter(config.n_embd) for _ in range(config.n_layer)])
「Transformer 输出后的适配器注入与损失计算」
这段逻辑发生在主干网络前向计算之后：先取出 transformer_outputs 的第 0 个元素作为 hidden_states，也就是序列最后一层的隐状态张量。 随后通过枚举遍历 self.adapters 列表，把每个适配器网络作用在 hidden_states 上，并以残差方式叠加（hidden_states = hidden_states + adapter(hidden_states)）。这种写法让轻量适配器在不改动预训练主干的前提下微调表示，参数量通常只占原模型的 5%~10%。 lm_head 把隐状态映射为词表维度的 lm_logits。若传入了 labels，则执行标准的语言模型移位：shift_logits 去掉末尾 token，shift_labels 去掉首个 token，使第 n 个位置预测第 n+1 个位置，再用 CrossEntropyLoss 在展平后的张量上算 loss。 返回分支兼顾了旧式元组与字典式输出：非 return_dict 时拼回 (loss,) + output，否则封装成 CausalLMOutputWithCrossAttentions，把 past_key_values、cross_attentions 等一并带出，方便后续生成或调试。
past_key_values=past_key_values,
attention_mask=attention_mask,
token_type_ids=token_type_ids,
position_ids=position_ids,
head_mask=head_mask,
inputs_embeds=inputs_embeds,
encoder_hidden_states=encoder_hidden_states,
encoder_attention_mask=encoder_attention_mask,
use_cache=use_cache,
output_attentions=output_attentions,
output_hidden_states=output_hidden_states,
return_dict=return_dict,
)
hidden_states = transformer_outputs[class="num">0]
# Apply adapters
for i, adapter in enumerate(self.adapters):
hidden_states = hidden_states + adapter(hidden_states)
lm_logits = self.lm_head(hidden_states)
loss = None
if labels is not None:
# Shift so that tokens < n predict the next token
shift_logits = lm_logits[..., :-class="num">1, :].contiguous()
shift_labels = labels[..., class="num">1:].contiguous()
# Flatten the tokens
loss_fct = nn.CrossEntropyLoss()
loss = loss_fct(shift_logits.view(-class="num">1, shift_logits.size(-class="num">1)), shift_labels.view(-class="num">1))
if not return_dict:
output = (lm_logits,) + transformer_outputs[class="num">1:]
class="kw">return ((loss,) + output) if loss is not None else output
class="kw">return modeling_outputs.CausalLMOutputWithCrossAttentions(
loss=loss,
logits=lm_logits,
past_key_values=transformer_outputs.past_key_values,
hidden_states=transformer_outputs.hidden_states,
attentions=transformer_outputs.attentions,
cross_attentions=transformer_outputs.cross_attentions,
)把适配器模型跑起来做推理
上面这段 Python 碎片接在前面训练流程之后，核心是把带 Adapter 的 GPT-2 真正训完并拿去生成。训练参数里 num_train_epochs=3、per_device_train_batch_size=32，意味着单卡一次塞 32 条样本、跑 3 个 epoch，save_strategy='no' 说明中间不落盘检查点，只在末尾 trainer.save_model 一次性导出。 初始化时 GPT2LMHeadModelWithAdapters.from_pretrained 加载基座再加适配器权重，Trainer 接好 data_collator 和 train_dataset 后直接 train()。训完用同一路径 reload 进推理，model.to(dvc) 把计算搬去指定设备，eval() 关掉 dropout 等随机项。 推理样本取自 df 最后一行第 1 到 19 列拼成 prompt，model.generate 开 do_sample=True、max_length=200，输出经 tokenizer.decode 去特殊符后打印。你在本地复现时要注意 df 列宽必须 ≥20，否则切片会越界；外汇与贵金属行情序列用这套生成，仅作概率性辅助参考，实盘高风险。
num_train_epochs=class="num">3, per_device_train_batch_size=class="num">32, save_strategy= &class="macro">#x27;no&class="macro">#x27;, ) # Initialize model with adapters model = GPT2LMHeadModelWithAdapters.from_pretrained(model_name_or_path) trainer = Trainer(model=model, args=training_args, data_collator=data_collator, train_dataset=train_dataset,) trainer.train() trainer.save_model(Tuned_model) # Load the model for inference model = GPT2LMHeadModelWithAdapters.from_pretrained(Tuned_model) model.to(dvc) model.eval() prompt = &class="macro">#x27; &class="macro">#x27;.join(map(str, df.iloc[:, class="num">1:class="num">20].values[-class="num">1])) generated = tokenizer.decode(model.generate(tokenizer.encode(prompt, return_tensors=&class="macro">#x27;pt&class="macro">#x27;).to(dvc), do_sample=True, max_length=class="num">200)[class="num">0], skip_special_tokens=True) print(f"test the model:{generated}")