查看代码
# 最小反思式优化回路:轨迹 → 评分 → 反思规则 → 新候选 → 留档
TRACE, CAND, K, MIN_GAIN = [], [], 3, 0.05
def metric(output, expect): # 前提 1:可自动计算的评分函数(没有它就没有改进方向)
return 1.0 if expect.strip() in output else 0.0
def run(prompt, batch, call_lm): # call_lm(prompt, x) 返回 (output, steps)
rows = []
for x in batch:
output, steps = call_lm(prompt, x['input'])
rows.append({'input': x['input'], 'output': output, 'steps': steps, # 前提 2:完整轨迹
'score': metric(output, x['expect'])})
TRACE.extend(rows) # 轨迹必须先落盘,再谈反思
return rows
def reflect(rows, call_reflect): # 反思只产出「改进规则」,不直接重写答案
bad = [r for r in rows if r['score'] == 0]
return call_reflect(bad) # 输入里要含中间步骤,否则模型只能对着结果猜原因
def step(prompt, batch, call_lm, call_reflect):
rows = run(prompt, batch, call_lm) # 基线:同一评测窗口上的得分
rule = reflect(rows, call_reflect) # 读失败轨迹 → 改进规则
cand = call_lm.rewrite(prompt, rule) # 只按规则改,其余原文不动
base = sum(r['score'] for r in rows) / len(batch)
scores = [sum(r['score'] for r in run(cand, batch, call_lm)) / len(batch) for _ in range(K)]
mean = sum(scores) / K
CAND.append({'rule': rule, 'base': base, 'mean': mean,
'spread': max(scores) - min(scores),
'adopted': mean - base >= MIN_GAIN})
return cand if CAND[-1]['adopted'] else prompt # 未过阈值 ⇒ 保留原提示词,候选只留档
预期观察:每次迭代都留下「基线分 / 候选分 / 波动幅度 / 是否采纳」四列;只有均值增益跨过阈值且波动没有恶化才采纳候选,否则保留原提示词进入下一轮。去掉重复次数 K 与 spread 就会把一次幸运运行当成真实增益(把 call_lm / call_reflect 换成你所用的框架调用即可,例如 DSPy 程序加 LM)。