Vector Retrieval Service for Milvus:Crie uma aplicação de perguntas e respostas para atendimento ao cliente inteligente usando a base de conhecimento do Alibaba Cloud Milvus

Observações de uso

• Tipos de campo — O tipo de campo pode ser string, int64, list, float32 ou bool.

• Bases de conhecimento existentes — É possível adicionar tags a uma base de conhecimento existente posteriormente, sem precisar reconstruí-la.

• Definições não são pré-requisito para gravar valores — Definir tags no console não é obrigatório para gravar valores de tags. Campos indefinidos ainda podem ser gravados junto com os documentos e usados para filtragem. Além disso, exclua uma definição de tag não apaga valores históricos já gravados. A definição não funciona como uma declaração independente de campo de índice. Na prática, as definições servem para exibição no console, gerenciamento de opções para tags do tipo lista e conversão de tipos de valor durante a filtragem por string, int64, float32, bool ou list. A prática recomendada é definir as tags primeiro para obter validação de tipo e maior gerenciabilidade, em vez de tratar isso como um pré-requisito obrigatório.

• Alterações nas definições não reconstroem dados — A caixa de diálogo avisa que alterações nas tags afetam a construção do índice, mas adicionar ou remover definições de tags não reconstrói nem migra dados já importados, e os valores históricos das tags não são apagados. Ainda assim, finalize as definições de tags antes da importação em lote para garantir que a exibição no console, o gerenciamento de opções e a conversão de tipos de valor permaneçam consistentes desde o início.

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

{"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"}}

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

{
  "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": "Intelligent Customer Service Knowledge Base Q&A",
    "system_prompt": "You are a corporate customer service assistant. Answer based only on the retrieved documents, in a clear and friendly customer service tone; cite [Source N] for key conclusions. If the retrieved documents do not directly cover the user's question, you must answer 'The available documents do not explicitly cover this.' Do not infer conclusions from partially relevant documents, and do not add contact methods or service channels beyond the documents.",
    "image_enabled": false,
    "sample_questions": [
      "How long after payment does it usually take to ship the product?",
      "Can the product still be repaired after the warranty expires?",
      "What conditions must be met to request a return?"
    ]
  }
}

"""Milvus knowledge base OpenAPI SDK: local upload with tags and search against a published version."""

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:
        # Successful responses do not return the code field (its value is None), so treat only an explicit non-zero code as a failure.
        if body is None:
            raise RuntimeError(f"{action} returned an empty response")
        if getattr(body, "success", None) is False or (
            getattr(body, "code", None) not in (None, 0, "0")
        ):
            raise RuntimeError(
                f"{action} failed: {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("The number of pre-signed URLs does not match the number of files")

        for doc, url in zip(docs, urls, strict=True):
            with doc.path.open("rb") as source:
                # The pre-signed URL is signed with an empty Content-Type; do not include Content-Type in the request.
                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"Failed to register data: {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()

"""Batch upload local documents by tag; after upload, wait for parsing in the console and publish a version."""

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"metadata on line {line_number} of the manifest must be an object")
        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="Files or directories; you can pass multiple")
    parser.add_argument("--config", default="config.json")
    parser.add_argument("--manifest", help="JSONL file; each line contains path and 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("No files found to upload")

    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 {submitted}/{len(entries)} documents; metadata={metadata}")

if __name__ == "__main__":
    main()

python3 upload.py --manifest documents.jsonl

Submitted 2/5 documents; metadata={'docType': 'policy', 'effectiveDate': '2026-01-01', 'productLine': 'all'}
Submitted 3/5 documents; metadata={'docType': 'policy', 'effectiveDate': '2026-03-01', 'productLine': 'phone'}
Submitted 4/5 documents; metadata={'docType': 'manual', 'effectiveDate': '2026-03-01', 'productLine': 'phone'}
Submitted 5/5 documents; metadata={'docType': 'faq', 'effectiveDate': '2026-02-01', 'productLine': 'all'}

Cotas de versão e exclusão

• Cota padrão — Por padrão, podem existir no máximo 3 versões publicadas simultaneamente.

• Escopo da cota — Esse limite é calculado por tenant e não aumenta com a especificação de CU da instância.

• Aumento de cota — Para elevar o limite, envie um ticket para avaliação. Atualmente, não há uma entrada de autoatendimento para solicitação de cota pelos usuários.

• Comportamento no limite — Ao atingir o limite, o botão Publish Version fica acinzentado, enquanto a página ainda mostra "There are N pending changes to publish". exclua versões antigas que não são mais necessárias em Version History antes de publicar novamente.

• Atualizações frequentes — Se os documentos forem atualizados frequentemente, mantenha apenas a versão vigente atual mais a versão histórica mais recente.

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

"""Knowledge base retrieval + OpenAI-compatible LLM Q&A service."""

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]]:
    """Extract the list of retrieved chunks from the response."""
    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 "The LLM is disabled. See the retrieval results below."
    if not results:
        return "The available documents do not explicitly cover this."
    context = "\n\n".join(
        f"[Source {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", "Answer based only on the documents.")},
                {"role": "user", "content": f"Question: {question}\n\nRetrieved documents:\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", "Knowledge Base Q&A"),
        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": "The question cannot be empty"}), 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)

<!doctype html>
<html lang="en">
<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">Answers are generated from the published knowledge base content, with retrieval sources displayed.</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="Enter your question"></textarea>
    <button id="ask" onclick="ask()">Send</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='Retrieving and generating...'; 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=`Source ${i+1}: ${item.documentName||''}\n${item.content||''}`;
    sources.appendChild(div);
  }
}
</script></body></html>

python3 app.py

curl -sS http://127.0.0.1:7860/api/ask \
  -H 'Content-Type: application/json' \
  -d '{"question":"How long after payment does it usually take to ship the product?"}'

Entendendo as pontuações de pesquisa

Cada resultado de pesquisa também retorna scoreDetails, por exemplo {"keywordScore": 0.368, "semanticScore": 0.819}, o que ajuda a julgar se a correspondência veio principalmente de palavras-chave ou de busca semântica.

As três pontuações significam o seguinte:

• keywordScore é a similaridade por palavras-chave.

• semanticScore é a similaridade vetorial quando o reranking está desativado, e a pontuação do modelo de rerank quando o reranking está ativado.

• score é a pontuação final de classificação e filtragem obtida pela ponderação das duas anteriores, calculada como score ≈ (1-semantic_weight) × keywordScore + semantic_weight × semanticScore (um rank feature também pode ser adicionado). O min_score filtra com base nessa pontuação final. A escala de pontuação muda entre modelos, com ou sem reranking e com diferentes pesos. As pontuações não podem ser comparadas entre configurações diferentes e não existe um limiar universalmente recomendado: defina inicialmente o min_score como 0, recupere um lote de resultados e rotule manualmente sua relevância, escolhendo então um limiar baseado na distribuição de recalls e falsos positivos. Recalibre sempre que ajustar o semantic_weight ou alternar o reranking.