跳至主要内容

Python SDK

套件： vecstruct
最低版本： Python 3.12+
GitHub： vecstruct/vecstruct-sdk-python

:::info Async-only Python SDK 只支援非同步（async/await）模式，底層使用 httpx.AsyncClient。 :::

安裝​

pip install vecstruct python-dotenv

初始化​

import asyncio
from dotenv import load_dotenv
from vecstruct import AsyncVecstruct

load_dotenv()

async def main():
    async with AsyncVecstruct() as client:
        # VECSTRUCT_API_KEY 環境變數會自動被讀取
        ...

asyncio.run(main())

也可以手動傳入 API Key：

client = AsyncVecstruct(api_key="sk-...")

進階設定：

import httpx
from vecstruct import AsyncVecstruct

# 自訂逾時
timeout = httpx.Timeout(connect=5.0, read=60.0, write=30.0, pool=5.0)
client = AsyncVecstruct(timeout=timeout)

# 自訂 HTTP client（設定 proxy 等）
http_client = httpx.AsyncClient(proxy="http://localhost:8080")
client = AsyncVecstruct(api_key="sk-...", http_client=http_client)

# 非生產環境 base_url
client = AsyncVecstruct(api_key="sk-...", base_url="https://api.staging.vecstruct.com/v1")

Chat Completions​

一般請求​

async with AsyncVecstruct() as client:
    completion = await client.chat.completions.create(
        model="openai/gpt-4o",
        messages=[
            {"role": "system", "content": "你是資深工程師"},
            {"role": "user", "content": "什麼是向量資料庫？"},
        ],
        temperature=0.7,
    )
    print(completion["choices"][0]["message"]["content"])

串流回應​

async with AsyncVecstruct() as client:
    async for chunk in client.chat.completions.stream(
        model="openai/gpt-4o",
        messages=[{"role": "user", "content": "用繁體中文介紹 RAG 技術"}],
    ):
        if "choices" in chunk:
            print(chunk["choices"][0]["delta"].get("content", ""), end="", flush=True)

啟用 RAG 與 Memory​

async with AsyncVecstruct() as client:
    completion = await client.chat.completions.create(
        model="openai/gpt-4o",
        messages=[{"role": "user", "content": "我們的退款政策是什麼？"}],
        vecstruct={
            "rag": True,
            "rag_top_k": 5,
            "use_memory": True,
            "metadata": {"user_id": "user-123"},
        },
    )

Embeddings​

async with AsyncVecstruct() as client:
    res = await client.embeddings.create(
        model="openai/text-embedding-3-small",
        input=["向量搜尋的原理", "如何評估 Embedding 品質"],
    )
    print(len(res["data"][0]["embedding"]))   # 向量維度

Rerank​

async with AsyncVecstruct() as client:
    result = await client.rerank.create(
        model="bge-reranker-v2-m3",
        query="OAuth 認證流程",
        documents=["OAuth 2.0 認證流程包含...", "今天天氣很好", "Authorization Code Flow..."],
        top_n=2,
    )
    for r in result["results"]:
        print(r["relevance_score"], r["document"][:50])

Models​

async with AsyncVecstruct() as client:
    all_models = await client.models.list()
    openai_models = await client.models.list(owned_by="openai")
    
    # 按 model_id 搜尋
    matches = await client.models.list(model_id="openai/gpt-4o")
    print(matches[0]["model_id"])

RAG 查詢（知識庫）​

async with AsyncVecstruct() as client:
    result = await client.rag.query(
        query="什麼是向量資料庫？",
        top_k=5,
        min_similarity=0.7,
        rag_query_expansion="rewrite",
        source_ids=["doc-id-1"],   # 可限定只搜尋特定文件
    )
    for r in result["results"]:
        print(f"{r['score']:.3f} | {r['content'][:80]}")

Memory（長期記憶）​

async with AsyncVecstruct() as client:
    # 新增
    mem = await client.memories.create(
        content="使用者偏好繁體中文回覆，技術術語保留英文",
        memory_type="rule",
        importance=0.9,
    )

    # 語義搜尋
    hits = await client.memories.search(
        query="使用者語言偏好",
        top_k=5,
        min_similarity=0.3,
    )

    # 從文字自動萃取
    extracted = await client.memories.extract(
        text="使用者使用 React 19 + TypeScript，偏好函數式寫法",
        deduplicate=True,
    )

    # 分頁列表
    memories = await client.memories.list(page=1, page_size=20)

    # 更新 / 刪除
    await client.memories.update(mem["id"], importance=0.7)
    await client.memories.delete(mem["id"])
    
    # 批次刪除
    await client.memories.delete_many(memory_type="context")

Documents（文件管理）​

async with AsyncVecstruct() as client:
    # 上傳（支援路徑字串、bytes 或 file 物件）
    doc = await client.documents.upload(
        file="./manual.pdf",
        metadata={"team": "support"},
        chunk_strategy="recursive",
        chunk_size=800,
        chunk_overlap=80,
    )
    
    # 也可以用 bytes
    inline_doc = await client.documents.upload(
        file=b"# 說明文件\n\nHello Vecstruct.",
        file_name="inline.md",
    )

    # 列表 / 取得 / 刪除
    documents = await client.documents.list(page=1, page_size=20)
    detail = await client.documents.get(doc["id"])
    await client.documents.delete(doc["id"])

    # 啟用 / 停用 / 重新索引
    await client.documents.deactivate(doc["id"])
    await client.documents.activate(doc["id"])
    await client.documents.reindex(doc["id"])

    # 更新切分設定
    await client.documents.update_chunking(
        doc["id"],
        chunk_strategy="semantic",
        chunk_semantic_threshold=0.75,
    )

    # 下載
    raw_bytes = await client.documents.download(doc["id"])
    markdown = await client.documents.download_markdown(doc["id"])

    # 查看段落
    chunks = await client.documents.chunks(doc["id"])

稽核日誌​

async with AsyncVecstruct() as client:
    # 列表
    logs = await client.audit.logs.list(
        start_time="2026-01-01T00:00:00Z",
        end_time="2026-01-31T23:59:59Z",
        action="gateway.chat",
        page=1,
        page_size=20,
    )

    # 詳情
    detail = await client.audit.logs.get(logs["logs"][0]["id"])

    # 匯出
    export_job = await client.audit.logs.export(
        start_time="2026-01-01T00:00:00Z",
        end_time="2026-01-31T23:59:59Z",
        format="json",
    )

認證資訊​

async with AsyncVecstruct() as client:
    info = await client.auth.me()
    print("User ID:", info["user_id"])
    print("Permissions:", info["permissions"])
    print("Project IDs:", info["project_ids"])

錯誤處理​

from vecstruct import (
    AsyncVecstruct,
    AuthenticationError,
    PermissionDeniedError,
    PaymentRequiredError,
    RateLimitError,
    NotFoundError,
    APIConnectionError,
    APIStatusError,
)

async with AsyncVecstruct() as client:
    try:
        await client.chat.completions.create(
            model="openai/gpt-4o",
            messages=[{"role": "user", "content": "hi"}],
        )
    except RateLimitError as e:
        print(f"速率限制：{e.status_code} / {e.error_code}")
    except PaymentRequiredError as e:
        print("餘額或配額不足")
    except AuthenticationError:
        print("API Key 無效")
    except APIStatusError as e:
        print(f"API 錯誤 {e.status_code}: {e.message}")
    except APIConnectionError as e:
        print(f"連線失敗: {e.message}")

API 錯誤實例包含：

  • status_code — HTTP 狀態碼
  • error_code — Vecstruct 業務錯誤碼
  • message — 錯誤描述
  • data — 後端額外資料（如有）