File size: 4,857 Bytes
e8ccbfa | 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 | # -*- coding: utf-8 -*-
"""
函式庫版整合範例(不開網頁)—— 直接在你的 Python 程式裡跑問答、拿 9 維權重。
與網頁版共用同一個接點 recommender_hook.deliver_weights():
- 網頁版:python run_web.py(瀏覽器問答,完成後自動交付)
- 函式庫版:本檔(程式內問答,完成後自動交付)
每一輪回答後都會輸出該輪的 μ(強度讀數)與 σ(不確定性),並把完整逐輪過程
存成 elicitation_trace.json,供學術報告完整留存萃取過程。
執行(互動測試):
python integrate_example.py
首次會載入 BGE-M3(本機有 encoder_model/ 則離線,否則自動下載 BAAI/bge-m3,約 1–2 分鐘)。
"""
import sys
import json
from pathlib import Path
BUNDLE_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(BUNDLE_DIR))
from phase3_system import Phase3Engine # noqa: E402
from recommender_hook import deliver_weights # noqa: E402 ← 共用接點
TRACE_PATH = BUNDLE_DIR / "elicitation_trace.json"
def extract_preferences(philosophy_text, answer_fn, max_questions=9, verbose=True):
"""跑完問答,回傳 (weights, snapshot, trace)。
trace: 逐輪過程清單,每輪含
step / dim_key / dim_label / mu(強度) / sigma(不確定性) /
gate_rel(離題守門可靠度) / revision(重問移動量) / is_reask /
Sigma_alpha(當輪整體確定度) / dim_Ew(該維權重) / dim_ci90(該維 90% 信賴區間)
—— 供學術報告完整呈現「每一題如何更新信念」。
"""
engine = Phase3Engine()
engine.start_session(philosophy_text)
trace = []
asked = 0
while asked < max_questions:
q = engine.next_question()
if q is None:
break
snap = engine.submit_answer(answer_fn(q))
asked += 1
lt = snap.get("last_turn", {}) or {}
dk = lt.get("dim_key")
rk = next((r for r in snap.get("ranking", []) if r.get("dim_key") == dk), {})
rec = {
"step": asked, "dim_key": dk, "dim_label": lt.get("dim_label"),
"mu": lt.get("mu"), "sigma": lt.get("sigma"),
"gate_rel": lt.get("gate_rel"), "revision": lt.get("revision"),
"is_reask": lt.get("is_reask"),
"Sigma_alpha": snap.get("Sigma_alpha"),
"dim_Ew": rk.get("Ew"), "dim_ci90": rk.get("ci90"),
}
trace.append(rec)
if verbose:
tag = "（重問）" if rec["is_reask"] else ""
ci = rec["dim_ci90"] or [None, None]
print(f" ↳ μ 強度={rec['mu']:.3f} σ 不確定性={rec['sigma']:.3f} "
f"gate={rec['gate_rel']:.2f} Σα={rec['Sigma_alpha']:.2f} "
f"該維 E[w]={rec['dim_Ew']:.3f} CI90=[{ci[0]:.3f},{ci[1]:.3f}]{tag}")
if snap.get("should_stop"):
break
snap = engine.snapshot()
return snap["Ew"], snap, trace
def run(philosophy_text, answer_fn, max_questions=9, verbose=True, save_trace=True):
"""完整流程:萃取權重(逐輪輸出 μ/σ)→ 透過 recommender_hook 交付 → 存逐輪 trace。
回傳 (weights, snapshot, trace, delivery_result)。"""
weights, snap, trace = extract_preferences(philosophy_text, answer_fn, max_questions, verbose)
result = deliver_weights(weights, snap) # ← 你的推薦邏輯(在 recommender_hook.py)
if save_trace:
TRACE_PATH.write_text(json.dumps({
"weights": weights,
"ranking": snap.get("ranking"),
"Sigma_alpha": snap.get("Sigma_alpha"),
"ci_trustworthy": snap.get("ci_trustworthy"),
"ci_note": snap.get("ci_note"),
"trace": trace,
}, ensure_ascii=False, indent=2), encoding="utf-8")
return weights, snap, trace, result
def _interactive():
print("=" * 60)
print(" Phase 3 偏好誘出 — 函式庫版整合測試(逐輪輸出 μ / 不確定性)")
print("=" * 60)
phil = input("\n請輸入投資理念(直接 Enter 用預設範例):\n> ").strip()
if not phil:
phil = "我希望長期穩健成長,能接受一點波動但很怕大跌,偏好低費用、分散持股的標的。"
print(f"(使用預設) {phil}")
def ask(q):
print(f"\n[第 {q['step']} 題 · {q['dim_label']}]")
return input(f" {q['question']}\n > ").strip()
weights, snap, trace, result = run(phil, ask)
print("\n" + "=" * 60)
print(" 9 維 ETF 偏好權重(總和=1)")
print("=" * 60)
for it in snap["ranking"]:
print(f" {it['Ew']:.3f} {'█'*int(it['Ew']*40):<40} {it['dim_label']}")
print(f"\n 確定度 Σα = {snap['Sigma_alpha']} ｜ {snap['ci_note']}")
print(f" 共 {len(trace)} 輪過程已存檔:{TRACE_PATH.name}")
print(f" recommender_hook 回傳 = {result}")
if __name__ == "__main__":
_interactive()
|