浅析优化文本提示技术 —— TextGrad
引言
大模型领域,现在除了不断推出各种底层大模型外,还涌现了许多包含复杂组件的复合系统,包括框架、工具等。
TextGrad VS DSPy
先来从大的概念范围看看 TextGrad 与 DSPy 的对比:
- TextGrad:
(1)将 LLM 文本反馈“反向传播”到各组件中,无需手动调整,实现优化复合系统;
(2)在实际的各种任务中,它具备多功能性,包括:问答场景、分子优化和治疗计划等;
(3)TextGrad 可显著提升性能,例如使用 GPT-4o 在 Google-Proof 问答任务中零样本准确率从51%提高到55%。
(4)TextGrad 通过提供自然语言梯度增强可解释性;
- DSPy:
(1)引入一种编程模型,用于将语言模型管道抽象为文本转换,同时自动优化这些管道,以最大化实现响应目标。
(2)DSPy 把 LM 调用表示为参数化模块,使用编译器优化这些模块组合,实现LM管道的系统化开发和优化。
(3)DSPy在数学文字题等任务上有超过25%和65%的性能提升。
(4)DSPy 用较小的LM(如T5和Llama2)可实现与专家编写的提示链或专有LM系统相媲美的性能。
核心组件
TextGrad 遵循 PyTorch 语法,好处就是灵活且易于使用。
PyTorch 框架把智能系统视为计算图,其中变量是复杂函数调用的输入和输出。这个框架下,LLM 提供丰富、通用的自然语言提示,优化这些计算图变量,从代码片段到分子结构。TextGrad “反向传播” LLM 提供的文本反馈,从而改进各个组件。
TextGrad 关键组件包括:
-
变量:计算图节点,包含非结构化数据(如文本)。
-
函数:变量转换(如LLM调用、模拟器)。
-
梯度:LLM 自然语言反馈,描述如何修改变量。
-
文本梯度下降(TGD):一种优化器,基于当前值和文本梯度更新变量。
TextGrad 使用链式法则构建计算图,类似于传统的自动微分算法,用语言模型提供的自然语言反馈代替了数值梯度,这是最核心的贡献。
在“梯度下降”优化过程中,“梯度”生成、“校正量”添加、以及损失函数的计算,都变成了对语言模型的调用。
这种新颖的方法利用语言模型提供表达能力来指导优化过程。
应用示例
TextGrad 在不同的场景应用下都具备通用性,比如:
- TextGrad 通过优化代码片段,提高处理难题的能力。
- TextGrad 通过优化问题解决方案,提高了LLM解决复杂任务(如科学问题)能力。
- TextGrad 通过优化提示,提高大模型推理能力,并能提升性能。
- TextGrad 通过优化化学结构设计,体现其在药物研发中潜力。
- TextGrad 通过优化治疗计划,改进患者的治疗效果。
TextGrad 关键应用是解决方案优化,目标是改进复杂问题的解决方案,优化过程涉及一个计算图,其中解决方案是要优化的参数,损失函数是通过使用 LLM 评估得到的。
在每次迭代中,LLM 会提示当前问题、当前解决方案和需要评估或调查当前迭代的测试指令,这个过程会在优化过程中逐渐改进解决方案。
举例,以下是使用 TextGrad 进行解决方案优化的示例实现:
!pip install TextGrad # 可能需要在安装TextGrad后重新启动笔记本
import argparse
import concurrent
from dotenv import load_dotenv
from tqdm import tqdm
import TextGrad as tg
from TextGrad.tasks import load_task
import numpy as np
import random
load_dotenv(override=True)
# 如果使用Colab
from google.colab import userdata
OPENAI_API_KEY = userdata.get('OPENAI_API_KEY')
os.environ['OPENAI_API_KEY'] = OPENAI_API_KEY
def set_seed(seed):
np.random.seed(seed)
random.seed(seed)
def eval_sample(item, eval_fn, model):
"""
该函数允许我们评估答案是否是对提示问题的好答案。
"""
x, y = item
x = tg.Variable(x, requires_grad=False, role_description="查询语言模型")
y = tg.Variable(y, requires_grad=False, role_description="查询的正确答案")
response = model(x)
try:
eval_output_variable = eval_fn(inputs=dict(prediction=response, ground_truth_answer=y))
return int(eval_output_variable.value)
except:
eval_output_variable = eval_fn([x, y, response])
eval_output_parsed = eval_fn.parse_output(eval_output_variable)
return int(eval_output_parsed)
def eval_dataset(test_set, eval_fn, model, max_samples: int=None):
if max_samples is None:
max_samples = len(test_set)
accuracy_list = []
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
futures = []
for _, sample in enumerate(test_set):
future = executor.submit(eval_sample, sample, eval_fn, model)
futures.append(future)
if len(futures) >= max_samples:
break
tqdm_loader = tqdm(concurrent.futures.as_completed(futures), total=len(futures), position=0)
for future in tqdm_loader:
acc_item = future.result()
accuracy_list.append(acc_item)
tqdm_loader.set_description(f"准确率: {np.mean(accuracy_list)}")
return accuracy_list
def run_validation_revert(system_prompt: tg.Variable, results, model, eval_fn, val_set):
val_performance = np.mean(eval_dataset(val_set, eval_fn, model))
previous_performance = np.mean(results["validation_acc"][-1])
print("验证性能: ", val_performance)
print("之前的性能: ", previous_performance)
previous_prompt = results["prompt"][-1]
if val_performance < previous_performance:
print(f"拒绝的提示: {system_prompt.value}")
system_prompt.set_value(previous_prompt)
val_performance = previous_performance
results["validation_acc"].append(val_performance)
set_seed(12)
llm_api_eval = tg.get_engine(engine_name="gpt-4o")
llm_api_test = tg.get_engine(engine_name="gpt-3.5-turbo-0125")
tg.set_backward_engine(llm_api_eval, override=True)
# 加载数据和评估函数
train_set, val_set, test_set, eval_fn = load_task("BBH_object_counting", evaluation_api=llm_api_eval)
print("训练/验证/测试集长度: ", len(train_set), len(val_set), len(test_set))
STARTING_SYSTEM_PROMPT = train_set.get_task_description()
train_loader = tg.tasks.DataLoader(train_set, batch_size=3, shuffle=True)
# 测试评估引擎的0样本性能
system_prompt = tg.Variable(STARTING_SYSTEM_PROMPT,
requires_grad=True,
role_description="系统提示语言模型")
model_evaluation = tg.BlackboxLLM(llm_api_eval, system_prompt)
system_prompt = tg.Variable(STARTING_SYSTEM_PROMPT,
requires_grad=True,
role_description="结构化系统提示,针对QA任务指定行为和策略的有一定能力的语言模型")
model = tg.BlackboxLLM(llm_api_test, system_prompt)
# 文本梯度下降优化器初始化评估模型的API和要优化的系统提示参数。
optimizer = tg.TextualGradientDescent(engine=llm_api_eval, parameters=[system_prompt])
results = {"test_acc": [], "prompt": [], "validation_acc": []}
results["test_acc"].append(eval_dataset(test_set, eval_fn, model))
results["validation_acc"].append(eval_dataset(val_set, eval_fn, model))
results["prompt"].append(system_prompt.get_value())
for epoch in range(3):
for steps, (batch_x, batch_y) in enumerate((pbar := tqdm(train_loader, position=0))):
pbar.set_description(f"训练步骤 {steps}. 轮次 {epoch}")
optimizer.zero_grad()
losses = []
for (x, y) in zip(batch_x, batch_y):
x = tg.Variable(x, requires_grad=False, role_description="查询语言模型")
y = tg.Variable(y, requires_grad=False, role_description="查询的正确答案")
response = model(x)
try:
eval_output_variable = eval_fn(inputs=dict(prediction=response, ground_truth_answer=y))
except:
eval_output_variable = eval_fn([x, y, response])
losses.append(eval_output_variable)
total_loss = tg.sum(losses)
total_loss.backward()
optimizer.step()
run_validation_revert(system_prompt, results, model, eval_fn, val_set)
print("系统提示: ", system_prompt)
test_acc = eval_dataset(test_set, eval_fn, model)
results["test_acc"].append(test_acc)
results["prompt"].append(system_prompt.get_value())
if steps == 3:
break
上述代码演示了如何使用 TextGrad 库对语言模型的系统提示进行基于梯度的优化,从而提高在特定问答任务上的性能。
TextGrad 库简化了将梯度应用于提示并评估模型性能的过程。
据研究表明,通过在测试时使用 TextGrad 进行自我优化,可明显提升大模型的问答能力。
小结
TextGrad 是一种通过文本反馈反向传播优化大模型系统的新范式,DSPy 则专注于通过参数化模块和编译器优化实现LM管道的系统化开发和优化。
随着开发人员继续探索 TextGrad 和其他创新框架的能力,我们可以期待在人工智能系统优化方面看到更多突破性的进展~~
本篇通译自:medium.com/@jelkhoury8…
转载自:https://juejin.cn/post/7383017171179618358