#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ harness_rsi.py — A6 RSI 双维度自修改（吸收 openJiuwen RSI：Harness 方法层 / Artifacts 成果层分离） ======================================================================================== 把"如何做"（Harness 方法层：Prompt / Skill / Tool / Rail）与"做出什么"（Artifacts 成果层） 分离为两类可独立迭代的对象： - Harness ：方法层，可被自修改，但修改须经 T0 人工审批锁档 - Artifacts ：成果层，每次运行产出，带哈希可溯源 RSIEngine 执行 Harness → 产出 Artifacts → 基于结果 propose Harness 修改（待审批） """ import json import os import uuid import hashlib from datetime import datetime ANCHOR = "Ω₀⊂⊙∞⊂Ω" DID = "DID-BR-000002" RSI_DB = os.environ.get("RSI_DB", "/var/lib/zr-rsi/rsi.json") 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 _sha(s): return hashlib.sha256(str(s).encode()).hexdigest()[:16] class Harness: """方法层对象：定义如何完成一类任务（可自修改、可版本化）。""" def __init__(self, name, prompt="", skills=None, tools=None, rails=None): self.name = name self.prompt = prompt self.skills = skills or [] self.tools = tools or [] self.rails = rails or [] # 约束/护栏 self.version = 1 self.created = now() def to_dict(self): d = {"name": self.name, "prompt": self.prompt, "skills": self.skills, "tools": self.tools, "rails": self.rails, "version": self.version, "created": self.created, "anchor": ANCHOR} d["hash"] = _sha(json.dumps({k: d[k] for k in ("prompt", "skills", "tools", "rails")}, ensure_ascii=False)) return d class Artifact: """成果层对象：一次运行的产出，与 Harness 分离，带哈希。""" def __init__(self, kind, content, meta=None): self.kind = kind self.content = content self.meta = meta or {} self.id = f"ART-{uuid.uuid4().hex[:8].upper()}" self.created = now() self.hash = _sha(f"{kind}|{content}") def to_dict(self): return {"id": self.id, "kind": self.kind, "content": self.content, "meta": self.meta, "hash": self.hash, "created": self.created, "anchor": ANCHOR} class RSIEngine: """RSI 引擎：Harness 执行 → 产出 Artifacts → 提议 Harness 自修改（走审批门）。""" def __init__(self): self.db = _load_json(RSI_DB, {"harness": {}, "artifacts": [], "pending_changes": [], "approved_changes": []}) def save_harness(self, harness: Harness): self.db["harness"][harness.name] = harness.to_dict() _save_json(RSI_DB, self.db) return harness.to_dict() def get_harness(self, name): return self.db["harness"].get(name) # ---- 执行：Harness → Artifact ---- def run(self, harness_name, input_text, executor=None): h = self.get_harness(harness_name) if not h: return {"ok": False, "error": "harness_not_found"} # 默认执行器：组装 prompt 调用本地微模型；可用外部 executor 覆盖 output = None if executor: try: output = executor(h, input_text) except Exception: output = None if output is None: try: from micro_brain import think output = think(f"{h['prompt']}\n输入：{input_text}") except Exception: output = {"status": "queued", "harness": harness_name, "input": str(input_text)[:80]} art = Artifact(kind="result", content=output, meta={"harness": harness_name, "version": h["version"], "input_preview": str(input_text)[:80]}) self.db["artifacts"].append(art.to_dict()) _save_json(RSI_DB, self.db) return {"ok": True, "artifact": art.to_dict()} # ---- 自修改：propose Harness 变更（待 T0 审批） ---- def propose_change(self, harness_name, field, new_value, reason): h = self.get_harness(harness_name) if not h: return {"ok": False, "error": "harness_not_found"} change = { "change_id": f"RSI-{uuid.uuid4().hex[:8].upper()}", "harness": harness_name, "field": field, "old_value": h.get(field), "new_value": new_value, "reason": reason, "status": "pending_t0_review", "created": now(), "did": DID, "anchor": ANCHOR, } self.db["pending_changes"].append(change) _save_json(RSI_DB, self.db) return {"ok": True, "change_id": change["change_id"], "status": "pending_t0_review"} def approve_change(self, change_id, reviewer="human"): c = next((x for x in self.db["pending_changes"] if x["change_id"] == change_id), None) if not c: return {"ok": False, "error": "change_not_found"} self.db["pending_changes"] = [x for x in self.db["pending_changes"] if x["change_id"] != change_id] c["status"] = "approved"; c["reviewer"] = reviewer; c["approved_at"] = now() # 应用到 Harness（版本递增） h = self.db["harness"].get(c["harness"]) if h: h[c["field"]] = c["new_value"] h["version"] = h.get("version", 1) + 1 self.db["approved_changes"].append(c) _save_json(RSI_DB, self.db) return {"ok": True, "change_id": change_id, "harness": c["harness"], "version": (h or {}).get("version")} def stats(self): return { "harness_count": len(self.db["harness"]), "artifacts": len(self.db["artifacts"]), "pending_changes": len(self.db["pending_changes"]), "approved_changes": len(self.db["approved_changes"]), "anchor": ANCHOR, } def _demo(): print("== A6 RSI 双维度自修改 demo ==") rsi = RSIEngine() h = Harness("qa-answer", prompt="请基于真值库回答问题，输出简洁结论。", skills=["retrieval"], tools=["truth_search"], rails=["禁编造"]) rsi.save_harness(h) print("Harness 方法层保存, hash:", h.to_dict()["hash"]) r = rsi.run("qa-answer", "openJiuwen 何时发布？") print("产出 Artifact id:", r["artifact"]["id"], "| kind:", r["artifact"]["kind"]) c = rsi.propose_change("qa-answer", "prompt", "请基于真值库回答问题，输出简洁结论，并附依据。", "badcase: 缺少依据") print("自修改候选:", c["change_id"], "|", c["status"]) ap = rsi.approve_change(c["change_id"], reviewer="demo") print("审批后 Harness 版本:", ap["version"]) print("RSI 统计:", json.dumps(rsi.stats(), ensure_ascii=False)) return rsi.stats() if __name__ == "__main__": import sys if len(sys.argv) > 1 and sys.argv[1] == "stats": print(json.dumps(RSIEngine().stats(), ensure_ascii=False, indent=2)) else: _demo()