File size: 7,954 Bytes
6d07351 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 |
# simulate.py (SimulationLayer) V13.3.3+ Attainment Snapshot
from __future__ import annotations
from typing import Any, Dict
import copy
import traceback
from .terminals import DSLTerminal
class SimulationLayer:
def simulate(self, match_result: Any, decomposition: Any, mined_knowledge: Dict[str, Any] = None) -> Dict[str, Any]:
terminal = DSLTerminal()
if not mined_knowledge:
mined_knowledge = {}
query_text = mined_knowledge.get("QUERY_TEXT") or getattr(decomposition, "text_part", "") or ""
ops_list = mined_knowledge.get("OPS") or [
{"op": "EXTRACT_CLAIMS", "args": {}},
{"op": "REGISTER_DEFS", "args": {"source": "mined"}},
{"op": "LEXICAL_MATCH", "args": {}},
{"op": "VERIFY_WITH_RULESET", "args": {}},
{"op": "VERIFY_NECESSITY", "args": {}},
{"op": "SOLVE_CHOICE", "args": {}},
]
# ✅ Fix: Extract these BEFORE using them in session_vars
mined_rules = mined_knowledge.get("MINED_RULES", [])
mined_definitions = mined_knowledge.get("DEFINITIONS", {})
session_vars = {
"choices": getattr(decomposition, "slots", {}).get("choices", {}),
"frame_props": getattr(decomposition, "slots", {}).get("frame_props", []),
"refute_mode": getattr(decomposition, "slots", {}).get("refute_mode", False),
"need_exactness": getattr(decomposition, "slots", {}).get("need_exactness", False),
"intent_candidates": getattr(decomposition, "slots", {}).get("intent_candidates", []),
"entity": getattr(decomposition, "slots", {}).get("entity", ""),
"mined_rules": mined_rules,
"mined_definitions": mined_definitions,
"query_text": query_text,
"claims": [],
"verified_map": [],
"rule_hits": [],
"refutation_hits": [],
"evidence_count": 0,
}
steps_log = []
verified_results = []
evidence_count = 0
# ★追加：Reasoning Attainment 用コンテナ
attainment = {
"total_ops": [op.get("op", "UNKNOWN") for op in ops_list],
"completed": [],
"failed_at": None,
"error": None,
"traceback": None,
"snapshots": {}, # op名 -> スナップショット
}
print(f"⚙️ [SIM] OPS List: {[op.get('op') for op in ops_list]}")
def snapshot_state(op_name: str, res: Dict[str, Any] | None):
"""
UIに出したい「直前までの状態」を最小限で保存
"""
attainment["snapshots"][op_name] = {
"op": op_name,
"verdict": (res or {}).get("verdict"),
"result": (res or {}).get("result"),
"frame_props": list(session_vars.get("frame_props", [])),
"choices": list((session_vars.get("choices") or {}).keys()),
"registered_defs_count": len(session_vars.get("mined_definitions") or {}),
"claims_count": len(session_vars.get("claims") or []),
"rule_hits_count": len(session_vars.get("rule_hits") or []),
"refutation_hits_count": len(session_vars.get("refutation_hits") or []),
"evidence_count": evidence_count,
}
final_verdict = "insufficient_evidence"
for op_data in ops_list:
op_name = op_data.get("op", "UNKNOWN")
print(f"⚙️ [SIM] Op: {op_name}")
try:
# Context Injection
if op_name.upper() == "SOLVE_CHOICE":
op_data.setdefault("args", {})
op_data["args"]["choices"] = session_vars["choices"]
op_data["args"]["question"] = query_text
session_vars["evidence_count"] = evidence_count
res = terminal.execute_op(op_data, session_vars)
if res is None:
res = {"verdict": "rejected", "error": "None result"}
# Capture final verdict from significant ops
if op_name.upper() in ["SOLVE_CHOICE", "GENERATE_EXPLANATION", "APPLY_SCHEMA"]:
final_verdict = res.get("verdict", "insufficient_evidence")
# --- 通常パイプライン ---
if "claims" in res:
session_vars["claims"] = res["claims"]
if "verified_map" in res:
session_vars["verified_map"] = res["verified_map"]
if "rule_hits" in res:
session_vars["rule_hits"].extend(res["rule_hits"])
if "refutation_hits" in res:
session_vars["refutation_hits"].extend(res["refutation_hits"])
# ログ
verdict = res.get("verdict")
result_txt = res.get("result", "")
if verdict == "accepted":
steps_log.append(f"✅ PASS: {op_name} -> {result_txt}")
print(f" → Verdict: {verdict}")
print(f" → Result: {result_txt}")
verified_results.append(result_txt)
if res.get("verified"):
added = int(res.get("count", 1))
evidence_count += added
steps_log.append(f" ⭐ Logic Evidence +{added} (Total: {evidence_count})")
print(f" → Evidence added: +{added} (Total: {evidence_count})")
elif verdict == "insufficient_evidence":
reason = res.get('reason', result_txt)
steps_log.append(f"⚠️ WAIT: {op_name} -> {reason}")
print(f" → Verdict: {verdict}")
print(f" → Reason: {reason}")
else:
reason = res.get('error') or res.get('reason') or "Unknown error"
steps_log.append(f"❌ FAIL: {op_name} ({reason})")
print(f" → Verdict: {verdict} (FAIL)")
print(f" → Reason: {reason}")
# ★追加：完了として記録 & スナップショット
attainment["completed"].append(op_name)
snapshot_state(op_name, res)
except Exception as e:
# ★追加：どこで落ちたか確定
attainment["failed_at"] = op_name
attainment["error"] = str(e)
attainment["traceback"] = traceback.format_exc(limit=2)
print(f"🔥 [SIM] ERROR at {op_name}: {e}")
print(traceback.format_exc())
# “直前まで” の状態を保存（resが無いのでNone）
snapshot_state(op_name, None)
# ログ上の表示（あなたの "logic error" の根拠にする）
steps_log.append(f"❌ FAIL: {op_name} (logic error: {e})")
break
content = {
"definition": f"生産型推論シミュレーション完了 (Evidence: {evidence_count}).",
"logic_check": steps_log,
"conclusion": "\n".join(verified_results) if verified_results else "",
"summary": f"AXIS-V13.3.x Verified {evidence_count} logic units.",
# ★追加：UI表示用
"reasoning_attainment": attainment,
# ✅ Fix: Structured Metadata for Controller
"meta": {
"proofs_count": len(session_vars.get("rule_hits", [])),
"refutations_count": len(session_vars.get("refutation_hits", [])),
"evidence_count": evidence_count,
"final_verdict": final_verdict
}
}
return content |