#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ text_gradient_optimizer.py — A3 文本梯度自动提示词优化（吸收 openJiuwen Textual-Gradient） ======================================================================================== 采集 bad case → 计算文本梯度（期望 vs 实际差异）→ 生成 prompt 优化候选 → 入评审队列。 优化结果**仅生成候选版本**，经 T0 人工审批锁档后才生效，杜绝反馈噪声与过度优化。 安全设计（规避 openJiuwen 三大坑）： - 防噪声：用户反馈不直接改 prompt，只作为 bad case 样本进入梯度分析 - 防过拟合：候选进入人工评审队列（pending_t0_review），审批通过才 sediment 锁档 - 可解释：每次演进记录 bad case → 梯度 → 改动 diff → 理由，全程可追溯 """ import json import os import uuid from datetime import datetime ANCHOR = "Ω₀⊂⊙∞⊂Ω" DID = "DID-BR-000002" # 评审队列（候选真值暂存处，待 T0 人工审批） REVIEW_DB = os.environ.get("TG_REVIEW_DB", "/var/lib/zr-tg/review_queue.json") # 已锁档 prompt 版本链 PROMPT_DB = os.environ.get("TG_PROMPT_DB", "/var/lib/zr-tg/prompt_versions.json") # 尝试对接 omega_kernel 的沉淀接口（审批通过后锁档用） try: from omega_kernel import sediment except Exception: sediment = None def now(): return datetime.now().isoformat(timespec="seconds") def _load_json(path, default): try: if os.path.exists(path): with open(path, "r", encoding="utf-8") as f: return json.load(f) except Exception: pass return default def _save_json(path, obj): os.makedirs(os.path.dirname(path) or ".", exist_ok=True) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(obj, f, ensure_ascii=False, indent=2) os.replace(tmp, path) # ---------------------------------------------------------------- 梯度计算 def compute_textual_gradient(badcase): """ 基于期望输出与实际输出差异，计算"文本梯度"（自然语言描述的有向改进信号）。 返回结构化梯度对象。 """ expected = str(badcase.get("expected", "")) actual = str(badcase.get("actual", "")) instruction = str(badcase.get("instruction", "")) if not actual: # 无实际输出（执行失败/超时）：梯度指向"缺失或失败" gradient = { "type": "missing_or_failed", "signal": "实际输出缺失，疑似执行失败或超时", "direction": "补充执行兜底、明确失败原因与重试策略", "examples": [], } elif expected and expected not in actual: # 期望存在但输出不符：定位差异片段 expected_tokens = expected.split("。") missed = [e for e in expected_tokens if e and e not in actual][:3] gradient = { "type": "content_mismatch", "signal": "输出缺失关键要点", "direction": "在 prompt 中显式要求覆盖要点：" + "；".join(missed), "examples": [{"expected": expected[:200], "actual": actual[:200]}], } elif not expected: gradient = { "type": "no_expectation", "signal": "缺少验收标准，无法精确判定", "direction": "补充明确的输出格式与验收约束", "examples": [], } else: # 基本吻合，仅微调 gradient = { "type": "minor_refine", "signal": "输出基本达标，可做细节优化", "direction": "保持现状，仅在格式/风格上微调", "examples": [], } gradient["instruction_signal"] = f"当前指令: {instruction[:80] or '（无）'}" return gradient # ---------------------------------------------------------------- 候选生成 def propose_prompt_update(badcase, current_prompt): """ 基于文本梯度生成 prompt 优化候选（指令 + 示例联合迭代）。 只返回候选，不落生效。 """ gradient = compute_textual_gradient(badcase) update_type = gradient["type"] # 指令层补丁 patch = None if update_type == "content_mismatch": patch = ("【输出要求】必须覆盖以下要点：\n" + "\n".join(f"- {p}" for p in gradient["examples"][0]["expected"].split("。") if p) if gradient["examples"] else "【输出要求】请确保回答完整覆盖任务全部要点。") elif update_type == "missing_or_failed": patch = "【执行要求】若一次执行未产出结果，请明确报告失败原因并给出兜底方案，禁止空回复。" elif update_type == "no_expectation": patch = "【输出要求】请以结构化列表输出结果，并附简短结论与依据。" else: patch = "【风格】请保持回答简洁、结构清晰、要点先行。" candidate = { "candidate_id": f"TG-{uuid.uuid4().hex[:10].upper()}", "current_prompt": current_prompt, "proposed_patch": patch, "gradient": gradient, "created": now(), "status": "pending_t0_review", # 关键：不自动生效，等待人工审批 "did": DID, "anchor": ANCHOR, } return candidate # ---------------------------------------------------------------- 评审入队 def enqueue_review(candidate, reason=""): """把 prompt 优化候选写入评审队列（T0 人工审批前不生效）。""" q = _load_json(REVIEW_DB, {"queue": []}) q.setdefault("queue", []) candidate["enqueue_reason"] = reason or "文本梯度自动优化候选" candidate["enqueued"] = now() q["queue"].append(candidate) _save_json(REVIEW_DB, q) return {"ok": True, "candidate_id": candidate["candidate_id"], "queue_size": len(q["queue"])} # ---------------------------------------------------------------- 审批（T0） def approve(candidate_id, reviewer="human"): """ T0 人工审批：通过才锁档生效（sediment 沉淀真值 + 记入版本链）。 """ q = _load_json(REVIEW_DB, {"queue": []}) queue = q["queue"] target = next((c for c in queue if c.get("candidate_id") == candidate_id), None) if not target: return {"ok": False, "error": "candidate_not_found"} queue = [c for c in queue if c.get("candidate_id") != candidate_id] q["queue"] = queue _save_json(REVIEW_DB, q) target["status"] = "approved" target["reviewer"] = reviewer target["approved_at"] = now() # 记入版本链（保留旧版本，可回滚） pd = _load_json(PROMPT_DB, {"versions": []}) pd.setdefault("versions", []) pd["versions"].append(target) _save_json(PROMPT_DB, pd) # 审批通过才锁档沉淀到真值（防未审批污染） if sediment is not None: try: sediment(f"prompt-gradient-{target['candidate_id']}", json.dumps(target, ensure_ascii=False)) except Exception: pass return {"ok": True, "candidate_id": candidate_id, "status": "approved", "versions": len(pd["versions"])} def reject(candidate_id, reviewer="human"): q = _load_json(REVIEW_DB, {"queue": []}) queue = q["queue"] target = next((c for c in queue if c.get("candidate_id") == candidate_id), None) if not target: return {"ok": False, "error": "candidate_not_found"} queue = [c for c in queue if c.get("candidate_id") != candidate_id] q["queue"] = queue _save_json(REVIEW_DB, q) target["status"] = "rejected" target["reviewer"] = reviewer target["rejected_at"] = now() return {"ok": True, "candidate_id": candidate_id, "status": "rejected"} # ---------------------------------------------------------------- CLI def _demo(): print("== A3 文本梯度优化 demo ==") badcase = { "instruction": "分析该产品竞品", "expected": "第一：市场概况；第二：竞品定位；第三：优劣势对比；第四：结论建议。", "actual": "这款产品很不错。", } current = "你是一个分析助手，请分析产品竞品。" cand = propose_prompt_update(badcase, current) print("梯度类型:", cand["gradient"]["type"]) print("优化候选 prompt 补丁:", cand["proposed_patch"]) r = enqueue_review(cand, reason="demo") print("入队:", r) ap = approve(cand["candidate_id"], reviewer="demo-human") print("审批:", ap) return cand if __name__ == "__main__": import sys if len(sys.argv) > 1 and sys.argv[1] == "queue": print(json.dumps(_load_json(REVIEW_DB, {"queue": []}), ensure_ascii=False, indent=2)) else: _demo()