# -*- coding: utf-8 -*- """ V2 自动评估体系 - eval 数据集：用例（input/expected）+ 评分标准（rubric），随任务运行自动沉淀 - 评估运行：Worker 逐条作答 → LLM-as-Judge 按 rubric 打分 → 平均分/排行榜 - 一键沉淀：把已验收通过的任务产出转为 eval 用例 """ import json import time import threading import traceback import db import llm_gateway import notify JUDGE_PROMPT = ( '你是严格、公正的评估裁判（LLM-as-Judge）。请对执行结果按评分标准打分。\n' '输出严格 JSON：{{"score": 0到100的整数, "judgment": "评分理由（50字内）"}}\n' '只输出 JSON。\n\n' '评分标准：{rubric}\n' '测试输入：{input}\n' '参考答案（预期）：{expected}\n' '执行结果：{output}' ) DEFAULT_RUBRIC = ( '内容准确性 40%（与参考答案一致性、事实正确性），' '完整性 30%（是否覆盖输入要求的所有要点），' '清晰度与格式 30%（结构清晰、语言通顺）。' ) BUILTIN_DATASETS = [ { 'name': '通用指令遵循', 'description': '基础指令遵循与回答质量基准集（内置）', 'rubric': '指令遵循度 50%（是否按要求格式/要点作答），准确性 30%，简洁性 20%。', 'tags': ['基础', '指令遵循'], 'cases': [ {'input': '用 3 句话向非技术人员解释什么是"大语言模型（LLM）"。', 'expected': '包含：基于海量文本训练、预测下一个词/生成文本、可对话问答等要点'}, {'input': '列出数据库事务的 ACID 四个特性，并各用一句话解释。', 'expected': '原子性、一致性、隔离性、持久性，且各有解释'}, {'input': '写一段 50 字以内的产品文案，推广一款 AI 项目管理工具。', 'expected': '≤50字、突出AI提效、有行动号召'}, ], }, { 'name': '代码生成', 'description': '代码生成正确性基准集（内置）', 'rubric': '功能正确性 60%，代码质量 25%（可读性/健壮性），格式规范 15%。', 'tags': ['代码'], 'cases': [ {'input': '用 Python 写一个函数 fibonacci(n)，返回斐波那契数列前 n 项。', 'expected': '正确实现、处理 n<=0、含示例'}, {'input': '写一个 SQL 查询：统计每个部门的员工人数，按人数降序。', 'expected': 'GROUP BY department, ORDER BY count DESC'}, {'input': '用 Python 写一个装饰器 @retry(times=3)，失败自动重试。', 'expected': '捕获异常重试、最多 times 次、可传参'}, ], }, ] def _extract_judge(text): """解析裁判 JSON，容错兜底""" try: t = text.strip() if t.startswith('```'): t = t.strip('`') if t.startswith('json'): t = t[4:] t = t.strip() start = t.find('{') end = t.rfind('}') + 1 data = json.loads(t[start:end]) if end > start else {} score = float(data.get('score', data.get('总分', 0))) judgment = data.get('judgment') or data.get('评分理由') or '' return max(0.0, min(100.0, score)), str(judgment)[:500] except Exception: import re m = re.search(r'(\d{1,3})\s*[/／]\s*100', text) if m: return float(m.group(1)), text[:300] m = re.search(r'"score"\s*:\s*(\d{1,3})', text) if m: return float(m.group(1)), text[:300] return 0.0, '裁判输出解析失败' def create_builtin_datasets(): """首次启动注入内置数据集""" c = db.q('SELECT COUNT(*) c FROM eval_datasets WHERE is_builtin=1')[0]['c'] if c > 0: return for ds in BUILTIN_DATASETS: did = db.w( 'INSERT INTO eval_datasets (name, description, rubric, tags, is_builtin, created_at, updated_at) ' 'VALUES (?,?,?,?,1,?,?)', (ds['name'], ds['description'], ds['rubric'], json.dumps(ds['tags']), db.now(), db.now())) for case in ds['cases']: db.w('INSERT INTO eval_cases (dataset_id, input, expected, tags, created_at) VALUES (?,?,?,?,?)', (did, case['input'], case['expected'], json.dumps(ds['tags']), db.now())) def sink_done_tasks(): """沉淀：把最近验收通过（done 且无打回）且尚未沉淀的任务转为 eval 用例（幂等）""" # 幂等标记：记录在 settings sunk = set(json.loads(db.get_setting('eval_sunk_task_ids', '[]'))) rows = db.q( 'SELECT id, title, description, output_text, updated_at FROM tasks ' 'WHERE status="done" AND rejection_count=0 AND output_text!="" ORDER BY id DESC LIMIT 200') created = 0 for t in rows: if t['id'] in sunk: continue # 找到/创建"任务沉淀"数据集 ds = db.q('SELECT id FROM eval_datasets WHERE name=?', ('任务产出沉淀',), one=True) if not ds: ds_id = db.w( 'INSERT INTO eval_datasets (name, description, rubric, tags, is_builtin, created_at, updated_at) ' 'VALUES (?,?,?,?,0,?,?)', ('任务产出沉淀', '从已验收任务自动沉淀的高质量问答对', DEFAULT_RUBRIC, json.dumps(['沉淀']), db.now(), db.now())) else: ds_id = ds['id'] # 避免重复用例 dup = db.q('SELECT id FROM eval_cases WHERE dataset_id=? AND input=?', (ds_id, t['description'] or t['title']), one=True) if dup: sunk.add(t['id']) continue db.w('INSERT INTO eval_cases (dataset_id, input, expected, tags, created_at) VALUES (?,?,?,?,?)', (ds_id, (t['description'] or t['title'])[:2000], t['output_text'][:4000], json.dumps(['沉淀', f'task#{t["id"]}']), db.now())) sunk.add(t['id']) created += 1 db.set_setting('eval_sunk_task_ids', json.dumps(list(sunk)[-2000:])) return created def evaluate_case(dataset, case, worker): """单条用例：Worker 作答 + Judge 打分。返回 (output, score, judgment, latency, cost, tokens)""" t0 = time.time() # 1) Worker 作答 r = llm_gateway.chat( worker['provider'], worker['model'], [{'role': 'system', 'content': worker['system_prompt'] or '你是待评估的执行 Agent，请直接回答问题。'}, {'role': 'user', 'content': case['input']}], temperature=worker['temperature'], max_tokens=worker['max_tokens'] or 2000, base_url=worker['base_url'] or None, api_key=worker['api_key'] or None) output = r['text'] latency = int((time.time() - t0) * 1000) tokens = r['total_tokens'] # 2) Judge 打分 judge_cost = 0.0 try: j = llm_gateway.chat( worker['provider'], worker['model'], [{'role': 'system', 'content': '你只输出 JSON。'}, {'role': 'user', 'content': JUDGE_PROMPT.format( rubric=dataset['rubric'] or DEFAULT_RUBRIC, input=case['input'][:2000], expected=(case['expected'] or '无参考答案，凭专业判断')[:3000], output=output[:4000])}], temperature=0.1, max_tokens=600, base_url=worker['base_url'] or None, api_key=worker['api_key'] or None) score, judgment = _extract_judge(j['text']) judge_cost = j['cost'] tokens += j['total_tokens'] except Exception as e: score, judgment = 0.0, f'裁判调用失败: {e}' total_cost = round(r['cost'] + judge_cost, 6) return output, score, judgment, latency, total_cost, tokens def run_eval(run_id): """后台线程：执行一次完整评估""" run = db.q('SELECT * FROM eval_runs WHERE id=?', (run_id,), one=True) if not run: return ds = db.q('SELECT * FROM eval_datasets WHERE id=?', (run['dataset_id'],), one=True) worker = db.q('SELECT * FROM workers WHERE id=?', (run['worker_id'],), one=True) if not ds or not worker: db.w('UPDATE eval_runs SET status="failed", finished_at=? WHERE id=?', (db.now(), run_id)) return cases = db.q('SELECT * FROM eval_cases WHERE dataset_id=? ORDER BY id', (ds['id'],)) total_tokens, total_cost, scores, done = 0, 0.0, [], 0 try: for case in cases: try: output, score, judgment, latency, cost, tokens = evaluate_case(ds, case, worker) db.w( 'INSERT INTO eval_results (run_id, case_id, worker_id, output, score, judgment, ' 'latency_ms, cost, created_at) VALUES (?,?,?,?,?,?,?,?,?)', (run_id, case['id'], worker['id'], output, score, judgment, latency, cost, db.now())) total_cost += cost total_tokens += tokens scores.append(score) done += 1 db.w('UPDATE eval_runs SET cases_done=? WHERE id=?', (done, run_id)) except Exception as e: db.w( 'INSERT INTO eval_results (run_id, case_id, worker_id, output, score, judgment, ' 'latency_ms, cost, created_at) VALUES (?,?,?,?,?,?,?,?,?)', (run_id, case['id'], worker['id'], '', 0, f'执行失败: {str(e)[:200]}', 0, 0, db.now())) done += 1 db.w('UPDATE eval_runs SET cases_done=? WHERE id=?', (done, run_id)) avg = round(sum(scores) / len(scores), 2) if scores else 0 db.w('UPDATE eval_runs SET status="done", score=?, total_tokens=?, cost=?, cases_done=?, finished_at=? ' 'WHERE id=?', (avg, total_tokens, round(total_cost, 6), done, db.now(), run_id)) try: notify.notify('task_done', f'评估完成：{ds["name"]}', f'Worker「{worker["name"]}」在数据集「{ds["name"]}」上平均分 {avg}/100，' f'共 {done}/{len(cases)} 条用例，成本 ¥{total_cost:.4f}', save_alert=True, level='info', atype='eval_done') except Exception: pass except Exception as e: db.w('UPDATE eval_runs SET status="failed", error=?, finished_at=? WHERE id=?', (str(e)[:300], db.now(), run_id)) _trace(e) def _trace(e): print('[eval]', traceback.format_exc()) class EvalRunner: def __init__(self): self._threads = {} def submit(self, run_id): if run_id in self._threads and self._threads[run_id].is_alive(): return False t = threading.Thread(target=run_eval, args=(run_id,), daemon=True) self._threads[run_id] = t t.start() return True runner = EvalRunner()