"""FinSight 微调 before/after 评估 harness。 对同一份分层抽样的 held-out eval 集,用**完全相同**的 prompt 和解码参数, 分别跑 base 模型与微调模型,按 Agent 用任务匹配的指标打分。 三层指标(为何这么分见 README / 对话说明): 1) 客观标签指标 —— sentiment 三分类准确率 + macro-F1;topic/FINNL 类别准确率 2) ROUGE-L(字级)—— summary / event / topic-FINRE / qa 的 n-gram 重合(含格式) 3) LLM-as-judge —— qa 开放问答的成对胜率(裁判换一个模型 + 位置互换去偏) 用法(用 .venv,只需 requests+jieba,不依赖 torch): .venv/bin/python training/evaluate.py --n 120 \ --base qwen2.5:7b-instruct --ft finsight-qwen --judge qwen3:8b """ import argparse, json, random, re, time, pathlib, collections import urllib.request ROOT = pathlib.Path(__file__).resolve().parent.parent EVAL = ROOT / "data" / "processed" / "finsight_eval.jsonl" OLLAMA = "http://localhost:11434" SENTI = ["积极", "消极", "中性"] FINNL_CATS = ["公司", "行业", "大盘", "中国", "国际", "经济", "政策", "期货", "债券", "房地产", "外汇", "虚拟货币", "新冠", "能源", "政治"] # ---------------- Ollama ---------------- def ollama_chat(model, user, num_predict=512, think=None): body = { "model": model, "messages": [{"role": "user", "content": user}], "stream": False, "options": {"temperature": 0, "num_predict": num_predict, "seed": 42}, } if think is not None: body["think"] = think req = urllib.request.Request( f"{OLLAMA}/api/chat", data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}, ) for _ in range(3): try: r = json.loads(urllib.request.urlopen(req, timeout=300).read()) return r["message"]["content"].strip() except Exception as e: last = e time.sleep(2) raise last def prompt_of(d): """对齐训练时的 alpaca→qwen 拼接:instruction + 换行 + input。""" return d["instruction"] + (("\n" + d["input"]) if d.get("input") else "") # ---------------- 指标 ---------------- def extract_senti(text): # 取文本中最后出现的情感词(模型常先解释再给结论) hits = [(text.rfind(w), w) for w in SENTI if w in text] return max(hits)[1] if hits else None def extract_finnl(text): hits = [(text.rfind(c), c) for c in FINNL_CATS if c in text] return max(hits)[1] if hits else None def lcs(a, b): m, n = len(a), len(b) dp = [0] * (n + 1) for i in range(1, m + 1): prev = 0 for j in range(1, n + 1): tmp = dp[j] dp[j] = prev + 1 if a[i - 1] == b[j - 1] else max(dp[j], dp[j - 1]) prev = tmp return dp[n] def rougeL(pred, gold): """字级 ROUGE-L F1(中文标准做法,去标点空白)。""" clean = lambda s: re.sub(r"[\s，。、；：！？,.　]", "", s) p, g = clean(pred), clean(gold) if not p or not g: return 0.0 l = lcs(p, g) prec, rec = l / len(p), l / len(g) return 0.0 if prec + rec == 0 else 2 * prec * rec / (prec + rec) def macro_f1(pairs, labels): """pairs: [(gold, pred)]; 对每个标签算 F1 再平均。""" f1s = [] for lab in labels: tp = sum(1 for g, p in pairs if g == lab and p == lab) fp = sum(1 for g, p in pairs if g != lab and p == lab) fn = sum(1 for g, p in pairs if g == lab and p != lab) prec = tp / (tp + fp) if tp + fp else 0 rec = tp / (tp + fn) if tp + fn else 0 f1s.append(2 * prec * rec / (prec + rec) if prec + rec else 0) return sum(f1s) / len(f1s) def llm_judge(judge, question, ans_a, ans_b): """成对评判,返回 'A'/'B'/'tie'。调用方负责位置互换去偏。""" sys = ( "你是金融问答评审。根据【问题】判断哪个回答更准确、专业、完整。" "只输出一个字母:A 表示回答A更好,B 表示回答B更好,T 表示二者相当。不要解释。" ) user = f"【问题】{question}\n\n【回答A】{ans_a}\n\n【回答B】{ans_b}\n\n更好的是(A/B/T):" out = ollama_chat(judge, sys + "\n\n" + user, num_predict=4, think=False).upper() if "A" in out and "B" not in out: return "A" if "B" in out and "A" not in out: return "B" return "tie" # ---------------- 抽样 ---------------- def sample(n, seed=42): by = collections.defaultdict(list) for line in open(EVAL): d = json.loads(line) by[d["agent"]].append(d) rng = random.Random(seed) out = [] for agent, rows in by.items(): rng.shuffle(rows) out.extend(rows[:n]) return out # ---------------- 主流程 ---------------- def generate(model, data): preds = [] for i, d in enumerate(data): np_ = 24 if d["agent"] in ("sentiment", "topic") else 512 preds.append(ollama_chat(model, prompt_of(d), num_predict=np_)) if (i + 1) % 50 == 0: print(f" {model}: {i+1}/{len(data)}") return preds def score(data, preds): """返回 per-agent 指标 dict。""" by = collections.defaultdict(list) for d, p in zip(data, preds): by[d["agent"]].append((d, p)) res = {} for agent, rows in by.items(): if agent == "sentiment": pairs = [(d["output"], extract_senti(p)) for d, p in rows] acc = sum(1 for g, p in pairs if g == p) / len(pairs) res[agent] = {"n": len(pairs), "accuracy": round(acc, 4), "macro_f1": round(macro_f1(pairs, SENTI), 4), "parse_fail": sum(1 for _, p in pairs if p is None)} else: r = sum(rougeL(p, d["output"]) for d, p in rows) / len(rows) entry = {"n": len(rows), "rougeL": round(r, 4)} if agent == "topic": finnl = [(d, p) for d, p in rows if d["task"] == "FINNL"] if finnl: acc = sum(1 for d, p in finnl if extract_finnl(d["output"]) == extract_finnl(p)) / len(finnl) entry["finnl_acc"] = round(acc, 4) entry["finnl_n"] = len(finnl) res[agent] = entry return res def run_judge(judge, data, preds_base, preds_ft): """qa 成对胜率,位置互换两轮。返回 {ft_win, base_win, tie, win_rate}。""" qa = [(d, b, f) for d, b, f in zip(data, preds_base, preds_ft) if d["agent"] == "qa"] ft_win = base_win = tie = 0 for d, b, f in qa: q = prompt_of(d) v1 = llm_judge(judge, q, f, b) # A=ft, B=base v2 = llm_judge(judge, q, b, f) # A=base, B=ft (互换) # 统计:把两轮折算到 ft 视角 for v, ft_is_a in ((v1, True), (v2, False)): if v == "tie": tie += 1 elif (v == "A") == ft_is_a: ft_win += 1 else: base_win += 1 total = ft_win + base_win + tie return {"qa_n": len(qa), "comparisons": total, "ft_win": ft_win, "base_win": base_win, "tie": tie, "ft_win_rate": round(ft_win / total, 4) if total else 0} def main(): ap = argparse.ArgumentParser() ap.add_argument("--n", type=int, default=120, help="每个 agent 抽样数") ap.add_argument("--base", default="qwen2.5:7b-instruct") ap.add_argument("--ft", default="finsight-qwen") ap.add_argument("--judge", default="qwen3:8b") ap.add_argument("--out", default=str(ROOT / "training" / "eval_report.json")) args = ap.parse_args() data = sample(args.n) print(f"抽样 {len(data)} 条 " f"({dict(collections.Counter(d['agent'] for d in data))})") print(f"\n[1/2] 生成 base 预测: {args.base}") pb = generate(args.base, data) print(f"[2/2] 生成 微调 预测: {args.ft}") pf = generate(args.ft, data) sb, sf = score(data, pb), score(data, pf) print(f"\n[judge] qa 成对评判: {args.judge}") judge = run_judge(args.judge, data, pb, pf) report = {"n_per_agent": args.n, "base": args.base, "ft": args.ft, "judge": args.judge, "scores": {"base": sb, "ft": sf}, "qa_judge": judge} pathlib.Path(args.out).write_text(json.dumps(report, ensure_ascii=False, indent=2)) # 打印对比表 print("\n" + "=" * 64) print(f"{'agent':<10}{'metric':<14}{'base':>10}{'finetune':>12}{'Δ':>10}") print("-" * 64) for agent in sb: for m in sb[agent]: if m in ("n", "parse_fail", "finnl_n"): continue bv, fv = sb[agent].get(m, 0), sf[agent].get(m, 0) print(f"{agent:<10}{m:<14}{bv:>10.4f}{fv:>12.4f}{fv-bv:>+10.4f}") print("-" * 64) j = judge print(f"qa LLM-judge ft胜率 {j['ft_win_rate']:.1%} " f"(ft赢{j['ft_win']} / base赢{j['base_win']} / 平{j['tie']}, 共{j['comparisons']}次比较)") print("=" * 64) print(f"\n报告已存:{args.out}") if __name__ == "__main__": main()