向量检索服务 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、访问日志脱敏、限流与审计。

• 关键政策类问答建议保留人工兜底入口。