scripts/_card_prompt.py
Rama main · réplica de lectura del repositorio original.
"""
_card_prompt.py
Shared knowledge card generation — used by ALL ingest scripts:
run_scan_worker.py, local_ingest.py, fetch_youtube_playlist.py, fetch_github_repos.py
One unified 9-section format, source_type adapts per script.
"""
from __future__ import annotations
import os
import re
import sys
from pathlib import Path
from typing import Any
# ── Unified LLM helper (config/llm.json → single model setting) ──────────────
sys.path.insert(0, str(Path(__file__).parent))
import xkb_text
from _llm import call as _llm_call
# ── XBrain integration (path resolved by xbrain_recall, never hardcoded) ────
try:
from xbrain_recall import (
GBRAIN_DIR as _GBRAIN_DIR_OR_NONE,
GBRAIN_AVAILABLE as _GBRAIN_AVAILABLE,
GEMINI_API_KEY as _GEMINI_API_KEY,
_make_subprocess_env,
)
_GBRAIN_DIR = _GBRAIN_DIR_OR_NONE
_GBRAIN_CLI = str(_GBRAIN_DIR / "src" / "cli.ts") if _GBRAIN_DIR else ""
_GBRAIN_ENV = _make_subprocess_env(semantic=True)
except ImportError:
_GBRAIN_DIR = None
_GBRAIN_CLI = ""
_GBRAIN_AVAILABLE = False
_GBRAIN_ENV = {**os.environ}
def gbrain_put(card_path: Path, slug: str) -> str:
"""Checked publication; failures remain in the durable outbox and raise."""
from gbrain_publish import publish
return publish(card_path, slug)
# ── Shared system prompt ──────────────────────────────────────────────────────
SYSTEM_PROMPT = """\
You are a knowledge card generator for a personal learning base. \
Given content from any source — academic paper, social media bookmark, \
YouTube video, GitHub repo, or local document — output one structured \
knowledge card in Traditional Chinese.
Strict rules:
- Leave sections empty with "無" if uncertain — never hallucinate
- Use only information from the provided content
- Do NOT use the reader's personal name in any section
Quality principles: conservative > hallucination, understanding > summary, structured > verbose"""
# ── Unified 9-section card template ──────────────────────────────────────────
CARD_PROMPT = """\
以下是一份知識來源的內容，請生成一張知識卡片。
如果內容包含 `## 10. Media Evidence`、`### OCR` 或 `### Vision Notes`：
- 必須把圖片視為一級來源，不是裝飾。
- 若圖片內有 prompt、表格、截圖文字或前後對比，請在卡片中保留可複用 pattern。
- 不要捏造圖片文字；OCR/vision notes 標示不確定時，卡片也要保守標示。
- 圖片證據請整合到各 section，並在最後保留 Media Evidence 摘要。
來源類型: {source_type_label}
來源網址: {source_url}
分類: {category}
內容:
{content}
{related_section}
請輸出以下格式（YAML frontmatter + Markdown）：
---
id: {card_id}
type: knowledge-card
source_type: {source_type}
source_url: {source_url}
category: {category}
tags: [tag1, tag2, tag3]
sensitivity: public
confidence: medium
---
# <標題>
## 1. 核心問題與結論
- **提問**：這份內容試圖解答什麼問題？（一句話）
- **結論**：核心答案或發現是什麼？（一句話）
- **可信度說明**：有數據/實驗/引用支撐，還是個人意見？
## 2. Claim 等級
- **等級**：[Attested | Scholarship | Inference]
- Attested：直接引用、有具體數據或實驗結果
- Scholarship：作者/領域分析觀點，有明確來源依據
- Inference：LLM 推論、個人猜測、尚未驗證的假設
- **主要主張**：（一句話說明被標記的核心主張）
- **依據**：（為什麼是這個等級？）
## 3. 關鍵論點
- 論點一
- 論點二
- 論點三
## 4. False Friends（如有）
這份內容涉及哪些看起來像普通詞彙但有特定技術含義的術語？
- term: （術語名稱）
common_misunderstanding: （多數人誤以為是...）
actual_meaning: （在此領域/內容中實際指的是...）
如果沒有：無
## 5. 驚訝點
讀者讀完後，可能感到意外或需要重新思考的是什麼？
（如果沒有明顯驚訝點，填「無」）
## 6. 與現有知識的關係
{related_cards_placeholder}
## 7. 雙語摘要（搜尋索引用）
ZH: <20-40字繁體中文摘要，說明核心發現>
EN: <15-30 word English summary of the core finding>
## 8. 對使用者的價值
- 可追蹤的方向
- 可執行的應用場景
- 適合哪個專案或工作流程
## 9. 原始來源
- 來源: {source_url}
- Links: （列出內容中出現的其他連結，如有）
## 10. Media Evidence（如有）
如果原文提供圖片 / OCR / Vision Notes，請列出：
- 圖片角色：prompt screenshot / result image / comparison / diagram / product mockup / unknown
- 圖片中最重要的可讀文字或 prompt 摘要
- 圖像 pattern：可複用的版面、風格、構圖、prompt 寫法
- 不確定處：不可讀文字、疑似誤辨、需要人工複核的地方
如果沒有圖片證據：無
"""
# Human-readable labels per source_type
SOURCE_LABELS: dict[str, str] = {
"x-bookmark": "X / Twitter 書籤",
"youtube": "YouTube 影片",
"github-star": "GitHub 倉庫（Star）",
"github-fork": "GitHub 倉庫（Fork）",
"local-paper": "學術論文 / 本地文件",
"local": "本地文件",
"pubmed": "PubMed / PMC 論文",
}
def source_label(source_type: str) -> str:
return SOURCE_LABELS.get(source_type, source_type)
# ── LLM call ─────────────────────────────────────────────────────────────────
# 2000 是這個參數還會被丟掉的年代留下的預設值。現在它會一路傳到 API，
# 於是產卡的上限從 4096 悄悄變成 2000——沒有人改過產卡的需求。
def llm_call(prompt: str, api_key: str = "", max_tokens: int = 4096,
system: str | None = SYSTEM_PROMPT) -> str:
"""api_key kept for backwards compatibility but is no longer used.
Model is configured via config/llm.json.
max_tokens 原本收下來就丟掉，現在會一路傳進 API payload。
但要說清楚：**目前的供應商（tu-zi 的 gpt-5.6-luna）不理這個欄位**。
2026-09-01 實測 max_tokens=16 與 =400 回傳長度完全相同，finish_reason
都是 stop；換成 max_completion_tokens 也一樣。所以這個參數現在是「有
傳，但不保證生效」——換供應商時它會自己開始起作用，不需要再改程式。
留著它而不是拿掉，是因為長度控制真正靠的是提示詞裡那句「400 字以內」，
而這個參數是第二道保險。一個被宣告、被刻意傳入、然後被默默丟棄的參數
比不存在更糟，但一個傳到底、只是供應商不理的參數是誠實的。
"""
return _llm_call(system or "", prompt, max_tokens=max_tokens)
# ── 長文 map-reduce 濃縮（TODOS 2026-07-13）─────────────────────────────────
# 卡片生成原本硬截 4000 字元，論文/長影片後半段直接丟失且不可逆。
# 超過門檻的內容改為：分段摘要（map）→ 合併（reduce），保留全文重點。
MAPREDUCE_THRESHOLD = int(os.getenv("XKB_MAPREDUCE_THRESHOLD", "4000"))
MAPREDUCE_CHUNK = int(os.getenv("XKB_MAPREDUCE_CHUNK", "6000"))
MAPREDUCE_MAX_CHUNKS = int(os.getenv("XKB_MAPREDUCE_MAX_CHUNKS", "12"))
_MAP_PROMPT = (
"以下是一份長文件的第 {i}/{n} 段。請用繁體中文濃縮這一段的關鍵資訊"
"（論點、數據、結論），400 字以內，保留專有名詞原文。只輸出濃縮內容：\n\n{chunk}"
)
def condense_long_content(content: str, verbose: bool = False) -> str:
"""長文 map-reduce：超過門檻的內容分段摘要後合併，取代硬截斷。
LLM 失敗時退回舊行為（截斷），絕不阻斷 ingest 主流程。"""
if len(content) <= MAPREDUCE_THRESHOLD:
return content
all_chunks = [content[i:i + MAPREDUCE_CHUNK]
for i in range(0, len(content), MAPREDUCE_CHUNK)]
chunks = all_chunks[:MAPREDUCE_MAX_CHUNKS]
# 超過上限的部分會被丟掉。原本丟得無聲無息，而產出的標頭還寫著「原文 N
# 字元」——一份 20 萬字的文件少掉三分之二，卡片上看不出來。這正是這支
# 函式要取代的那個行為，換一層又出現。
dropped_chars = sum(len(c) for c in all_chunks[MAPREDUCE_MAX_CHUNKS:])
if dropped_chars:
print(f" [map-reduce] 超出 {MAPREDUCE_MAX_CHUNKS} 段上限，"
f"最後 {dropped_chars} 字元未納入濃縮", file=sys.stderr, flush=True)
summaries: list[str] = []
failed = 0
for i, chunk in enumerate(chunks, 1):
try:
s = llm_call(_MAP_PROMPT.format(i=i, n=len(chunks), chunk=chunk),
max_tokens=600, system=None)
summaries.append(s.strip())
if verbose:
print(f" [map-reduce] chunk {i}/{len(chunks)} → {len(s)} chars")
except Exception as e:
# 一段失敗不該丟掉已經付過錢的其他段。原本直接退回截斷，
# 等於把十一次成功的呼叫連同結果一起扔掉。
failed += 1
summaries.append(f"（第 {i} 段濃縮失敗，內容未納入）")
if verbose:
print(f" [map-reduce] chunk {i} failed ({e})")
if not any(not x.startswith("（第 ") for x in summaries):
# 全部都失敗才退回截斷——那時候確實沒有東西可用。
return content[:MAPREDUCE_THRESHOLD]
notes = []
if dropped_chars:
notes.append(f"末尾 {dropped_chars} 字元超出分段上限未納入")
if failed:
notes.append(f"{failed}/{len(chunks)} 段濃縮失敗")
caveat = f"（{'；'.join(notes)}）" if notes else ""
head = content[:800]
reduced = "（以下為長文件分段濃縮，原文 %d 字元%s）\n\n%s\n\n--- 原文開頭 ---\n%s" % (
len(content), caveat, "\n\n".join(summaries), head)
return reduced
# ── Summary extraction ────────────────────────────────────────────────────────
def extract_summary(card: str) -> str:
"""Extract ZH + EN summary from section 7, with legacy fallback."""
# New format: ## 7. 雙語摘要 with ZH:/EN: lines
bilingual = re.search(
r"##\s+7\.\s*雙語摘要[^\n]*\n(.+?)(?=\n##|\Z)", card, re.DOTALL
)
if bilingual:
block = bilingual.group(1)
zh_m = re.search(r"^ZH:\s*(.+)$", block, re.MULTILINE)
en_m = re.search(r"^EN:\s*(.+)$", block, re.MULTILINE)
parts = [m.group(1).strip() for m in [zh_m, en_m] if m and m.group(1).strip()]
if parts:
return " | ".join(parts)
# Legacy
zh = re.search(r"##\s*📝 一句話摘要\s*\n+(.+?)(\n##|\Z)", card, re.DOTALL)
en = re.search(r"##\s*📝 English Summary\s*\n+(.+?)(\n##|\Z)", card, re.DOTALL)
parts = [x.group(1).strip() for x in [zh, en] if x]
if parts:
return " | ".join(parts)
lines = [l.strip() for l in card.splitlines()
if l.strip() and not l.startswith("#") and not l.startswith("---")]
return lines[0] if lines else ""
# ── Related context search ────────────────────────────────────────────────────
def find_related_context(content: str, existing_items: list[dict], top_k: int = 3) -> str:
"""Find related cards for context injection.
Uses gbrain hybrid search if available, falls back to keyword search against existing_items."""
if _GBRAIN_AVAILABLE:
try:
from xbrain_recall import xbrain_query as gbrain_query
query = content[:300].replace("\n", " ").strip()
results = gbrain_query(query, limit=top_k, no_expand=True)
if results:
lines = []
for r in results:
title = r.get("title", r.get("slug", ""))[:60]
chunk = r.get("chunk_text", "")[:100].replace("\n", " ")
lines.append(f"- **{title}**：{chunk}")
return "\n".join(lines)
except Exception:
pass
# Keyword fallback
# 斷詞只有一個定義：xkb_text。這裡原本各自帶一份只切 2-gram、
# 而且把整串中文吃成一個 token 的正則。
tokens = set(xkb_text.tokenize(content[:1000], xkb_text.STOPWORDS))
if not tokens:
return "（無相關既有卡片）"
scored = []
for item in existing_items:
combined = " ".join([
(item.get("title") or "").lower(),
(item.get("summary") or "").lower(),
" ".join(item.get("tags") or []).lower(),
])
score = sum(1 for t in tokens if t in combined)
if score > 0:
scored.append((item, score))
scored.sort(key=lambda x: x[1], reverse=True)
if not scored:
return "（無相關既有卡片）"
lines = []
for item, _ in scored[:top_k]:
lines.append(
f"- **{item.get('title', '(untitled)')}**："
f"{(item.get('summary') or '')[:80]}"
)
return "\n".join(lines)
def build_prompt(
content: str,
card_id: str,
source_type: str,
source_url: str,
category: str,
related_context: str = "（無相關既有卡片）",
) -> str:
"""Fill in the unified CARD_PROMPT template."""
return CARD_PROMPT.format(
source_type_label=source_label(source_type),
source_url=source_url,
category=category,
content=content,
card_id=card_id,
source_type=source_type,
related_section=(
f"相關既有卡片（供 Section 6 參考）：\n{related_context}\n"
if related_context else ""
),
related_cards_placeholder=related_context,
)