# app.py — minimal for LLM + RAG (no MCP)
import os, json, random, uuid
from datetime import datetime
import streamlit as st
import rag_utils # 你已經有的：提供 search_tarot / search_numerology / ensure_indexes
from data import load_tarot_data_full, numerology_data_full
TAROT_DATA = load_tarot_data_full() # List[Dict]
NUMEROLOGY_DATA = numerology_data_full() # List[Dict]
# 若你需要只要牌名/數字，可自行萃取：
TAROT_NAMES = [c.get("name") or c.get("card_name") for c in TAROT_DATA if (c.get("name") or c.get("card_name"))]
# ========== 基本設定 ==========
st.set_page_config(page_title="SoulCompass", page_icon="🧭", layout="wide")
# 使用者暫時 UID（未登入先用本機隨機 ID）
if "uid" not in st.session_state:
st.session_state.uid = str(uuid.uuid4())[:8]
UID = st.session_state.uid
USER_DIR = os.path.join("data", "users", UID)
os.makedirs(USER_DIR, exist_ok=True)
JOURNAL_PATH = os.path.join(USER_DIR, "journal.jsonl")
st.markdown("""
🧭 SoulCompass — Tarot • Numerology • Journal • Guide
""", unsafe_allow_html=True)
# ========== 讀取 Tarot / Numerology 資料 ==========
@st.cache_resource
def load_tarot_names():
path = os.path.join("data", "tarot_data_full.json")
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
names = []
for c in data:
nm = c.get("name") or c.get("card_name")
if nm:
names.append(nm)
if not names:
raise ValueError("No card names found in data/tarot_data_full.json")
return names
@st.cache_resource
def load_numerology_data():
path = os.path.join("data", "numerology_data_full.json")
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
by_num, nums = {}, []
for r in data:
if "number" not in r:
continue
n = int(r["number"])
nums.append(n)
by_num[n] = {
"name": r.get("name",""),
"keywords": r.get("keywords",[]),
"description": r.get("description",""),
"advice": r.get("advice","")
}
nums = sorted(set(nums))
if not nums:
raise ValueError("No numerology numbers found in data/numerology_data_full.json")
return nums, by_num
def life_path_number_from_date(dt):
s = dt.strftime("%Y%m%d")
total = sum(int(d) for d in s)
while total > 9 and total not in (11, 22, 33):
total = sum(int(d) for d in str(total))
return total
TAROT_NAMES = load_tarot_data_full()
NUMBERS, NUM_MAP = load_numerology_data_full()
# ========== Sidebar：模型與 RAG 參數 ==========
st.sidebar.header("⚙️ Settings")
model_choice = st.sidebar.selectbox(
"LLM Model",
["cerebras/btlm-3b-8k-base", "mistralai/Mistral-7B-Instruct-v0.2"],
help="BTLM 輕量、CPU 可跑；Mistral 口條更好但較吃資源。"
)
rag_k = st.sidebar.slider("RAG Top-K", 1, 5, 3)
temperature = st.sidebar.slider("Temperature", 0.2, 1.2, 0.7, 0.1)
# ========== RAG + LLM Loader ==========
@st.cache_resource
def load_rag():
try:
import rag_utils
return rag_utils
except Exception as e:
st.warning("無法載入 rag_utils，請確認檔案存在於專案根目錄。")
return None
@st.cache_resource
def load_llm(model_id: str):
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
tok = AutoTokenizer.from_pretrained(model_id, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
model_id,
trust_remote_code=True,
torch_dtype=torch.float32, # 免費 CPU 最穩
device_map="auto",
low_cpu_mem_usage=True,
offload_folder="/tmp",
)
return pipeline(
"text-generation",
model=model,
tokenizer=tok,
max_new_tokens=360,
temperature=temperature,
top_p=0.9,
repetition_penalty=1.08,
do_sample=True,
)
rag = load_rag()
llm = load_llm(model_choice)
def rag_answer(query: str, domain: str = "Tarot", k: int = 3):
if rag is None:
return "RAG 模組未載入，請檢查 rag_utils.py。", []
hits = rag.search_tarot(query, k) if domain == "Tarot" else rag.search_numerology(query, k)
context = "\n\n".join([h["text"] for h in hits]) if hits else "（無檢索結果）"
prompt = f"""You are SoulCompass, a compassionate spiritual guide.
Use ONLY the CONTEXT to answer the USER in Traditional Chinese with empathy and clarity.
If info is missing, admit briefly and suggest a next step.
CONTEXT:
{context}
USER QUESTION: {query}
請用 5~8 句繁體中文回覆，語氣溫柔、務實，最後給出 1 句可行的小建議。"""
out = llm(prompt)[0]["generated_text"]
answer = out[len(prompt):].strip()
return answer, hits
# ========== Journal 長期保存（JSONL）==========
def load_journal(path=JOURNAL_PATH):
entries = []
if os.path.exists(path):
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
entries.append(json.loads(line))
except Exception:
pass
return entries
def append_journal(entry: dict, path=JOURNAL_PATH):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "a", encoding="utf-8") as f:
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
def save_all_journal(entries, path=JOURNAL_PATH):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
for e in entries:
f.write(json.dumps(e, ensure_ascii=False) + "\n")
def export_journal_csv(path=JOURNAL_PATH):
if not os.path.exists(path):
return None
rows = load_journal(path)
out_csv = os.path.join(USER_DIR, "journal_export.csv")
import csv
with open(out_csv, "w", encoding="utf-8", newline="") as f:
w = csv.writer(f)
w.writerow(["time", "mood", "emotion", "text"])
for r in rows:
w.writerow([r.get("time",""), r.get("mood",""), r.get("emotion",""), r.get("text","")])
return out_csv
if "journal" not in st.session_state:
st.session_state.journal = load_journal()
# ========== AI 心靈導師（安全護欄＋生成）==========
CRISIS_KEYWORDS = [
"自殺","想死","結束生命","活不下去","傷害自己","割腕",
"殺了我","想不開","無法活","輕生","自殘","不如死了"
]
def is_crisis(text: str) -> bool:
if not text:
return False
# 中文大小寫無差；直接檢索原文關鍵字
return any(k in text for k in CRISIS_KEYWORDS)
def guide_answer(user_msg: str, topic: str = "General"):
if is_crisis(user_msg):
crisis_reply = (
"我聽見你現在非常難受，謝謝你願意說出來。你的感受很重要，也值得被好好照顧。\n\n"
"目前這些想法可能代表你需要更即時、面對面的支持。請立刻聯絡你所在地的緊急服務，"
"或尋求可信任的人陪伴（家人、朋友、學校輔導/公司 EAP）。\n\n"
"你也可以尋找在地的心理諮商與危機支援資源（醫院身心科、社福單位、24 小時關懷專線）。\n\n"
"現在先做三件小事：1) 找一個安全的空間，2) 打給一位你信任的人，3) 緩慢做 5 次深呼吸（吸4、停2、吐6）。\n\n"
"我在這裡，願意繼續聽你說。"
)
return crisis_reply, []
# 可選：混入少量 RAG 提示（用 Numerology 當中性語料）
try:
ctx_hits = rag.search_numerology("自我照顧 情緒 調節 建議", k=2) if rag else []
context = "\n\n".join([h["text"] for h in ctx_hits]) if ctx_hits else "（無檢索）"
except Exception:
ctx_hits, context = [], "（無檢索）"
prompt = f"""You are SoulCompass, a supportive, non-clinical companion.
- Respond in Traditional Chinese.
- Be empathetic, specific, and action-oriented.
- Do NOT diagnose conditions or offer medical advice.
- Offer 2~4 small, doable next steps, tailored to the user's message.
- Keep it under 10 sentences.
OPTIONAL CONTEXT (self-care ideas):
{context}
TOPIC: {topic}
USER: {user_msg}
請用溫柔、務實、去評判的語氣回覆，先同理、再聚焦、最後給 2~4 個可行的小步驟。"""
out = llm(prompt)[0]["generated_text"]
return out[len(prompt):].strip(), ctx_hits
# ========== Tabs ==========
tab1, tab2, tab3, tab4 = st.tabs(["🔮 Tarot", "🔢 Numerology", "📖 Journal", "🤖 AI 心靈導師"])
# ---- Tarot ----
with tab1:
st.subheader("🔮 Tarot Reading")
q = st.text_input("想詢問什麼？（例：我適合換工作嗎？）", key="tarot_q")
spread = st.radio("抽牌陣列", ["Single (1)", "Three (3)", "Five (5)"], horizontal=True, key="tarot_spread")
if st.button("抽牌並解讀", key="tarot_go"):
if not q.strip():
st.warning("請先輸入問題")
else:
n = 1 if "1" in spread else (3 if "3" in spread else 5)
cards = random.sample(TAROT_NAMES, k=min(n, len(TAROT_NAMES)))
st.success("你抽到：" + ", ".join(cards))
query = f"問題：{q}；抽到的牌：{', '.join(cards)}。請用塔羅牌義做整體解讀。"
with st.spinner("RAG 檢索 + 生成中…"):
ans, refs = rag_answer(query, domain="Tarot", k=rag_k)
st.markdown(ans)
if refs:
with st.expander("參考段落（RAG 來源）"):
for i, r in enumerate(refs, 1):
tag = r.get("card_name") or r.get("name") or "Card"
st.markdown(f"**#{i} {tag}** · score {r['score']:.3f}")
st.write(r["text"])
st.markdown("---")
# ---- Numerology ----
with tab2:
st.subheader("🔢 Numerology")
b = st.date_input("你的生日", key="dob")
col1, col2 = st.columns(2)
with col1:
if st.button("計算生命靈數", key="lp_go"):
if b:
lp = life_path_number_from_date(b)
st.success(f"你的生命靈數：{lp}（{NUM_MAP.get(lp,{}).get('name','')}）")
qn = f"生命靈數 {lp} 的核心特質、盲點與行動建議？"
with st.spinner("RAG 檢索 + 生成中…"):
ans, refs = rag_answer(qn, domain="Numerology", k=rag_k)
st.markdown(ans)
with col2:
name = st.text_input("你的姓名（可選）", key="name_num")
if st.button("名字數字", key="name_go"):
if name.strip():
n = (len(name.replace(" ", "")) % 9) + 1
st.info(f"名字數字：{n}（{NUM_MAP.get(n,{}).get('name','')}）")
qn = f"名字數字 {n} 的人在人際與自我表達上的建議？"
ans, _ = rag_answer(qn, domain="Numerology", k=rag_k)
st.write(ans)
# ---- Journal ----
with tab3:
st.subheader("📖 Soul Journal")
st.caption(f"本機使用者 ID：{UID}")
mood = st.slider("今天的心情（1-10）", 1, 10, 7, key="mood")
emotion = st.selectbox("主要情緒", ["😊 開心","😔 失落","😰 焦慮","😌 平靜","🤗 感恩"], key="emotion")
entry = st.text_area("寫下今天的想法", height=140, placeholder="你最近在思考什麼？今天最感謝的一件事是？", key="entry")
c1, c2, c3 = st.columns(3)
with c1:
if st.button("💾 儲存日記", key="save_journal"):
if entry.strip():
item = {
"time": datetime.now().strftime("%Y-%m-%d %H:%M"),
"mood": st.session_state.mood,
"emotion": st.session_state.emotion,
"text": entry.strip()
}
st.session_state.journal.append(item)
append_journal(item)
st.success("已儲存 🌟")
else:
st.warning("請先寫點內容")
with c2:
if st.button("⬇️ 匯出 CSV", key="export_journal"):
out_csv = export_journal_csv()
if out_csv and os.path.exists(out_csv):
with open(out_csv, "rb") as f:
st.download_button("下載日記 CSV", f, file_name="journal_export.csv")
else:
st.warning("目前沒有日記可匯出")
with c3:
if st.button("🗑️ 清空全部日記", key="clear_journal"):
save_all_journal([])
st.session_state.journal = []
st.success("全部日記已清空")
if st.session_state.journal:
st.markdown("---")
st.markdown("**最近紀錄（新→舊）**")
for j in reversed(st.session_state.journal[-50:]):
st.markdown(f"- **{j['time']} · {j['emotion']} · 心情 {j['mood']}** \n{j['text']}")
if st.button("🧠 生成 7 日回顧建議", key="review7"):
recent = "\n".join([j["text"] for j in st.session_state.journal[-20:]])
qj = f"這是我最近的日記重點：\n{recent}\n請用溫柔務實的語氣，總結 3 點觀察，並給 3 個可行的小建議。"
ans, _ = rag_answer(qj, domain="Numerology", k=2) # 用較中性的語料
st.markdown(ans)
# ---- AI 心靈導師 ----
with tab4:
st.subheader("🤖 AI 心靈導師")
st.caption("此功能提供一般性支持與自我照顧建議，非醫療或心理治療之替代。若有危機，請立即尋求在地緊急支援。")
topic = st.selectbox(
"主題",
["General", "Stress & Anxiety", "Relationships", "Work & Burnout", "Self-esteem", "Grief & Loss"],
index=0,
key="guide_topic"
)
user_msg = st.text_area("想跟我說點什麼？", height=140, placeholder="把現在的感受、困擾或情境說給我聽…", key="guide_msg")
if st.button("💬 取得支持", key="guide_go"):
if not user_msg.strip():
st.warning("請先輸入一段想說的話")
else:
with st.spinner("AI 正在整理回應…"):
reply, refs = guide_answer(user_msg, topic)
st.markdown(reply)
if refs:
with st.expander("參考段落（RAG 來源）"):
for i, r in enumerate(refs, 1):
st.markdown(f"**#{i}** · score {r['score']:.3f}")
st.write(r["text"])
st.markdown("---")
# ========== Footer ==========
st.markdown("
", unsafe_allow_html=True)
st.caption("🧭 SoulCompass · demo MVP · RAG + LLM powered")