搭建个人知识问答应用-向量检索服务 Milvus 版(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 和访问日志脱敏。
接口开放范围和权限要求以产品文档与控制台实际为准。