Vector Retrieval Service for Milvus：通過阿里雲Milvus知識庫搭建智能客服問答應用

方案說明

整體鏈路為：在控制台定義標籤並建立知識庫 → 按標籤大量匯入客服資料 → 發布版本 → 通過 SDK 檢索（可按標籤過濾）→ 交由大模型產生帶來源標註的回答 → 用 Flask 提供問答頁面。

本文側重客服情境特有的三件事：標籤（metadata）過濾、重排模型與答案可追溯。知識庫的基礎鏈路（預簽名上傳、註冊資料、發布版本）與最簡問答實現，請參見搭建個人知識問答應用。

建議先準備 5~20 篇 PDF、DOCX、Markdown 或 TXT 資料，整體耗時約 20~30 分鐘（資料解析耗時取決於數量與大小）。

前提條件

• 已在支援知識庫的地區（華東1（杭州）、華北2（北京）、華北3（張家口）、華南1（深圳））建立知識庫，並記錄知識庫 ID（形如 kd-803ae9b10cc31）。

• 已使用主帳號建立 RAM 使用者、勾選使用永久 AccessKey 訪問，並為其授予系統策略 AliyunMilvusFullAccess。

• 已準備一個支援 OpenAI chat/completions 協議的大模型地址與 API Key，例如大模型服務平台百鍊的 OpenAI 相容地址。

• 本地已安裝 Python 3.8 或以上版本。

步驟一：定義標籤

標籤用於給資料打上業務維度（如資料類型、產品線、生效日期），匯入時寫入知識庫，檢索時用於過濾，是客服情境區分「政策」「手冊」「FAQ」的關鍵。

1. 登入向量檢索服務 Milvus 版控制台，進入目標知識庫的詳情頁。

2. 在基本資料下方找到標籤，單擊管理。

3. 在標籤管理對話方塊中，填寫標籤名、選擇欄位類型，單擊添加。本文使用以下三個標籤，均為 string 類型：

   標籤名	說明	樣本值
   docType	資料類型	policy、manual、faq
   productLine	適用產品線	all、phone
   effectiveDate	生效日期	2026-01-01

4. 三個標籤添加完成後單擊完成，詳情頁的標籤會顯示為「3 個標籤」。

步驟二：準備資料與匯入清單

1. 將客服資料放入本地 documents/ 目錄，例如：

   kb-demo/
   ├── documents/
   │   ├── shipping-policy.md
   │   ├── return-policy.md
   │   ├── warranty-policy.md
   │   ├── phone-manual.md
   │   └── invoice-faq.md

2. 建立 documents.jsonl，每行描述一篇資料及其標籤。標籤名必須與步驟一定義的一致。

   {"path": "documents/shipping-policy.md", "metadata": {"docType": "policy", "productLine": "all", "effectiveDate": "2026-01-01"}}
   {"path": "documents/return-policy.md", "metadata": {"docType": "policy", "productLine": "all", "effectiveDate": "2026-01-01"}}
   {"path": "documents/warranty-policy.md", "metadata": {"docType": "policy", "productLine": "phone", "effectiveDate": "2026-03-01"}}
   {"path": "documents/phone-manual.md", "metadata": {"docType": "manual", "productLine": "phone", "effectiveDate": "2026-03-01"}}
   {"path": "documents/invoice-faq.md", "metadata": {"docType": "faq", "productLine": "all", "effectiveDate": "2026-02-01"}}

步驟三：配置運行參數

1. 建立工程目錄並安裝依賴。

   mkdir -p kb-demo/documents kb-demo/templates
   cd kb-demo
   python3 -m venv .venv
   source .venv/bin/activate
   pip install Flask==3.1.1 requests==2.32.4 alibabacloud-milvusknowledgebase20260604==1.0.0

2. 建立 config.json，填入自己的憑證與知識庫資訊。

   {
     "aliyun": {
       "access_key_id": "YOUR_ACCESS_KEY_ID",
       "access_key_secret": "YOUR_ACCESS_KEY_SECRET",
       "region_id": "cn-hangzhou",
       "knowledge_base_id": "kd-xxxxxxxxxxxxx",
       "knowledge_base_version": "LATEST_PUBLISHED"
     },
     "llm": {
       "enabled": true,
       "base_url": "https://YOUR_OPENAI_COMPATIBLE_ENDPOINT/v1",
       "api_key": "YOUR_LLM_API_KEY",
       "model": "YOUR_MODEL_NAME"
     },
     "upload": {
       "default_meta_fields": {}
     },
     "retrieval": {
       "page_size": 6,
       "candidate_count": 48,
       "min_score": 0.35,
       "semantic_weight": 0.7,
       "enable_query_expansion": true,
       "rerank_model_name": "qwen3-rerank",
       "tag_filter": {
         "relation": "and",
         "conditions": []
       }
     },
     "scenario": {
       "title": "智能客服知識庫問答",
       "system_prompt": "你是企業客服助手。只依據檢索資料回答，使用清晰、友好的客服口吻；關鍵結論標註[來源N]。如果檢索資料沒有直接覆蓋使用者的問題，必須回答“現有資料中沒有明確說明”，不得根據部分相關的資料推斷結論，也不得補充資料以外的連絡方式或辦理渠道。",
       "image_enabled": false,
       "sample_questions": [
         "商品通常在付款後多久發貨？",
         "超過保修期後還能維修嗎？",
         "申請退貨需要滿足哪些條件？"
       ]
     }
   }

本情境檢索參數的取值考慮：

• min_score=0.35：減少低相關內容進入回答。

• semantic_weight=0.7：偏語義檢索，覆蓋口語化的客服問法。

• rerank_model_name：對候選切片二次排序，提升首條準確率。

• tag_filter.conditions：留空表示不過濾；按標籤過濾的寫法見步驟六。

步驟四：編寫知識庫用戶端

將下面內容儲存為 kb_client.py，封裝帶標籤的上傳與帶過濾的檢索。

"""Milvus 知識庫 OpenAPI SDK：帶標籤的本地上傳與發行版本檢索。"""

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Sequence

import requests
from alibabacloud_milvusknowledgebase20260604 import models as models
from alibabacloud_milvusknowledgebase20260604.client import Client
from alibabacloud_tea_openapi import models as openapi_models


@dataclass(frozen=True)
class LocalDocument:
    path: Path
    object_path: str

    @classmethod
    def from_path(cls, value: str | Path) -> "LocalDocument":
        path = Path(value).expanduser().resolve()
        if not path.is_file():
            raise FileNotFoundError(path)
        return cls(path=path, object_path=path.name)


@dataclass(frozen=True)
class TagCondition:
    field: str
    op: str
    value: Any


@dataclass(frozen=True)
class SearchOptions:
    version: str = "LATEST_PUBLISHED"
    page_size: int = 6
    candidate_count: int = 48
    min_score: float = 0.0
    semantic_weight: float = 0.5
    enable_query_expansion: bool = True
    rerank_model_name: str | None = None
    tag_relation: str = "and"
    tag_conditions: tuple[TagCondition, ...] = ()


class KnowledgeBaseClient:
    def __init__(
        self,
        access_key_id: str,
        access_key_secret: str,
        region_id: str,
    ) -> None:
        endpoint = f"milvusknowledgebase.{region_id}.aliyuncs.com"
        self.client = Client(
            openapi_models.Config(
                access_key_id=access_key_id,
                access_key_secret=access_key_secret,
                region_id=region_id,
                endpoint=endpoint,
                connect_timeout=10_000,
                read_timeout=60_000,
            )
        )
        self.http = requests.Session()

    @staticmethod
    def _check(body: Any, action: str) -> None:
        # 成功響應中 code 欄位不返回（值為 None），因此只把明確的非零 code 視為失敗。
        if body is None:
            raise RuntimeError(f"{action} 返回空響應")
        if getattr(body, "success", None) is False or (
            getattr(body, "code", None) not in (None, 0, "0")
        ):
            raise RuntimeError(
                f"{action} 失敗：{getattr(body, 'message', 'unknown')}；"
                f"requestId={getattr(body, 'request_id', '')}"
            )

    def upload(
        self,
        knowledge_base_id: str,
        file_paths: Sequence[str | Path],
        meta_fields: Mapping[str, Any] | None = None,
    ) -> dict[str, Any]:
        docs = [LocalDocument.from_path(path) for path in file_paths]
        presign_docs = [
            models.GetKnowledgeBasePreSignedUrlRequestDocuments(
                path=doc.object_path,
                name=doc.path.name,
                size=doc.path.stat().st_size,
            )
            for doc in docs
        ]
        response = self.client.get_knowledge_base_pre_signed_url(
            knowledge_base_id,
            models.GetKnowledgeBasePreSignedUrlRequest(
                knowledge_base_id=knowledge_base_id,
                documents=presign_docs,
                expires_in=3600,
            ),
        )
        body = response.body
        self._check(body, "GetKnowledgeBasePreSignedUrl")

        urls = list(body.data.pre_signed_urls or [])
        if len(urls) != len(docs):
            raise RuntimeError("預簽名 URL 數量與檔案數量不一致")

        for doc, url in zip(docs, urls, strict=True):
            with doc.path.open("rb") as source:
                # 預簽名 URL 按空 Content-Type 簽發，請求不要攜帶 Content-Type。
                self.http.put(url, data=source, timeout=120).raise_for_status()

        add_docs = [
            models.AddDocumentsRequestDocuments(
                path=doc.object_path,
                name=doc.path.name,
                size=doc.path.stat().st_size,
            )
            for doc in docs
        ]
        response = self.client.add_documents(
            knowledge_base_id,
            models.AddDocumentsRequest(
                knowledge_base_id=knowledge_base_id,
                import_type="LOCAL_UPLOAD",
                documents=add_docs,
                meta_fields=dict(meta_fields) if meta_fields else None,
                dedup=models.AddDocumentsRequestDedup(
                    doc_name_dedup=True,
                    content_dedup=False,
                ),
            ),
        )
        body = response.body
        self._check(body, "AddDocuments")

        errors = list(getattr(body.data, "errors", None) or [])
        if errors:
            raise RuntimeError(f"資料註冊失敗：{errors}")
        return body.to_map()

    def search(
        self,
        knowledge_base_id: str,
        query: str,
        options: SearchOptions,
        image_url: str | None = None,
    ) -> dict[str, Any]:
        tag_filter = None
        if options.tag_conditions:
            tag_filter = models.SearchKnowledgeBaseRequestTagFilter(
                relation=options.tag_relation,
                conditions=[
                    models.SearchKnowledgeBaseRequestTagFilterConditions(
                        field=condition.field,
                        op=condition.op,
                        value=condition.value,
                    )
                    for condition in options.tag_conditions
                ],
            )
        response = self.client.search_knowledge_base(
            knowledge_base_id,
            models.SearchKnowledgeBaseRequest(
                query=query,
                version=options.version,
                page_number=1,
                page_size=options.page_size,
                rerank_model_name=options.rerank_model_name,
                tag_filter=tag_filter,
                image=(
                    models.SearchKnowledgeBaseRequestImage(url=image_url)
                    if image_url
                    else None
                ),
                retrieval_config=models.SearchKnowledgeBaseRequestRetrievalConfig(
                    candidate_count=options.candidate_count,
                    min_score=options.min_score,
                    semantic_weight=options.semantic_weight,
                    enable_query_expansion=options.enable_query_expansion,
                ),
            ),
        )
        body = response.body
        self._check(body, "SearchKnowledgeBase")
        return body.to_map()

步驟五：編寫批量上傳指令碼

將下面內容儲存為 upload.py。AddDocuments 的 MetaFields 對整批生效，因此指令碼先按標籤分組，再按 100 篇分批提交。

"""按標籤批量上傳本地資料；上傳後需到控制台等待解析並發布版本。"""

from __future__ import annotations

import argparse
import json
from collections import defaultdict
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from kb_client import KnowledgeBaseClient


@dataclass(frozen=True)
class ManifestEntry:
    path: Path
    metadata: dict[str, Any]


def load_manifest(path: Path) -> list[ManifestEntry]:
    entries: list[ManifestEntry] = []
    for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
        if not raw_line.strip():
            continue
        value = json.loads(raw_line)
        file_path = (path.parent / str(value["path"])).resolve()
        metadata = value.get("metadata") or {}
        if not isinstance(metadata, dict):
            raise ValueError(f"manifest 第 {line_number} 行 metadata 必須是對象")
        entries.append(ManifestEntry(file_path, metadata))
    return entries


def discover(paths: list[str], default_metadata: dict[str, Any]) -> list[ManifestEntry]:
    entries: list[ManifestEntry] = []
    for value in paths:
        path = Path(value).expanduser()
        files = sorted(item for item in path.rglob("*") if item.is_file()) if path.is_dir() else [path]
        entries.extend(ManifestEntry(item.resolve(), default_metadata) for item in files)
    return entries


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("paths", nargs="*", help="檔案或目錄，可同時傳多個")
    parser.add_argument("--config", default="config.json")
    parser.add_argument("--manifest", help="JSONL 檔案；每行包含 path 和 metadata")
    args = parser.parse_args()

    config = json.loads(Path(args.config).read_text(encoding="utf-8"))
    aliyun = config["aliyun"]
    upload_config = config.get("upload") or {}
    default_metadata = upload_config.get("default_meta_fields") or {}
    entries = (
        load_manifest(Path(args.manifest).expanduser().resolve())
        if args.manifest
        else discover(args.paths, default_metadata)
    )
    if not entries:
        raise SystemExit("沒有找到可上傳檔案")

    grouped: dict[str, list[ManifestEntry]] = defaultdict(list)
    for entry in entries:
        key = json.dumps(entry.metadata, ensure_ascii=False, sort_keys=True)
        grouped[key].append(entry)

    client = KnowledgeBaseClient(
        aliyun["access_key_id"], aliyun["access_key_secret"], aliyun["region_id"]
    )
    submitted = 0
    for metadata_key, group in grouped.items():
        metadata = json.loads(metadata_key)
        for start in range(0, len(group), 100):
            batch = group[start : start + 100]
            client.upload(
                aliyun["knowledge_base_id"],
                [entry.path for entry in batch],
                meta_fields=metadata,
            )
            submitted += len(batch)
            print(f"已提交 {submitted}/{len(entries)} 篇；metadata={metadata}")


if __name__ == "__main__":
    main()

執行上傳：

python upload.py --manifest documents.jsonl

輸出會按標籤分組逐批回顯，例如 5 篇資料、4 種標籤組合時會提交 4 批：

已提交 2/5 篇；metadata={'docType': 'policy', 'effectiveDate': '2026-01-01', 'productLine': 'all'}
已提交 3/5 篇；metadata={'docType': 'policy', 'effectiveDate': '2026-03-01', 'productLine': 'phone'}
已提交 4/5 篇；metadata={'docType': 'manual', 'effectiveDate': '2026-03-01', 'productLine': 'phone'}
已提交 5/5 篇；metadata={'docType': 'faq', 'effectiveDate': '2026-02-01', 'productLine': 'all'}

步驟六：發布版本

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

1. 進入知識庫詳情頁，單擊版本管理頁簽，確認存在待發布變更後單擊發布版本。

2. 在確認變更步驟核對變更日誌（會逐條列出新增的資料），單擊下一步。

3. 在填寫說明步驟輸入發布說明（不超過 200 字元），單擊發布。

4. 在版本記錄中確認新版本狀態為發行，並記錄版本號碼（首次發布為 v1，此後依次為 v2、v3）。

config.json 中 knowledge_base_version 使用 LATEST_PUBLISHED 時會自動檢索最新發行版本，也可以改為明確的版本號碼。請勿使用 DRAFT 對外提供穩定問答服務。

每條檢索結果還會返回 scoreDetails，形如 {"keywordScore": 0.368, "semanticScore": 0.819}，用於判斷該命中主要來自關鍵詞還是語義匹配。

三個分數的含義是：keywordScore 是關鍵詞相似性；semanticScore 在未啟用重排時是向量相似性，啟用重排後是重排模型的分數；score 是兩者按權重加權後的最終排序與過濾分數，計算方式為 score ≈ (1-semantic_weight) × keywordScore + semantic_weight × semanticScore（另可能疊加 rank feature），min_score 就是在這個最終分數上做過濾。

不同模型、是否啟用重排、不同權重下分數量綱都會變化，不能跨配置橫向比較，也沒有一個通用的推薦閾值：建議先把 min_score 設為 0 取回一批結果並人工標註相關性，再按召回與誤召回的分布選定閾值，之後每次調整 semantic_weight 或開關重排都要重新校準。

步驟七：按標籤過濾檢索

在 config.json 的 retrieval.tag_filter.conditions 中配置過濾條件，即可把檢索範圍限定到指定標籤。例如只檢索政策類資料：

"tag_filter": {
  "relation": "and",
  "conditions": [
    {"field": "docType", "op": "=", "value": "policy"}
  ]
}

relation 支援 and（同時滿足）與 or（滿足任一）。op 支援以下運算子：

以本文語料為例，不同過濾條件的檢索範圍：

步驟八：編寫問答服務

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

"""知識庫檢索 + OpenAI 相容大模型問答服務。"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

import requests
from flask import Flask, jsonify, render_template, request

from kb_client import KnowledgeBaseClient, SearchOptions, TagCondition


CONFIG = json.loads(Path("config.json").read_text(encoding="utf-8"))
ALIYUN = CONFIG["aliyun"]
LLM = CONFIG.get("llm", {})
SCENARIO = CONFIG.get("scenario", {})
RETRIEVAL = CONFIG.get("retrieval", {})
KB = KnowledgeBaseClient(
    ALIYUN["access_key_id"], ALIYUN["access_key_secret"], ALIYUN["region_id"]
)
app = Flask(__name__)


def search_options() -> SearchOptions:
    raw_conditions = (RETRIEVAL.get("tag_filter") or {}).get("conditions") or []
    return SearchOptions(
        version=ALIYUN.get("knowledge_base_version", "LATEST_PUBLISHED"),
        page_size=int(RETRIEVAL.get("page_size", 6)),
        candidate_count=int(RETRIEVAL.get("candidate_count", 48)),
        min_score=float(RETRIEVAL.get("min_score", 0.0)),
        semantic_weight=float(RETRIEVAL.get("semantic_weight", 0.5)),
        enable_query_expansion=bool(RETRIEVAL.get("enable_query_expansion", True)),
        rerank_model_name=str(RETRIEVAL.get("rerank_model_name") or "") or None,
        tag_relation=str((RETRIEVAL.get("tag_filter") or {}).get("relation", "and")),
        tag_conditions=tuple(
            TagCondition(str(item["field"]), str(item["op"]), item.get("value"))
            for item in raw_conditions
        ),
    )


def find_results(payload: Any) -> list[dict[str, Any]]:
    """從響應中提取檢索切片列表。"""
    if isinstance(payload, list):
        return [item for item in payload if isinstance(item, dict)]
    if not isinstance(payload, dict):
        return []
    for key in ("results", "Results"):
        if isinstance(payload.get(key), list):
            return payload[key]
    for key in ("data", "Data"):
        found = find_results(payload.get(key))
        if found:
            return found
    return []


def field(item: dict[str, Any], *names: str) -> Any:
    for name in names:
        if item.get(name) not in (None, ""):
            return item[name]
    return ""


def llm_answer(question: str, results: list[dict[str, Any]]) -> str:
    if not LLM.get("enabled", True):
        return "大模型未啟用；請查看下方檢索結果。"
    if not results:
        return "現有資料中沒有明確說明。"
    context = "\n\n".join(
        f"[來源{index}] {field(item, 'documentName', 'DocumentName')}\n"
        f"{field(item, 'content', 'Content')}"
        for index, item in enumerate(results, 1)
    )
    url = str(LLM["base_url"]).rstrip("/") + "/chat/completions"
    response = requests.post(
        url,
        headers={"Authorization": f"Bearer {LLM['api_key']}"},
        json={
            "model": LLM["model"],
            "temperature": 0.1,
            "messages": [
                {"role": "system", "content": SCENARIO.get("system_prompt", "只依據資料回答。")},
                {"role": "user", "content": f"問題：{question}\n\n檢索資料：\n{context}"},
            ],
        },
        timeout=90,
    )
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"].strip()


@app.get("/")
def index():
    return render_template(
        "index.html",
        title=SCENARIO.get("title", "知識庫問答"),
        sample_questions=SCENARIO.get("sample_questions", []),
        image_enabled=bool(SCENARIO.get("image_enabled", False)),
    )


@app.post("/api/ask")
def ask():
    payload = request.get_json(silent=True) or {}
    question = str(payload.get("question", "")).strip()
    image_url = str(payload.get("image_url", "")).strip() or None
    if not question:
        return jsonify({"error": "問題不可為空"}), 400
    try:
        raw = KB.search(
            ALIYUN["knowledge_base_id"],
            question,
            options=search_options(),
            image_url=image_url,
        )
        results = find_results(raw)
        return jsonify({"answer": llm_answer(question, results), "sources": results})
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


if __name__ == "__main__":
    app.run(host="127.0.0.1", port=7860, debug=False)

將下面內容儲存為 templates/index.html。

<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width,initial-scale=1">
  <title>{{ title }}</title>
  <style>
    body{margin:0;background:#f6f7fb;color:#1f2937;font:15px system-ui,sans-serif}
    main{max-width:860px;margin:0 auto;padding:42px 18px}
    .card{background:#fff;border:1px solid #e5e7eb;border-radius:18px;padding:24px}
    textarea{box-sizing:border-box;width:100%;min-height:100px;border:1px solid #d1d5db;border-radius:12px;padding:14px;font:inherit}
    button{margin-top:12px;border:0;border-radius:10px;padding:11px 18px;background:#4f46e5;color:#fff;cursor:pointer}
    .chip{background:#eef2ff;color:#3730a3;margin:4px;padding:7px 10px}
    pre{white-space:pre-wrap;line-height:1.65}.muted{color:#6b7280}.source{border-top:1px solid #eee;padding:12px 0}
  </style>
</head>
<body><main><h1>{{ title }}</h1><p class="muted">回答由發行知識庫內容產生，並展示檢索依據。</p>
  <div>{% for q in sample_questions %}<button class="chip" onclick='setQ({{ q|tojson }})'>{{ q }}</button>{% endfor %}</div>
  <section class="card">
    <textarea id="q" placeholder="輸入問題"></textarea>
    <button id="ask" onclick="ask()">發送</button>
    <pre id="answer"></pre>
    <div id="sources"></div>
  </section>
</main><script>
const q=document.querySelector('#q'), answer=document.querySelector('#answer'), sources=document.querySelector('#sources');
function setQ(value){q.value=value;q.focus()}
async function ask(){
  const text=q.value.trim(); if(!text) return;
  answer.textContent='檢索與產生中…'; sources.innerHTML='';
  const res=await fetch('/api/ask',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({question:text})});
  const data=await res.json();
  answer.textContent=data.answer||('Error: '+data.error);
  for(const [i,item] of (data.sources||[]).entries()){
    const div=document.createElement('div'); div.className='source';
    div.textContent=`來源 ${i+1}：${item.documentName||''}\n${item.content||''}`;
    sources.appendChild(div);
  }
}
</script></body></html>

步驟九：啟動並驗證

1. 啟動服務。

   python app.py

2. 瀏覽器訪問 http://127.0.0.1:7860，單擊樣本問題或手動輸入問題後單擊發送。

3. 也可以直接調用介面驗證。

curl -sS http://127.0.0.1:7860/api/ask \
  -H 'Content-Type: application/json' \
  -d '{"question":"商品通常在付款後多久發貨？"}'

按以下四條驗收：

• 頁面能正常開啟，提交問題後同時返回答案與檢索來源。

• 回答中的 [來源N] 能在下方來源區找到對應的資料切片。

• 提問資料沒有覆蓋的內容時，回答為「現有資料中沒有明確說明」，而不是補寫事實。

• 補充資料並在控制台重新發布版本後，頁面能檢索到新內容。

效果調優建議

• 按資料形態調整切片長度：客服資料多為短條目的 FAQ 與政策條款，可在知識庫詳情頁的處理策略中單擊建立策略，切片方式選擇智能切分或按長度切分。注意最大分段長度的單位是字元（預設 512 字元），短條目情境建議設定為 380~580 字元（約相當於 256~384 tokens）。策略建立後可在匯入時通過 AddDocumentsRequest.strategy_id 指定。

• 重排與 min_score 需要一起調：開啟 rerank_model_name 後，重排分數與向量相似性分數不是同一量綱，沿用原來的 min_score 可能把結果裁剪得過少（實測三個客服問題在開啟重排後各自只保留 1 條命中）。若問題需要綜合多篇資料作答，請在開啟重排時適當降低 min_score。

• 警惕"部分相關"資料引起的錯誤結論：當提問與某篇資料字面相關但語義不覆蓋時（例如資料唯寫了發貨時效，使用者問"是否支援貨到付款"），大模型可能據此給出錯誤結論。請在 system_prompt 中明確要求"資料未直接覆蓋問題時必須回答不知道，不得據部分相關資料推斷"，並在上線前用真實客戶問法抽查。

• 用真實客戶問法測試，不要只用手冊標題做查詢；semantic_weight 偏高（如 0.7）更有利於覆蓋口語化表達。

• 優先匯入正式生效的政策，避免同時保留相互衝突的舊版本；政策更新後重新發布版本，並用 effectiveDate 標籤區分。

常見問題

上線前檢查

• 使用獨立、最小許可權、可輪換的 RAM 使用者，不要長期使用主帳號 AccessKey。

• 只匯入有權處理的資料，回答必須能在來源片段中找到依據。

• 對外部署時不要繼續使用 Flask 程式開發伺服器，應改用生產 WSGI 服務，並增加身份認證、HTTPS、訪問日誌脫敏、限流與審計。

• 關鍵政策類問答建議保留人工兜底入口。