搭建個人知識問答應用 - Vector Retrieval Service for Milvus

方案說明

整體鏈路分為三段：在控制台建立知識庫並匯入資料、發布版本使內容可被檢索、通過 SDK 檢索並將結果交給大模型產生回答。其中資料匯入與檢索可通過 SDK 完成，版本發布需在控制台操作（當前未開放對應 OpenAPI）。

前提條件

已建立阿里雲帳號，且在支援知識庫的地區（華東1（杭州）、華北2（北京）、華北3（張家口）、華南1（深圳））下操作。
已為調用帳號完成 RAM 授權：使用主帳號登入RAM 控制台，建立 RAM 使用者並勾選使用永久 AccessKey 訪問，然後為該使用者授予系統策略 AliyunMilvusFullAccess。

說明
建立帶 AccessKey 的 RAM 使用者時會觸發安全驗證（MFA 碼、手機驗證碼或掃臉）。AccessKey Secret 僅在建立後顯示一次，請通過頁面的儲存結果下載留存。如需更嚴格的最小許可權，可另行建立僅包含 milvusknowledgebase:* 的自訂策略並替換上述系統策略。
已準備一個支援 OpenAI chat/completions 協議的大模型 API，例如大模型服務平台百鍊的 OpenAI 相容地址與 API Key。
本地已安裝 Python 3.8 或以上版本。

重要

AccessKey 與大模型 API Key 只儲存在本機環境變數中，不要寫入文檔、聊天記錄、截圖或代碼倉庫。

步驟一：建立知識庫

登入向量檢索服務 Milvus 版控制台，在頂部地區下拉框中切換到目標地區，然後在左側導覽列單擊知識庫服務。

在知識庫頁面單擊建立知識庫，完成以下配置。

配置項	說明
名稱	2~64 個字元，不支援中文，同一租戶內唯一。
資料類型	可選全模態知識庫或結構化知識庫（按表格行產生切片，僅支援 `xls`、`xlsx`、`csv`），建立後不可更改。本文選擇全模態知識庫，支援 `mp4`、`mp3`、`png`、`pdf`、`ppt`、`txt`、`markdown`、`docs`、`xlsx`、`csv`、`jsonl`、`faq`。
向量模型	可選內建模型或私人/外部模型，建立後不可更改。本文使用內建模型，預設 `text-embedding-v3`，向量維度自動顯示 1024；選擇私人/外部模型時向量維度顯示為 `-` 且不可編輯。
規格配置	必選，當前最小規格為 4CU（4 vCPU / 16 GiB），建立後可在詳情頁調整。
網路設定	必選，指定專用網路與交換器，也可現場建立。

說明

切片策略不在建立頁配置。進入知識庫詳情頁的處理策略地區可查看系統提供的預設策略（智能切分，最大長度 512 字元），也可單擊建立策略自訂；通過 SDK 註冊資料時可用 strategy_id 指定策略。

單擊建立知識庫，等待知識庫狀態由建立中變為運行中。
在知識庫列表中記錄知識庫 ID（形如 kd-803ae9b10cc31），後續 SDK 調用需要該 ID。

步驟二：準備資料

建議先用少量 Markdown、TXT、PDF 或 Word 文檔驗證效果。檔案名稱應能表達文檔主題，本文應包含完整標題和上下文。

如果沒有合適的文檔，可以使用公開中文法律案例資料 LeCaRDv2。法律案例包含具體案號、案情和裁判結果，比通用常識更容易驗證答案是否真正來自知識庫。該資料僅用於檢索技術示範，不構成法律意見。

步驟三：配置本地環境

建立虛擬環境並安裝依賴。

python3 -m venv .venv
source .venv/bin/activate
pip install alibabacloud_milvusknowledgebase20260604 requests streamlit

配置運行參數。其中 KB_ID 為步驟一記錄的知識庫 ID，KB_REGION 為知識庫所在地區。

export ALIBABA_CLOUD_ACCESS_KEY_ID="YOUR_ACCESS_KEY_ID"
export ALIBABA_CLOUD_ACCESS_KEY_SECRET="YOUR_ACCESS_KEY_SECRET"
export KB_REGION="cn-hangzhou"
export KB_ID="kd-xxxxxxxxxxxxx"
export KB_VERSION="LATEST_PUBLISHED"

export LLM_BASE_URL="https://your-openai-compatible-api.example.com/v1"
export LLM_API_KEY="YOUR_LLM_TOKEN"
export LLM_MODEL="YOUR_MODEL_NAME"

步驟四：編寫範例程式碼

將下面內容儲存為 kb_demo.py，它包含用戶端初始化、上傳資料、檢索和大模型問答四部分。

from __future__ import annotations

import os
from pathlib import Path

import requests
from alibabacloud_milvusknowledgebase20260604.client import Client
from alibabacloud_milvusknowledgebase20260604 import models as milvus_kb_models
from alibabacloud_tea_openapi import models as open_api_models

REGION = os.getenv("KB_REGION", "cn-hangzhou")
KB_ID = os.environ["KB_ID"]
KB_VERSION = os.getenv("KB_VERSION", "LATEST_PUBLISHED")
ENDPOINT = f"milvusknowledgebase.{REGION}.aliyuncs.com"

client = Client(open_api_models.Config(
    access_key_id=os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"],
    access_key_secret=os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"],
    endpoint=ENDPOINT,
    region_id=REGION,
    connect_timeout=10_000,
    read_timeout=60_000,
))


def upload_document(file_path: str) -> dict:
    path = Path(file_path).resolve()
    size = path.stat().st_size

    presigned = client.get_knowledge_base_pre_signed_url(
        KB_ID,
        milvus_kb_models.GetKnowledgeBasePreSignedUrlRequest(
            knowledge_base_id=KB_ID,
            documents=[
                milvus_kb_models.GetKnowledgeBasePreSignedUrlRequestDocuments(
                    path=path.name, name=path.name, size=size,
                )
            ],
            expires_in=3600,
        ),
    )
    upload_url = presigned.body.data.pre_signed_urls[0]

    # 預簽名 URL 按空 Content-Type 簽發，PUT 請求不要攜帶 Content-Type
    with path.open("rb") as source:
        response = requests.put(upload_url, data=source, timeout=120)
        response.raise_for_status()

    added = client.add_documents(
        KB_ID,
        milvus_kb_models.AddDocumentsRequest(
            knowledge_base_id=KB_ID,
            import_type="LOCAL_UPLOAD",
            documents=[
                milvus_kb_models.AddDocumentsRequestDocuments(
                    path=path.name, name=path.name, size=size,
                )
            ],
            dedup=milvus_kb_models.AddDocumentsRequestDedup(
                doc_name_dedup=True, content_dedup=False,
            ),
        ),
    )
    return added.body


def search_knowledge_base(query: str, page_size: int = 6):
    resp = client.search_knowledge_base(
        KB_ID,
        milvus_kb_models.SearchKnowledgeBaseRequest(
            query=query,
            version=KB_VERSION,
            page_number=1,
            page_size=page_size,
            retrieval_config=milvus_kb_models.SearchKnowledgeBaseRequestRetrievalConfig(
                candidate_count=48,
                min_score=0,
                semantic_weight=0.5,
                enable_query_expansion=False,
            ),
        ),
    )
    return resp.body


def chat(messages: list[dict[str, str]]) -> str:
    base_url = os.environ["LLM_BASE_URL"].rstrip("/")
    response = requests.post(
        f"{base_url}/chat/completions",
        headers={"Authorization": f"Bearer {os.environ['LLM_API_KEY']}"},
        json={
            "model": os.environ["LLM_MODEL"],
            "messages": messages,
            "temperature": 0.1,
        },
        timeout=60,
    )
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"].strip()


def ask(question: str) -> tuple[str, object]:
    search_query = chat([
        {
            "role": "system",
            "content": "把使用者問題改寫為自包含、適合知識庫檢索的中文 Query，只輸出 Query。",
        },
        {"role": "user", "content": question},
    ])
    search_result = search_knowledge_base(search_query)
    results = search_result.results or []

    context = "\n\n".join(
        f"[來源 {i}] {item.document_name or ''}\n{item.content}"
        for i, item in enumerate(results, start=1)
    )
    answer = chat([
        {
            "role": "system",
            "content": "只根據給定資料回答；引用資料時標註[來源 N]；資料不足時明確說明。",
        },
        {"role": "user", "content": f"問題：{question}\n\n資料：\n{context}"},
    ])
    return answer, search_result

步驟五：上傳資料

上傳分為三步：擷取預簽名地址、將檔案寫入 OSS、註冊資料。upload_document() 已封裝完整過程。

from kb_demo import upload_document

result = upload_document("./documents/example.md")
print(result.request_id)

批量上傳時，建議先從 10 篇一組開始。

from pathlib import Path
from kb_demo import upload_document

for path in sorted(Path("./documents").glob("*.md")):
    upload_document(str(path))
    print("已提交：", path.name)

重要

預簽名地址按空 Content-Type 簽發，PUT 請求不能攜帶 Content-Type 要求標頭，否則 OSS 返回 403 SignatureDoesNotMatch。若 PUT 因網路原因失敗，可直接重試。

重要

註冊資料時，path 與 name 均只填檔案名稱，不要填預簽名地址中的物件路徑。填入 direct_upload/… 這類內部路徑會返回 400 Path must not use an internal storage prefix.；完全不填 path 會返回 400 Path can't be empty.

註冊成功的響應中，data.documents 可能是空數組，這不代表註冊失敗。請以知識庫詳情頁的資料數量、切片數量或後續檢索結果為準。

只把檔案 PUT 到 OSS 並不會讓資料進入知識庫，必須調用 AddDocuments 完成註冊。註冊成功後響應中 run 為 RUNNING，表示正在解析與切片。

上傳完成後在控制台的資料管理頁查看資料狀態與切片數，狀態為處理完成後再發布版本；單擊查看切片可預覽切分結果。

步驟六：發布版本

資料匯入後處於未發布狀態，需發布版本後才能被檢索。當前該操作僅支援在控制台完成。

進入知識庫詳情頁，單擊版本管理頁簽，確認頁面提示存在待發布變更後單擊發布版本。
在確認變更步驟核對變更日誌，單擊下一步。
在填寫說明步驟輸入發布說明（不超過 200 字元），單擊發布，等待提示版本發布成功。
在版本記錄中查看版本號碼，首次發布為 v1，狀態為發行。

發布完成後更新本地環境變數。

export KB_VERSION="v1"

步驟七：檢索資料

from kb_demo import search_knowledge_base

result = search_knowledge_base("合約解除需要滿足哪些條件？")

for item in result.results or []:
    print(item.document_name, item.score)
    print(item.content[:300])

version 建議使用明確版本號碼（如 v1、v2），也可以使用 LATEST_PUBLISHED 檢索最新發行版本。請勿使用 DRAFT 對外提供穩定問答服務。

說明

樣本中 min_score=0 會把相關度較低的切片一併返回。生產環境建議結合實際效果調高 min_score，或減小 page_size，以避免不相關內容進入大模型上下文。

您也可以在控制台的檢索驗證頁調整 TopK、語義檢索權重、重排模型與篩選範圍，快速對比檢索效果。

步驟八：接入大模型問答

ask() 會先讓大模型把問題改寫為適合檢索的 Query，再調用 SearchKnowledgeBase，最後讓大模型僅依據返回片段匯總答案。

from kb_demo import ask

answer, search_result = ask("這批文檔對合約解除是怎麼規定的？")
print(answer)
print("Request ID:", search_result.request_id)

步驟九：構建網頁應用

將下面內容儲存為 app.py。

import streamlit as st
from kb_demo import ask

st.set_page_config(page_title="知識庫問答")
st.title("知識庫問答")

question = st.chat_input("請輸入問題")
if question:
    with st.chat_message("user"):
        st.write(question)
    with st.chat_message("assistant"):
        with st.spinner("正在檢索知識庫…"):
            answer, result = ask(question)
        st.write(answer)
        with st.expander("查看檢索來源"):
            for item in result.results or []:
                st.markdown(f"**{item.document_name or '未命名文檔'}**")
                st.caption(f"score: {item.score}  ·  Request ID: {result.request_id}")
                st.write(item.content)

啟動應用。
```
streamlit run app.py
```

瀏覽器會自動開啟本地知識庫問答頁面。

常見問題

現象	原因與處理
調用介面返回 `403 Permission denied by RAM authentication`	調用帳號未獲得知識庫許可權。為該 RAM 使用者授予 `AliyunMilvusFullAccess`，授權生效後重試。
PUT 上傳返回 `403 SignatureDoesNotMatch`	請求攜帶了 `Content-Type` 要求標頭。預簽名地址按空 Content-Type 簽發，去掉該要求標頭後重試。
檢索返回 `404 Knowledge base version ... does not exist`	知識庫尚未發布版本，或 `KB_VERSION` 與實際版本號碼不一致。先在控制台發布版本，再將 `KB_VERSION` 設為版本號碼或 `LATEST_PUBLISHED`。
PUT 上傳偶發串連失敗	OSS 串連抖動，直接重試該檔案即可。
控制台左側沒有知識庫服務	當前地區不支援，或菜單尚未載入完成。切換到支援的地區後重新查看。

上線前檢查

使用獨立、最小許可權、可輪換的 RAM 使用者，不要長期使用主帳號 AccessKey。
只上傳有權處理的資料。
回答必須能夠在展開的來源片段中找到依據。
對外部署頁面時增加身份認證、HTTPS 和訪問日誌脫敏。
介面開放範圍和許可權要求以產品文檔與控制台實際為準。