Ollama 企業知識庫教學：用 RAG 查 SOP、技術文件、產品規格與客服 FAQ

				
					import os
import re
import uuid
import requests

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct


QDRANT_URL = "http://localhost:6333"
COLLECTION_NAME = "taki_sop_knowledge"

OLLAMA_EMBED_URL = "http://localhost:11434/api/embed"
EMBED_MODEL = "nomic-embed-text"

DOCS_DIR = "./docs"


def embed_text(text: str) -> list[float]:
    response = requests.post(
        OLLAMA_EMBED_URL,
        json={
            "model": EMBED_MODEL,
            "input": text
        },
        timeout=120
    )

    response.raise_for_status()
    data = response.json()

    if "embeddings" in data:
        return data["embeddings"][0]

    if "embedding" in data:
        return data["embedding"]

    raise RuntimeError(f"無法取得 embedding：{data}")


def split_markdown_by_heading(text: str, max_chars: int = 1200) -> list[str]:
    parts = re.split(r"\n(?=#{1,6}\s)", text)
    chunks = []

    for part in parts:
        part = part.strip()

        if len(part) < 50:
            continue

        if len(part) <= max_chars:
            chunks.append(part)
            continue

        paragraphs = part.split("\n\n")
        buffer = ""

        for paragraph in paragraphs:
            if len(buffer) + len(paragraph) + 2 > max_chars:
                if len(buffer.strip()) >= 50:
                    chunks.append(buffer.strip())
                buffer = paragraph
            else:
                buffer = f"{buffer}\n\n{paragraph}".strip()

        if len(buffer.strip()) >= 50:
            chunks.append(buffer.strip())

    return chunks


def load_documents() -> list[dict]:
    documents = []

    for filename in os.listdir(DOCS_DIR):
        if not filename.endswith((".md", ".txt")):
            continue

        path = os.path.join(DOCS_DIR, filename)

        with open(path, "r", encoding="utf-8") as f:
            content = f.read()

        chunks = split_markdown_by_heading(content)

        for index, chunk in enumerate(chunks):
            documents.append({
                "id": str(uuid.uuid4()),
                "text": chunk,
                "metadata": {
                    "source": filename,
                    "chunk_index": index,
                    "category": "SOP",
                    "department": "Engineering",
                    "permission": "internal",
                    "version": "2026-05"
                }
            })

    return documents


def main():
    client = QdrantClient(url=QDRANT_URL)

    docs = load_documents()

    if not docs:
        raise RuntimeError("沒有找到可匯入的文件，請確認 ./docs 目錄中有 .md 或 .txt 檔案")

    sample_vector = embed_text(docs[0]["text"])
    vector_size = len(sample_vector)

    if client.collection_exists(COLLECTION_NAME):
        client.delete_collection(COLLECTION_NAME)

    client.create_collection(
        collection_name=COLLECTION_NAME,
        vectors_config=VectorParams(
            size=vector_size,
            distance=Distance.COSINE
        )
    )

    points = []

    for doc in docs:
        vector = embed_text(doc["text"])

        points.append(
            PointStruct(
                id=doc["id"],
                vector=vector,
                payload={
                    "text": doc["text"],
                    **doc["metadata"]
                }
            )
        )

    client.upsert(
        collection_name=COLLECTION_NAME,
        points=points,
        wait=True
    )

    print(f"匯入完成：{len(points)} 個 chunks")
    print(f"Collection：{COLLECTION_NAME}")
    print(f"Embedding model：{EMBED_MODEL}")
    print(f"Vector size：{vector_size}")


if __name__ == "__main__":
    main()