🏠🤖 地端 AI 進階應用：使用 Ollama 與 Python 建立私有技術文件助手

把「公司內部 SOP、維運筆記、Runbook、故障排查手冊」丟給雲端 AI 來問答，最大的痛點通常不是效果，而是資料外流風險與合規問題。

這篇要做的是一個「地端私有化」的技術文件助手：

• LLM 在地端跑（Ollama）
• 文件向量化（embeddings）後存進 SQLite（不用額外架向量資料庫也能跑）
• 查詢時先檢索相關片段，再交給模型整理（RAG）
• 答案附上「引用片段來源」，方便你回頭核對

---

📌 目錄

• 1. 整體架構（RAG）
• 2. 安裝 Ollama 與模型準備
• 3. Ollama API 快速測試
• 4. Python 專案初始化
• 5. 實作：私有技術文件助手（SQLite 向量檢索 + 問答）
• 6. 實務調校：準確度、安全性、效能
• 7. 維運建議：更新、備份、權限隔離
• FAQ
• 🔗 延伸閱讀

---

1. 整體架構（RAG）

RAG（Retrieval-Augmented Generation）的重點：先找資料，再叫模型回答，避免模型憑印象亂補。

使用者問題
   │
   ├─(1) 向量化（embedding）
   │
   ├─(2) SQLite 取 Top-K 相似片段（cosine similarity）
   │
   └─(3) 把「片段 + 問題」丟給 LLM 生成答案
           └─(4) 答案附上來源（檔名/段落）

你會同時跑兩種模型：

• 聊天/生成模型：負責「整理成可讀答案」
• Embedding 模型：負責把文字變成向量，拿來做相似度檢索

---

2. 安裝 Ollama 與模型準備

2.1 Linux 安裝 Ollama

curl -fsSL https://ollama.com/install.sh | sh

若你希望它以 systemd 服務形式長駐（伺服器環境很常見），可以用官方文件的方式建立並啟用服務（或直接使用安裝後附帶的服務）。需要調整環境變數時，可用 systemctl edit 做 override。

2.2 下載模型（建議一個聊天模型 + 一個 embeddings 模型）

# 聊天模型（擇一）
ollama pull gemma3
# 或
ollama pull llama3.2

# Embeddings 模型（擇一）
ollama pull embeddinggemma
# 或
ollama pull all-minilm

小提醒：embedding 模型通常比聊天模型輕量；你也可以依硬體資源調整選擇（CPU-only 也能跑，只是速度會比較慢）。

---

3. Ollama API 快速測試

Ollama 的 API 預設在本機提供服務（base URL：http://localhost:11434/api），你可以用 curl 先驗證。
（REST API 預設是串流輸出，想一次拿完整結果可以加 "stream": false）

3.1 文字生成（/api/generate）

curl http://localhost:11434/api/generate -d '{
  "model": "gemma3",
  "prompt": "請用三點列出：RAG 的核心價值是什麼？",
  "stream": false
}'

3.2 對話（/api/chat）

curl http://localhost:11434/api/chat -d '{
  "model": "gemma3",
  "messages": [
    { "role": "system", "content": "你是嚴謹的技術文件助手，回答要附上引用來源。" },
    { "role": "user", "content": "什麼情境適合用 Active Check？" }
  ],
  "stream": false
}'

3.3 Embeddings（/api/embed）

curl http://localhost:11434/api/embed -d '{
  "model": "embeddinggemma",
  "input": "Zabbix Proxy 可以緩衝資料並降低 Server 壓力",
  "truncate": true
}'

---

4. Python 專案初始化

這裡用最少依賴：requests + sqlite3（標準庫）就能完成。

mkdir private-docs-assistant && cd private-docs-assistant
python -m venv .venv
source .venv/bin/activate

pip install -U pip
pip install requests

建議專案結構：

private-docs-assistant/
  docs/                  # 你的私有文件（md/txt/html 都可先放這）
  assistant.py           # 主程式（含 ingest + ask）
  kb.sqlite3             # SQLite 知識庫（自動產生）

---

5. 實作：私有技術文件助手（SQLite 向量檢索 + 問答）

這支腳本提供兩個指令：

• python assistant.py ingest ./docs：把資料切片 + embeddings + 存 SQLite
• python assistant.py ask "你的問題"：檢索 Top-K 片段 + LLM 生成答案（附來源）

#!/usr/bin/env python3
# assistant.py
import argparse
import json
import math
import os
import re
import sqlite3
from dataclasses import dataclass
from pathlib import Path
from typing import List, Tuple

import requests

OLLAMA_BASE = os.environ.get("OLLAMA_BASE", "http://localhost:11434/api")
CHAT_MODEL = os.environ.get("OLLAMA_CHAT_MODEL", "gemma3")
EMBED_MODEL = os.environ.get("OLLAMA_EMBED_MODEL", "embeddinggemma")
DB_PATH = os.environ.get("KB_DB", "kb.sqlite3")

# --- 文字處理：簡單切片（可依你文件型態再加強） ---
def read_text_file(path: Path) -> str:
    text = path.read_text(encoding="utf-8", errors="ignore")
    # 粗略清理：多空白縮減
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()

def chunk_text(text: str, chunk_size: int = 900, overlap: int = 150) -> List[str]:
    if not text:
        return []
    chunks = []
    i = 0
    n = len(text)
    while i < n:
        j = min(n, i + chunk_size)
        chunk = text[i:j].strip()
        if chunk:
            chunks.append(chunk)
        i = j - overlap
        if i < 0:
            i = 0
        if j == n:
            break
    return chunks

# --- Ollama API ---
def ollama_embed(texts: List[str]) -> List[List[float]]:
    # /api/embed 支援 input 為字串或字串陣列
    r = requests.post(
        f"{OLLAMA_BASE}/embed",
        json={"model": EMBED_MODEL, "input": texts, "truncate": True},
        timeout=300,
    )
    r.raise_for_status()
    data = r.json()
    return data["embeddings"]

def ollama_chat(system: str, user: str) -> str:
    payload = {
        "model": CHAT_MODEL,
        "messages": [
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
        "stream": False,
    }
    r = requests.post(f"{OLLAMA_BASE}/chat", json=payload, timeout=300)
    r.raise_for_status()
    data = r.json()
    return data["message"]["content"]

# --- 向量運算 ---
def cosine_sim(a: List[float], b: List[float]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(y * y for y in b))
    if na == 0 or nb == 0:
        return 0.0
    return dot / (na * nb)

# --- SQLite KB ---
def init_db(conn: sqlite3.Connection) -> None:
    conn.execute("""
    CREATE TABLE IF NOT EXISTS chunks (
      id INTEGER PRIMARY KEY AUTOINCREMENT,
      path TEXT NOT NULL,
      chunk_index INTEGER NOT NULL,
      content TEXT NOT NULL,
      embedding TEXT NOT NULL
    )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_chunks_path ON chunks(path)")
    conn.commit()

def save_chunks(conn: sqlite3.Connection, path: str, chunks: List[str], embeds: List[List[float]]) -> None:
    conn.execute("DELETE FROM chunks WHERE path = ?", (path,))
    rows = [
        (path, i, chunks[i], json.dumps(embeds[i], ensure_ascii=False))
        for i in range(len(chunks))
    ]
    conn.executemany(
        "INSERT INTO chunks(path, chunk_index, content, embedding) VALUES (?, ?, ?, ?)",
        rows,
    )
    conn.commit()

def load_all(conn: sqlite3.Connection) -> List[Tuple[str, int, str, List[float]]]:
    cur = conn.execute("SELECT path, chunk_index, content, embedding FROM chunks")
    out = []
    for path, idx, content, emb_json in cur.fetchall():
        out.append((path, idx, content, json.loads(emb_json)))
    return out

@dataclass
class Hit:
    score: float
    path: str
    chunk_index: int
    content: str

def search(conn: sqlite3.Connection, query: str, top_k: int = 5) -> List[Hit]:
    q_emb = ollama_embed([query])[0]
    items = load_all(conn)

    scored = []
    for path, idx, content, emb in items:
        s = cosine_sim(q_emb, emb)
        scored.append(Hit(score=s, path=path, chunk_index=idx, content=content))

    scored.sort(key=lambda x: x.score, reverse=True)
    return scored[:top_k]

# --- CLI ---
def cmd_ingest(args) -> None:
    docs_dir = Path(args.dir).resolve()
    assert docs_dir.exists(), f"docs dir not found: {docs_dir}"

    conn = sqlite3.connect(DB_PATH)
    init_db(conn)

    files = []
    for ext in ("*.md", "*.txt", "*.log", "*.html"):
        files.extend(docs_dir.rglob(ext))

    if not files:
        print("找不到可匯入的檔案（md/txt/log/html）。")
        return

    for fp in sorted(files):
        text = read_text_file(fp)
        chunks = chunk_text(text, chunk_size=args.chunk_size, overlap=args.overlap)

        if not chunks:
            continue

        embeds = ollama_embed(chunks)
        rel_path = str(fp.relative_to(docs_dir))
        save_chunks(conn, rel_path, chunks, embeds)
        print(f"[OK] {rel_path}  chunks={len(chunks)}")

    print("完成：知識庫已更新。")

def cmd_ask(args) -> None:
    conn = sqlite3.connect(DB_PATH)
    init_db(conn)

    hits = search(conn, args.question, top_k=args.top_k)

    if not hits:
        print("找不到相關片段，請確認已執行 ingest，或調整問題描述。")
        return

    context_lines = []
    for i, h in enumerate(hits, 1):
        context_lines.append(
            f"[{i}] {h.path}#{h.chunk_index} (score={h.score:.3f})\n{h.content}\n"
        )
    context = "\n---\n".join(context_lines)

    system = (
        "你是『私有技術文件助手』，只能根據使用者提供的【引用片段】回答。"
        "若片段不足以得出結論，請直接說『文件片段不足，建議補充：...』，不要自行編造。"
        "回答格式要求：\n"
        "1) 先給結論（條列）\n"
        "2) 再給操作步驟（若適用）\n"
        "3) 最後列出你引用的片段編號（例如：引用：[1][3]）\n"
    )

    user = (
        f"【問題】\n{args.question}\n\n"
        f"【引用片段】\n{context}\n\n"
        "請開始回答。"
    )

    ans = ollama_chat(system, user)
    print(ans)

def main():
    p = argparse.ArgumentParser(description="Private Tech Docs Assistant (Ollama + SQLite)")
    sub = p.add_subparsers(dest="cmd", required=True)

    p_ing = sub.add_parser("ingest", help="ingest docs into sqlite kb")
    p_ing.add_argument("dir", help="docs folder, e.g. ./docs")
    p_ing.add_argument("--chunk-size", type=int, default=900)
    p_ing.add_argument("--overlap", type=int, default=150)
    p_ing.set_defaults(func=cmd_ingest)

    p_ask = sub.add_parser("ask", help="ask a question with retrieval")
    p_ask.add_argument("question", help="your question")
    p_ask.add_argument("--top-k", type=int, default=5)
    p_ask.set_defaults(func=cmd_ask)

    args = p.parse_args()
    args.func(args)

if __name__ == "__main__":
    main()

5.1 使用方式

# 1) 先匯入你的文件
python assistant.py ingest ./docs

# 2) 開始問問題
python assistant.py ask "請整理這份 Runbook 裡，服務無法啟動時的排查順序"

---

6. 實務調校：準確度、安全性、效能

6.1 準確度：Top-K、切片大小、以及「回答規則」

• Top-K：5 通常夠用；若文件很大、答案容易漏段落，可調到 8~12。
• chunk size：900 字元左右是常見折衷；太小會碎片化、太大會混雜不同主題。
• 回答規則（system prompt）：一定要明確要求「只能根據引用片段」，不足就說不足。

6.2 安全性：不要隨便把 Ollama API 直接對外

Ollama 預設只在本機提供 API（本篇也以 localhost 為前提）。如果你真的需要跨機器呼叫，建議優先用：

• SSH Tunnel（最省事）
• 或反向代理加上存取控管（例如 Basic Auth / IP allowlist / mTLS）

若你是以 systemd 方式跑 Ollama，也可以用 service override 設定環境變數（例如 OLLAMA_HOST）。設定環境變數的方式可參考官方 FAQ 的範例。

6.3 效能：模型常駐與快取

• 本篇用 /api/chat 與 /api/embed，都可透過 keep_alive 讓模型常駐，減少反覆載入成本。
• 文件匯入（ingest）是離線作業，建議排程晚上跑；白天只做 ask。

---

7. 維運建議：更新、備份、權限隔離

7.1 更新 Ollama

curl -fsSL https://ollama.com/install.sh | sh

7.2 備份知識庫

cp kb.sqlite3 kb.sqlite3.bak

7.3 權限隔離

• 把 docs/ 與 kb.sqlite3 放在只有管理者可讀的路徑。
• 若是多人使用，建議把 Assistant 包成內網服務，再用帳號權限做控管（避免每個人直接讀到所有文件）。

---

FAQ

Q1：為什麼 embeddings 不直接用聊天模型做？

因為 embeddings 是「為檢索而生」：向量品質更穩、速度更快、成本更低。聊天模型主要負責整理語句與輸出答案。

Q2：SQLite 真的夠用嗎？

小到中型（幾千到幾萬個 chunks）通常沒問題。若你上升到十萬級以上，或多人併發查詢，就建議換成專用向量資料庫或至少把檢索層做服務化。

Q3：怎麼讓答案更「像技術文件」而不是聊天？

把 system prompt 寫得更硬一點：要求「條列、步驟、指令、風險、回復方式」，並強制附上引用片段編號。

---

💬 留言聊聊你的文件類型

你的文件主要是 SOP / Runbook / Markdown 筆記 / Wiki / HTML 手冊？
你希望回答「更嚴謹」還是「更像協作同事」？我可以依你的內容型態，幫你把切片規則、Top-K 與提示詞調成更準。

🔗 延伸閱讀

• AI 工具與應用整理
• ComfyUI / Stable Diffusion 安裝與入門
• 🐧 Linux systemd 深入解析與啟動流程管理