Tablestore:Build your first knowledge base

Prerequisites

• An AgentStorage instance has been created, and you have obtained the AgentStorage name and endpoint. The knowledge base will be created in this instance.

• An OSS bucket has been created in the same region, and the authorization for Tablestore to read from and write to OSS has been completed.

• You have prepared your AccessKey ID and AccessKey Secret. In production environments, use a RAM user or temporary credentials.

• A PDF, Word, Markdown, or text file is available locally for testing.

Step 1: Install the SDK

pip install tablestore-agent-storage

Step 2: Create the client

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>",
)

Step 3: Create a knowledge base and upload documents

client.create_knowledge_base({
    "knowledgeBaseName": "product_docs_kb",
    "description": "Product documentation knowledge base",
})

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

The SDK first uploads the local file to OSS, and then submits an asynchronous import task. When you create the knowledge base, if no retrieval configuration is specified, the service uses the default configuration.

Step 4: Wait for the document to become searchable

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", "Document processing failed"))
    time.sleep(5)
else:
    raise TimeoutError("Timed out waiting for document indexing")

A document cannot be retrieved while it is in the pending or indexing state. Production code should use polling with backoff and a total timeout, and handle the failed state.

Step 5: Perform retrieval

response = client.retrieve({
    "knowledgeBaseName": "product_docs_kb",
    "retrievalQuery": {
        "type": "TEXT",
        "text": "What are the installation steps for the product?",
    },
})

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

The preceding steps have completed the full pipeline from raw document through chunk indexing to relevant content retrieval. The retrieval results are evidence chunks for RAG applications to use, not final answers generated by a model.

Next steps

• Learn more about the Knowledge Storage Service: Knowledge Storage Service

• Import OSS files or directories: Document management

• Tune hybrid retrieval and reranking: Retrieval and Ranking

• Use TypeScript, the CLI, or the Dashboard: Agent Storage SDK, Command-line tool