构建第一个知识库

前提条件

• 已创建 AgentStorage 实例，并获得 AgentStorage 名称和访问地址（endpoint）。知识库将创建在该实例中。

• 已在同地域创建 OSS Bucket，并完成表格存储读写 OSS 的授权。

• 已准备 AccessKey ID 和 AccessKey Secret。生产环境请使用 RAM 用户或临时凭证。

• 本地有一个用于测试的 PDF、Word、Markdown 或文本文件。

步骤1：安装 SDK

pip install tablestore-agent-storage

步骤2：创建客户端

from tablestore_agent_storage import AgentStorageClient
import time

client = AgentStorageClient(
    access_key_id="<AccessKey ID>",
    access_key_secret="<AccessKey Secret>",
    ots_endpoint="https://<instance>.<region>.ots.aliyuncs.com",
    ots_instance_name="<instance-name>",
    oss_endpoint="https://oss-<region>.aliyuncs.com",
    oss_bucket_name="<bucket-name>",
)

步骤3：创建知识库并上传文档

client.create_knowledge_base({
    "knowledgeBaseName": "product_docs_kb",
    "description": "产品文档知识库",
})

response = client.upload_documents({
    "knowledgeBaseName": "product_docs_kb",
    "documents": [{"filePath": "/path/to/product-guide.pdf"}],
})
doc_id = response["data"]["documentDetails"][0]["docId"]

SDK 先把本地文件上传到 OSS，再提交异步导入任务。创建知识库时未指定检索配置，服务使用默认配置。

步骤4：等待文档可检索

deadline = time.time() + 300

while time.time() < deadline:
    response = client.get_document({
        "knowledgeBaseName": "product_docs_kb",
        "docId": doc_id,
    })
    document = response["data"][0]
    status = document["status"]

    if status == "completed":
        break
    if status == "failed":
        raise RuntimeError(document.get("message", "文档处理失败"))
    time.sleep(5)
else:
    raise TimeoutError("等待文档索引超时")

文档在 pending 或 indexing 状态时还不能被检索。生产代码应使用带退避和总超时的轮询，并处理 failed 状态。

步骤5：执行检索

response = client.retrieve({
    "knowledgeBaseName": "product_docs_kb",
    "retrievalQuery": {
        "type": "TEXT",
        "text": "产品的安装步骤是什么？",
    },
})

for item in response["data"]["retrievalResults"]:
    print(f"[{item['score']:.4f}] {item['content'][:120]}")

以上步骤已经跑通“原始文档 → 切片索引 → 相关内容召回”的完整流程。召回结果是供 RAG 应用使用的证据切片，不是模型生成的最终答案。

下一步

• 了解更多知识存储服务：知识存储服务

• 导入 OSS 文件或目录：文档管理

• 调整混合检索与重排：检索和排序

• 使用 TypeScript、CLI 或 Dashboard：Agent Storage SDK、命令行工具