在 Day 25，我們完成了對話記憶與來源追蹤的完整設計規劃。今天，我們將把這些設計概念轉化為實際可運作的程式碼，並透過實測發現系統的改進空間。
今日目標：
notion-llm-assistant/
├── app.py # Streamlit 主應用程式（本日更新）
├── src/
│ ├── __init__.py
│ ├── rag/
│ │ ├── __init__.py
│ │ ├── notion_rag_backend.py # RAG 核心邏輯（本日更新）
│ │ ├── conversation_memory.py # 對話記憶管理（本日新增）
│ │ └── source_tracker.py # 來源追蹤器（本日新增）
│ └── prompt/
│ ├── __init__.py
│ └── prompt_templates.py # Prompt 模板（本日新增）
├── tests/
│ ├── test_conversation_memory.py # 記憶功能測試
│ └── test_source_tracker.py # 來源追蹤測試
├── db/
│ └── chroma_db/ # 向量資料庫
├── .env # 環境變數
├── requirements.txt # 依賴套件
└── README.md # 專案說明
# requirements.txt
streamlit>=1.28.0
openai>=1.3.0
chromadb>=0.4.0
python-dotenv>=1.0.0
pandas>=2.0.0
安裝指令：
pip install -r requirements.txt
這是整個系統的記憶核心，負責儲存、管理和檢索對話歷史。
| 功能說明 | 關鍵方法 | 功能摘要 |
|---|---|---|
| 訊息儲存 | add_message() |
儲存使用者與助理的對話內容 |
| 滑動窗口 | _trim_history() |
自動維持對話記憶在合理範圍內 |
| 上下文提取 | get_context() |
格式化對話歷史，供 LLM 使用 |
| 統計資訊 | get_stats() |
追蹤對話輪數與記憶使用率 |
| 對話匯出 | export_to_markdown() |
匯出 Markdown 格式的完整對話記錄 |
src/rag/conversation_memory.py
from datetime import datetime
from typing import List, Dict, Optional
import json
import time
class ConversationMemory:
"""
對話記憶管理器
功能：
1. 儲存對話歷史
2. 實作滑動窗口機制
3. 格式化對話上下文
4. 支援對話匯出
"""
def __init__(self, max_turns: int = 10):
"""
初始化記憶管理器
Args:
max_turns: 保留的最大對話輪數（預設 10 輪）
"""
self.max_turns = max_turns
self.messages: List[Dict] = []
self.session_id = self._generate_session_id()
def _generate_session_id(self) -> str:
"""
生成唯一的 Session ID
使用時間戳記加上微秒，確保唯一性
"""
timestamp = datetime.now()
microseconds = timestamp.microsecond
return f"session_{timestamp.strftime('%Y%m%d_%H%M%S')}_{microseconds}"
def add_message(
self,
role: str,
content: str,
sources: Optional[List[Dict]] = None,
metadata: Optional[Dict] = None
) -> None:
"""
添加一則訊息到記憶中
Args:
role: "user" 或 "assistant"
content: 訊息內容
sources: 來源列表（僅 assistant）
metadata: 額外的元數據
"""
message = {
"role": role,
"content": content,
"timestamp": datetime.now().isoformat(),
"sources": sources or [],
"metadata": metadata or {}
}
self.messages.append(message)
self._trim_history()
def _trim_history(self) -> None:
"""
滑動窗口：保持記憶在限制內
策略：保留最近的 N 輪對話（1輪 = user + assistant）
"""
if len(self.messages) > self.max_turns * 2:
# 保留最近的對話
self.messages = self.messages[-(self.max_turns * 2):]
def get_context(self, include_sources: bool = False) -> str:
"""
獲取格式化的對話上下文
Args:
include_sources: 是否包含來源資訊
Returns:
格式化的對話歷史字串
"""
if not self.messages:
return ""
context_lines = []
for msg in self.messages:
role = msg["role"]
content = msg["content"]
# 格式化角色名稱
role_display = "使用者" if role == "user" else "助理"
context_lines.append(f"{role_display}: {content}")
# 可選：包含來源資訊
if include_sources and msg.get("sources"):
sources_text = self._format_sources_brief(msg["sources"])
context_lines.append(f" [來源: {sources_text}]")
return "\n".join(context_lines)
def _format_sources_brief(self, sources: List[Dict]) -> str:
"""簡短格式化來源資訊"""
titles = [s.get("title", "未知") for s in sources[:3]]
if len(sources) > 3:
return f"{', '.join(titles)}... 等 {len(sources)} 則"
return ", ".join(titles)
def get_messages(self) -> List[Dict]:
"""獲取所有訊息"""
return self.messages.copy()
def get_last_n_turns(self, n: int = 3) -> List[Dict]:
"""
獲取最近 N 輪對話
Args:
n: 要獲取的輪數
Returns:
最近的對話列表
"""
return self.messages[-(n * 2):] if self.messages else []
def clear(self) -> None:
"""清空記憶並生成新的 Session ID"""
self.messages = []
# 確保新的 Session ID 與舊的不同
time.sleep(0.001) # 等待 1 毫秒
self.session_id = self._generate_session_id()
def export_to_markdown(self) -> str:
"""
匯出對話記錄為 Markdown
Returns:
Markdown 格式的對話記錄
"""
lines = [
f"# 對話記錄",
f"**Session ID**: {self.session_id}",
f"**訊息數量**: {len(self.messages)}",
"",
"---",
""
]
for i, msg in enumerate(self.messages, 1):
role = "👤 使用者" if msg["role"] == "user" else "🤖 助理"
timestamp = msg["timestamp"]
content = msg["content"]
lines.append(f"## {i}. {role}")
lines.append(f"*時間: {timestamp}*")
lines.append("")
lines.append(content)
# 添加來源資訊
if msg.get("sources"):
lines.append("")
lines.append("**📚 參考來源**:")
for source in msg["sources"]:
lines.append(f"- {source.get('title')} ({source.get('category')})")
lines.append("")
lines.append("---")
lines.append("")
return "\n".join(lines)
def get_stats(self) -> Dict:
"""
獲取記憶統計資訊
Returns:
統計資訊字典
"""
user_messages = [m for m in self.messages if m["role"] == "user"]
assistant_messages = [m for m in self.messages if m["role"] == "assistant"]
return {
"session_id": self.session_id,
"total_messages": len(self.messages),
"user_messages": len(user_messages),
"assistant_messages": len(assistant_messages),
"turns": len(user_messages),
"max_turns": self.max_turns,
"memory_usage": f"{len(user_messages)}/{self.max_turns} 輪"
}
pythondef _trim_history(self):
if len(self.messages) > self.max_turns * 2:
# 保留最近的對話
self.messages = self.messages[-(self.max_turns * 2):]
user + assistant = 2 則訊息pythondef get_context(self):
# 將對話歷史轉換為 LLM 可理解的格式
context_lines = []
for msg in self.messages:
context_lines.append(f"{role}: {content}")
return "\n".join(context_lines)
JSON 格式：結構化數據，方便程式處理Markdown 格式：可讀性高，方便分享負責追蹤、格式化和顯示答案的來源資訊。
| 功能說明 | 關鍵方法 | 功能摘要 |
|---|---|---|
| 來源格式化 | format_sources() |
將檢索結果轉為結構化來源格式 |
| 可信度評估 | _calculate_confidence() |
根據相似度分數評估來源可信度 |
| 文字截斷 | _create_snippet() |
智慧截斷長文字為可閱讀的預覽片段 |
| 視覺化 | _format_similarity_bar() |
生成相似度進度條，提升可讀性 |
| 統計追蹤 | get_stats() |
記錄並分析所有來源的統計資訊 |
src/rag/source_tracker.py
from typing import List, Dict, Tuple
from datetime import datetime
class SourceTracker:
"""
來源追蹤器
功能：
1. 格式化檢索到的來源
2. 評估來源可信度
3. 生成來源顯示內容
"""
def __init__(self):
self.tracked_sources = []
def format_sources(
self,
context_pairs: List[Tuple], # [(doc, metadata), ...]
similarities: List[float] = None, # 可選的相似度分數
max_snippet_length: int = 150
) -> List[Dict]:
"""
格式化檢索到的來源
Args:
context_pairs: retrieve_context 返回的 (doc, metadata) 對
similarities: 相似度分數列表（如果有的話）
max_snippet_length: 文字片段的最大長度
Returns:
格式化後的來源列表
"""
sources = []
for i, (doc, metadata) in enumerate(context_pairs):
# 如果沒有提供相似度，預設為 0.8
similarity = similarities[i] if similarities and i < len(similarities) else 0.8
source = {
"title": metadata.get("Title", metadata.get("title", "未命名筆記")),
"category": metadata.get("Category", metadata.get("category", "未分類")),
"snippet": self._create_snippet(doc, max_snippet_length),
"similarity": round(float(similarity), 3),
"block_id": metadata.get("block_id", ""),
"confidence": self._calculate_confidence(similarity),
"retrieved_at": datetime.now().isoformat()
}
sources.append(source)
# 按相似度排序
sources.sort(key=lambda x: x["similarity"], reverse=True)
# 記錄追蹤
self.tracked_sources.extend(sources)
return sources
def _create_snippet(self, text: str, max_length: int) -> str:
"""
創建文字預覽片段
Args:
text: 完整文字
max_length: 最大長度
Returns:
截斷後的文字片段
"""
if not text:
return "（無內容）"
# 移除多餘空白
text = " ".join(text.split())
if len(text) <= max_length:
return text
# 截斷並添加省略號
return text[:max_length].rsplit(' ', 1)[0] + "..."
def _calculate_confidence(self, similarity_score: float) -> str:
"""
根據相似度分數計算可信度等級
Args:
similarity_score: 相似度分數 (0-1)
Returns:
可信度等級: "高", "中", "低"
"""
if similarity_score >= 0.85:
return "高"
elif similarity_score >= 0.70:
return "中"
else:
return "低"
def generate_display_text(self, sources: List[Dict], max_display: int = 3) -> str:
"""
生成來源顯示文字（用於 UI）
Args:
sources: 來源列表
max_display: 最多顯示幾個來源
Returns:
格式化的顯示文字
"""
if not sources:
return "📚 無參考來源"
display_sources = sources[:max_display]
lines = [f"📚 參考來源 ({len(sources)} 則筆記)：\n"]
for i, source in enumerate(display_sources, 1):
# 可信度圖示
confidence_icon = self._get_confidence_icon(source["confidence"])
lines.append(
f"{i}. {confidence_icon} **{source['title']}** — {source['category']}"
)
lines.append(f" {source['snippet']}")
lines.append(f" 相似度: {self._format_similarity_bar(source['similarity'])}")
lines.append("")
if len(sources) > max_display:
lines.append(f"... 還有 {len(sources) - max_display} 則相關筆記")
return "\n".join(lines)
def _get_confidence_icon(self, confidence: str) -> str:
"""根據可信度返回對應圖示"""
icons = {
"高": "✅",
"中": "⚠️",
"低": "ℹ️"
}
return icons.get(confidence, "📄")
def _format_similarity_bar(self, similarity: float) -> str:
"""
格式化相似度為進度條
Args:
similarity: 相似度分數 (0-1)
Returns:
視覺化的進度條
"""
filled = int(similarity * 10)
bar = "█" * filled + "░" * (10 - filled)
percentage = int(similarity * 100)
return f"{bar} {percentage}%"
def get_stats(self) -> Dict:
"""
獲取來源追蹤統計
Returns:
統計資訊字典
"""
if not self.tracked_sources:
return {
"total_sources": 0,
"avg_similarity": 0,
"high_confidence": 0,
"medium_confidence": 0,
"low_confidence": 0
}
similarities = [s["similarity"] for s in self.tracked_sources]
confidences = [s["confidence"] for s in self.tracked_sources]
return {
"total_sources": len(self.tracked_sources),
"avg_similarity": round(sum(similarities) / len(similarities), 3),
"max_similarity": max(similarities),
"min_similarity": min(similarities),
"high_confidence": confidences.count("高"),
"medium_confidence": confidences.count("中"),
"low_confidence": confidences.count("低")
}
def clear_tracking(self):
"""清空追蹤記錄"""
self.tracked_sources = []
pythondef _calculate_confidence(self, similarity_score):
if similarity_score >= 0.85:
return "高" # 非常相關
elif similarity_score >= 0.70:
return "中" # 中等相關
else:
return "低" # 弱相關，建議確認
pythondef _format_similarity_bar(self, similarity):
filled = int(similarity * 10)
bar = "█" * filled + "░" * (10 - filled)
# 結果: ████████░░ 80%
pythondef _create_snippet(self, text, max_length):
# 不在單字中間截斷
return text[:max_length].rsplit(' ', 1)[0] + "..."
集中管理所有 Prompt 模板，方便調整和維護。
src/prompt/prompt_templates.py
class PromptTemplates:
"""Prompt 模板管理"""
@staticmethod
def get_rag_prompt_with_memory(
conversation_history: str,
context_text: str, # ← 確保參數名稱是 context_text
user_query: str
) -> str:
"""
生成包含對話記憶的 RAG Prompt
Args:
conversation_history: 對話歷史
context_text: 檢索到的筆記內容
user_query: 當前問題
Returns:
完整的 Prompt
"""
# 如果有對話歷史，加入提示
history_section = ""
if conversation_history:
history_section = f"""
【對話歷史】
{conversation_history}
請參考上面的對話歷史來理解當前問題中的代詞（它、那個、剛才說的等）。
"""
prompt = f"""你是一位資料助理，請根據以下內容回答問題。
{history_section}
【筆記內容】
{context_text}
【當前問題】
{user_query}
【回答指引】
1. 如果有對話歷史，請理解問題中的代詞指的是什麼
2. 根據提供的筆記內容回答，若無相關資訊請說「資料庫中尚無此內容」
3. 使用繁體中文回答，保留英文專有名詞
4. 回答要清晰有條理
請回答："""
return prompt
@staticmethod
def get_system_prompt() -> str:
"""獲取系統提示"""
return "You are a helpful assistant for knowledge retrieval from personal notes."
@staticmethod
def get_no_context_response() -> str:
"""當沒有相關上下文時的回應"""
return """抱歉，我在您的筆記中找不到相關的內容。
您可以：
1. 嘗試用不同的關鍵字重新描述問題
2. 確認相關筆記是否已經匯入系統
3. 檢查筆記內容是否包含您要查詢的資訊"""
現在我們要將記憶和來源追蹤整合到原有的 RAG 系統中。
notion_rag_backend.py
import sys
import os
from typing import List, Tuple, Dict
import chromadb
from openai import OpenAI
from dotenv import load_dotenv
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, project_root)
from conversation_memory import ConversationMemory
from source_tracker import SourceTracker
from src.prompt.prompt_templates import PromptTemplates
# 1. 載入環境變數
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# 2. 初始化 Chroma Client
chroma_client = chromadb.PersistentClient(path="db/chroma_db")
collection = chroma_client.get_or_create_collection("notion_notes")
# 3. 初始化記憶和追蹤器（全域變數，供 Streamlit 使用）
memory = ConversationMemory(max_turns=10)
source_tracker = SourceTracker()
prompt_templates = PromptTemplates()
# 4. 原有的檢索函式（保持不變）
def retrieve_context(query, top_k=3):
"""
檢索相關筆記
Args:
query: 查詢文字
top_k: 返回的結果數量
Returns:
List of (doc, metadata) tuples
"""
# 將 query 轉換為 embedding
embedding = client.embeddings.create(
model="text-embedding-3-small",
input=query
).data[0].embedding
# 從 ChromaDB 檢索最相關的 chunks
results = collection.query(
query_embeddings=[embedding],
n_results=top_k
)
docs = results["documents"][0]
metadatas = results["metadatas"][0]
return list(zip(docs, metadatas))
# 5. 新版：帶記憶的回答生成
def generate_answer_with_memory(query, top_k=3):
"""
生成包含記憶的回答（Day 26 新增）
Args:
query: 使用者問題
top_k: 檢索數量
Returns:
(answer, sources) tuple
"""
# 5.1 儲存使用者問題到記憶
memory.add_message("user", query)
# 5.2 檢索相關筆記（使用原有函式）
context_pairs = retrieve_context(query, top_k)
# 5.3 格式化來源
sources = source_tracker.format_sources(context_pairs)
# 5.4 檢查是否有相關內容
if not sources or all(s["similarity"] < 0.5 for s in sources):
no_context_response = prompt_templates.get_no_context_response()
memory.add_message("assistant", no_context_response, sources=[])
return no_context_response, []
# 5.5 組合筆記內容
context_text = "\n\n".join([f"- {doc}" for doc, _ in context_pairs])
# 5.6 獲取對話歷史
conversation_history = memory.get_context()
# 5.7 構建 Prompt（使用模板）
prompt = prompt_templates.get_rag_prompt_with_memory(
conversation_history=conversation_history,
context_text=context_text,
user_query=query
)
# 5.8 呼叫 LLM
try:
completion = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": prompt_templates.get_system_prompt()
},
{
"role": "user",
"content": prompt
}
],
temperature=0.7
)
answer = completion.choices[0].message.content
except Exception as e:
error_msg = f"生成回答時發生錯誤：{str(e)}"
memory.add_message("assistant", error_msg, sources=[])
return error_msg, sources
# 5.9 儲存助理回答到記憶
memory.add_message("assistant", answer, sources=sources)
return answer, sources
# 6. 原有的回答生成函式（保留向下相容）
def generate_answer(query):
"""
原有的回答生成函式（不帶記憶）
保留此函式以確保向下相容
"""
context_pairs = retrieve_context(query)
context_text = "\n\n".join([f"- {doc}" for doc, _ in context_pairs])
prompt = f"""
你是一位資料助理，請根據以下內容回答問題，若無相關資訊請說「資料庫中尚無此內容」：
---
{context_text}
---
問題：{query}
請以繁體中文回答，保留英文專有名詞。
"""
completion = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are a helpful assistant for knowledge retrieval."},
{"role": "user", "content": prompt}
]
)
return completion.choices[0].message.content
# 7. 新增：記憶管理函式
def clear_memory():
"""清空對話記憶"""
memory.clear()
source_tracker.clear_tracking()
def get_memory_stats():
"""獲取記憶統計"""
return memory.get_stats()
def get_source_stats():
"""獲取來源統計"""
return source_tracker.get_stats()
def export_conversation(format="markdown"):
"""
匯出對話記錄
Args:
format: "markdown" 或 "json"
"""
if format == "markdown":
return memory.export_to_markdown()
# 可以擴展支援 JSON 格式
# 8. 主程式入口（保持不變，但增加選項）
if __name__ == "__main__":
print("=== Notion RAG 系統 ===")
print("1. 使用對話記憶模式（輸入問題）")
print("2. 查看記憶統計（輸入 'stats'）")
print("3. 清空記憶（輸入 'clear'）")
print("4. 匯出對話（輸入 'export'）")
print("5. 退出（輸入 'quit'）\n")
while True:
query = input("請輸入指令或問題：").strip()
if query.lower() == 'quit':
break
elif query.lower() == 'stats':
print("\n📊 記憶統計：")
print(get_memory_stats())
print("\n📚 來源統計：")
print(get_source_stats())
elif query.lower() == 'clear':
clear_memory()
print("\n✅ 記憶已清空\n")
elif query.lower() == 'export':
markdown = export_conversation()
print("\n" + markdown + "\n")
else:
answer, sources = generate_answer_with_memory(query)
print("\n🔍 回答：\n", answer)
if sources:
print("\n📚 參考來源：")
for i, source in enumerate(sources, 1):
print(f"{i}. {source['title']} ({source['category']})")
print(f" 相似度: {source['similarity']}")
print()
# 保留原有函式
def generate_answer(query): # 不帶記憶的版本
# 原有邏輯不變
...
# 新增帶記憶的函式
def generate_answer_with_memory(query, top_k=3): # 帶記憶的版本
# 新功能
...
這樣的設計可以：
generate_answer() 如果不需要記憶generate_answer_with_memory() 享受記憶功能┌─────────────────┐
│ 使用者輸入問題 │
└────────┬────────┘
│
▼
┌─────────────────────────┐
│ memory.add_message() │ ← 儲存問題
└────────┬────────────────┘
│
▼
┌─────────────────────────┐
│ retrieve_context() │ ← 檢索相關筆記
└────────┬────────────────┘
│
▼
┌─────────────────────────┐
│ source_tracker.format() │ ← 格式化來源
└────────┬────────────────┘
│
▼
┌─────────────────────────┐
│ memory.get_context() │ ← 取得對話歷史
└────────┬────────────────┘
│
▼
┌─────────────────────────┐
│ 構建完整 Prompt │ ← 包含歷史 + 筆記
└────────┬────────────────┘
│
▼
┌─────────────────────────┐
│ 呼叫 LLM API │
└────────┬────────────────┘
│
▼
┌─────────────────────────┐
│ memory.add_message() │ ← 儲存回答
└────────┬────────────────┘
│
▼
┌─────────────────────────┐
│ 返回 (answer, sources) │
└─────────────────────────┘
app.py
import streamlit as st
from datetime import datetime
from src.rag import notion_rag_backend
# ===== 頁面配置 =====
st.set_page_config(
page_title="Notion × LLM 智慧助理",
page_icon="💬",
layout="wide",
initial_sidebar_state="expanded"
)
# ===== 自訂 CSS 樣式 =====
st.markdown("""
<style>
/* 主題色調 */
.stApp {
background-color: #f8f9fa;
}
/* 聊天訊息樣式 */
.chat-message {
padding: 1rem;
border-radius: 0.5rem;
margin-bottom: 1rem;
}
/* 來源區塊樣式 */
.source-box {
background-color: #fff3cd;
border-left: 4px solid #ffc107;
padding: 0.75rem;
margin-top: 0.5rem;
border-radius: 0.25rem;
}
/* 統計資訊樣式 */
.stats-box {
background-color: #e7f3ff;
padding: 1rem;
border-radius: 0.5rem;
border-left: 4px solid #2196F3;
}
</style>
""", unsafe_allow_html=True)
# ===== 標題 =====
st.title("💬 Notion × LLM 智慧助理")
st.markdown("透過對話的方式，探索你的 Notion 筆記")
# ===== 側邊欄設定 =====
with st.sidebar:
st.header("⚙️ 設定")
# 檢索設定
st.subheader("🔍 檢索設定")
top_k = st.slider(
"檢索數量",
min_value=1,
max_value=10,
value=3,
help="從知識庫中檢索幾筆相關內容"
)
similarity_threshold = st.slider(
"相似度門檻",
min_value=0.5,
max_value=0.95,
value=0.7,
step=0.05,
help="只顯示相似度高於此門檻的來源"
)
# LLM 設定
st.subheader("🤖 LLM 設定")
temperature = st.slider(
"創造力",
min_value=0.0,
max_value=1.0,
value=0.7,
step=0.1,
help="越高越有創意，越低越精確"
)
# 顯示設定
st.subheader("📊 顯示設定")
show_sources = st.checkbox("顯示筆記來源", value=True)
show_similarity = st.checkbox("顯示相似度分數", value=False)
show_timestamp = st.checkbox("顯示時間戳記", value=True)
# 記憶設定
st.subheader("🧠 記憶設定")
enable_memory = st.checkbox("啟用對話記憶", value=True, help="記住先前的對話內容")
# 分隔線
st.markdown("---")
# 操作按鈕
st.subheader("🛠️ 操作")
col1, col2 = st.columns(2)
with col1:
if st.button("🗑️ 清除對話", use_container_width=True):
notion_rag_backend.clear_memory()
st.session_state.chat_history = []
st.rerun()
with col2:
if st.button("📊 查看統計", use_container_width=True):
st.session_state.show_stats = True
if st.button("📥 匯出對話", use_container_width=True):
markdown = notion_rag_backend.export_conversation()
st.download_button(
label="下載 Markdown",
data=markdown,
file_name=f"conversation_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md",
mime="text/markdown"
)
# 統計資訊
if st.session_state.get('show_stats', False):
st.markdown("---")
st.subheader("📈 統計資訊")
memory_stats = notion_rag_backend.get_memory_stats()
source_stats = notion_rag_backend.get_source_stats()
st.markdown(f"""
<div class="stats-box">
<b>記憶統計</b><br>
對話輪數: {memory_stats['turns']}<br>
記憶使用: {memory_stats['memory_usage']}<br>
<br>
<b>來源統計</b><br>
總來源數: {source_stats['total_sources']}<br>
平均相似度: {source_stats.get('avg_similarity', 0):.2f}
</div>
""", unsafe_allow_html=True)
if st.button("關閉統計"):
st.session_state.show_stats = False
st.rerun()
# ===== 初始化 session_state =====
if "chat_history" not in st.session_state:
st.session_state.chat_history = []
# ===== 主要對話區域 =====
# 顯示歷史對話
for i, msg in enumerate(st.session_state.chat_history):
with st.chat_message(msg["role"]):
# 顯示訊息內容
st.markdown(msg["content"])
# 顯示時間戳記
if show_timestamp and "timestamp" in msg:
st.caption(f"⏰ {msg['timestamp']}")
# 顯示來源（僅助理訊息）
if msg["role"] == "assistant" and show_sources and "sources" in msg:
sources = msg["sources"]
if sources:
with st.expander(f"📚 參考來源 ({len(sources)} 則筆記)", expanded=False):
for j, source in enumerate(sources, 1):
# 根據可信度選擇圖示
confidence_icons = {"高": "✅", "中": "⚠️", "低": "ℹ️"}
icon = confidence_icons.get(source.get('confidence', '中'), "📄")
# 顯示來源資訊
st.markdown(f"""
**{j}. {icon} {source['title']}** — {source['category']}
""")
# 顯示文字片段
st.caption(source['snippet'])
# 可選：顯示相似度
if show_similarity:
similarity = source['similarity']
st.progress(similarity, text=f"相似度: {similarity:.2%}")
if j < len(sources):
st.markdown("---")
# ===== 使用者輸入 =====
user_query = st.chat_input("💬 請輸入你的問題...")
if user_query:
# 立即顯示使用者訊息
timestamp = datetime.now().strftime("%H:%M:%S")
with st.chat_message("user"):
st.markdown(user_query)
if show_timestamp:
st.caption(f"⏰ {timestamp}")
# 儲存使用者訊息
st.session_state.chat_history.append({
"role": "user",
"content": user_query,
"timestamp": timestamp
})
# 顯示載入狀態
with st.chat_message("assistant"):
with st.spinner("🔄 思考中..."):
try:
# 根據設定選擇是否使用記憶
if enable_memory:
answer, sources = notion_rag_backend.generate_answer_with_memory(
user_query,
top_k=top_k
)
else:
# 使用舊版本（無記憶）
answer = notion_rag_backend.generate_answer(user_query)
sources = []
# 過濾低相似度的來源
if sources:
sources = [s for s in sources if s['similarity'] >= similarity_threshold]
# 顯示回答
st.markdown(answer)
# 顯示時間戳記
assistant_timestamp = datetime.now().strftime("%H:%M:%S")
if show_timestamp:
st.caption(f"⏰ {assistant_timestamp}")
# 顯示來源
if sources and show_sources:
with st.expander(f"📚 參考來源 ({len(sources)} 則筆記)", expanded=False):
for j, source in enumerate(sources, 1):
confidence_icons = {"高": "✅", "中": "⚠️", "低": "ℹ️"}
icon = confidence_icons.get(source.get('confidence', '中'), "📄")
st.markdown(f"""
**{j}. {icon} {source['title']}** — {source['category']}
""")
st.caption(source['snippet'])
if show_similarity:
similarity = source['similarity']
st.progress(similarity, text=f"相似度: {similarity:.2%}")
if j < len(sources):
st.markdown("---")
# 儲存助理回答
st.session_state.chat_history.append({
"role": "assistant",
"content": answer,
"sources": sources,
"timestamp": assistant_timestamp
})
except Exception as e:
st.error(f"❌ 發生錯誤: {str(e)}")
st.exception(e)
# 重新執行以更新顯示
st.rerun()
# ===== 頁尾資訊 =====
st.markdown("---")
col1, col2, col3 = st.columns(3)
with col1:
if st.session_state.chat_history:
turns = len([m for m in st.session_state.chat_history if m["role"] == "user"])
st.metric("對話輪數", turns)
with col2:
memory_stats = notion_rag_backend.get_memory_stats()
st.metric("記憶使用", memory_stats['memory_usage'])
with col3:
if enable_memory:
st.success("🧠 對話記憶已啟用")
else:
st.info("💤 對話記憶已停用")
# 啟動 Streamlit 應用
streamlit run app.py
python# source_tracker.py 中
source = {
"title": metadata.get("Title", metadata.get("title", "未命名筆記")),
"category": metadata.get("Category", metadata.get("category", "未分類")),
# ...
}
Title 或 Category 欄位# 檢查 ChromaDB 內容
results = collection.get(limit=5)
for metadata in results['metadatas']:
print(metadata.keys()) # 查看實際的欄位名稱
# notion_rag_backend.py 中的邏輯
if not sources or all(s["similarity"] < 0.5 for s in sources):
return prompt_templates.get_no_context_response(), []
# 方案 1：提高相似度門檻
if not sources or all(s["similarity"] < 0.7 for s in sources):
return no_context_response, []
# 方案 2：檢查內容相關性
def check_content_relevance(query, sources):
# 檢查關鍵詞是否出現在來源中
keywords = extract_keywords(query)
for source in sources:
if any(kw in source['snippet'].lower() for kw in keywords):
return True
return False
今天我們成功實作了對話記憶與來源追蹤兩大核心功能，並透過實測發現了系統的改進空間。雖然遇到了一些問題，但這正是開發過程中寶貴的學習機會。
這些問題為我們指明了下一步的改進方向。明天開始我們將逐步解決今天發現的問題：
修正 Metadata 問題
Title 和 Category 正確儲存優化檢索邏輯
改善使用者體驗
iThome鐵人賽
看影片追技術