schift 0.8.0

Embedding model migration SDK. Switch models without re-embedding.

Python SDK

Python client and local migration toolkit for workspace search, chat, and embedding migration.

The package currently exposes two public client styles:

• WorkspaceClient: modular client for live API operations such as bucket ingest, catalog lookup, embedding, routing, search, usage, and hosted bucket management.
• Client: legacy projection client for fitting and downloading Projection objects that run locally.

Install

Base package:

pip install schift

Optional adapters:

pip install "schift[postgres]"
pip install "schift[qdrant]"
pip install "schift[all]"

Development extras:

pip install -e ".[dev]"

Authentication And Base URLs

The modular client reads SCHIFT_API_KEY and a hosted API origin from the environment. The base URL is a bare origin — no /v1 suffix; version prefixes belong to request paths:

export SCHIFT_API_KEY=sch_your_key_here
export SCHIFT_API_URL=https://api.schift.io

Equivalent explicit construction:

import os

from schift import WorkspaceClient

client = WorkspaceClient(
    api_key="sch_your_key_here",
    base_url=os.environ["SCHIFT_API_URL"],  # bare origin, e.g. https://api.schift.io
    timeout=60.0,
)

Notes:

• WorkspaceClient expects a bare origin (e.g. https://api.schift.io). A legacy trailing /v1 is tolerated and stripped automatically, so old configs keep working — but new configs should not include it.
• If base_url is omitted, set SCHIFT_API_URL or SCHIFT_BASE_URL.
• The legacy Client also requires base_url, SCHIFT_API_URL, or SCHIFT_BASE_URL (bare origin) and appends /v1/... internally.

Client Lifecycle

WorkspaceClient holds a shared httpx.Client, so prefer a context manager for short-lived scripts:

from schift import WorkspaceClient

with WorkspaceClient() as client:
    models = client.catalog.list()

For long-running processes, keep one WorkspaceClient instance and call close() during shutdown.

Bucket Ingest And Search

from schift import WorkspaceClient

with WorkspaceClient() as client:
    bucket = client.buckets.create(name="finance-docs")
    upload = client.buckets.upload(
        bucket["id"],
        [("files", ("q1-report.pdf", open("q1-report.pdf", "rb").read(), "application/pdf"))],
    )

    jobs = client.buckets.list_jobs(bucket_id=bucket["id"])
    resp = client.buckets.search(
        bucket["id"],
        "revenue guidance",
        top_k=5,
        filter={"metadata": {"quarter": "Q1"}},
    )

    print(upload["bucket_id"])
    print(jobs[0]["status"] if jobs else "queued")
    print(resp.context)                 # paste-ready context with [n] markers
    print(resp.citations)               # [(1, "report.pdf p.3"), ...]
    print(resp[0].text if resp else "no hits")

search() returns a typed SearchResponse (the answer-ready v2 envelope: results with citations, warnings, degraded) and derives context / citations for direct LLM consumption. For raw chunks use the retrieve surface:

hits = client.buckets.retrieve(bucket["id"], "revenue guidance", top_k=8)
print(hits.status, hits.operational_status)
for hit in hits:
    print(hit.chunk_id, hit.score, hit.text[:80])

Use client.buckets and client.query(..., bucket=...) for retrieval, POST /v1/chat for bucket-backed RAG chat with sources, and POST /v1/chat/completions for OpenAI-compatible LLM routing without bucket context.

The bucket helper also exposes list_collections(), get_job(), list_jobs(), wait_for_job(), and poll_job() so scripts can keep ingest orchestration in one place.

Quickstart

from schift import WorkspaceClient

with WorkspaceClient() as client:
    models = client.catalog.list()
    vector = client.embed(
        "quarterly revenue report",
        model="openai/text-embedding-3-small",
    )

    bucket = client.buckets.create("finance-docs")
    client.db.upsert(
        collection="finance-docs",
        vectors=[
            {
                "id": "doc-1",
                "values": vector.tolist(),
                "metadata": {"source": "q1-report"},
            }
        ],
    )

    hits = client.query(
        "revenue guidance",
        bucket="finance-docs",
        top_k=5,
    )

    print(models[0]["id"] if models else "no models")
    print(hits[0].text if hits else "no hits")  # typed SearchResponse

Module Reference

catalog

List available models and inspect one model by ID.

from schift import WorkspaceClient

with WorkspaceClient() as client:
    models = client.catalog.list()
    model = client.catalog.get("openai/text-embedding-3-small")

embed

The embed module is callable for single-text requests and exposes batch() for multiple texts.

from schift import WorkspaceClient

with WorkspaceClient() as client:
    one = client.embed(
        "hello world",
        model="openai/text-embedding-3-small",
    )

    many = client.embed.batch(
        texts=["hello", "goodbye"],
        model="openai/text-embedding-3-small",
        dimensions=1024,
    )

Return values are NumPy arrays. Both calls go through the OpenAI-shaped POST /v1/embeddings endpoint ({"input": ..., "model": ...}).

providers

Register your own LLM provider API key (BYOK) so client.chat and client.completions call the provider directly instead of consuming the hosted platform shared LLM quota. Supported providers: "openai", "google", "anthropic".

from schift import WorkspaceClient

with WorkspaceClient() as client:
    # Register a key
    client.providers.set(
        "google",
        api_key="AIza...",
        # endpoint_url="https://custom-proxy.example.com",  # optional
    )

    # Check whether a provider is configured (api_key is never returned)
    status = client.providers.get("openai")
    # {"provider": "openai", "configured": True | False, "endpoint_url": None | str}

Rotation: a stored BYOK record shadows any server-side env var or secret for that provider. To rotate, call set() again with the new key — changing env vars alone has no effect on orgs with a BYOK record.

routing

Read or update the server-side routing policy used by Schift when a model is omitted.

from schift import WorkspaceClient

with WorkspaceClient() as client:
    current = client.routing.get()
    updated = client.routing.set(
        primary="openai/text-embedding-3-small",
        fallback="google/gemini-embedding-001",
        mode="failover",
    )

pii

Mask Korean PII before sending text to an LLM, vector store, or workflow step. Use types when you want the masking contract to be visible and selectable.

from schift import WorkspaceClient

PII_TYPES = [
    "resident_id",
    "alien_registration",
    "passport",
    "driver_license",
    "address",
    "phone",
    "bank_account",
]

with WorkspaceClient() as client:
    result = client.redact_pii(
        "주민등록번호 850205-1234567, 연락처 010-1234-5678",
        types=PII_TYPES,
    )
    print(result["masked"])

    masked = client.mask(
        "계좌 123-45-678901",
        types=["bank_account"],
    )

The default token format returns tokens such as [PII_PHONE_1] and a reverse_map so the caller can restore the original values after the workflow step. Keep that map only in your temporary restore path; Schift does not persist or gateway-cache it for pii_type_index requests. Set token_format="label_index" only when you need legacy tokens such as [PHONE_1].

Send only masked into the LLM, agent, vector store, or workflow step. Never include reverse_map in the AI payload; use it only after the AI result returns if your app needs to restore values.

restored = client.restore_pii(
    "고객 연락처 [PII_PHONE_1]로 안내하세요.",
    result["reverse_map"],
)

restore_pii() is a stateless API call. It requires an API key and does not persist or cache the map.

bench

Run a server-side benchmark between two model IDs. The SDK returns BenchReport.

from schift import WorkspaceClient

with WorkspaceClient() as client:
    report = client.bench.run(
        source="openai/text-embedding-3-small",
        target="google/gemini-embedding-001",
        data="./eval_queries.jsonl",
    )

    print(report.verdict)
    print(report.summary())

db

Manage hosted buckets and write vectors or raw documents.

from schift import WorkspaceClient

with WorkspaceClient() as client:
    vector = client.embed(
        "Schift reduces vector migration downtime.",
        model="openai/text-embedding-3-small",
    )

    bucket = client.db.create_collection(
        name="product-docs",
        dimension=len(vector),
    )

    client.db.upsert(
        collection="product-docs",
        vectors=[
            {
                "id": "doc-1",
                "values": vector.tolist(),
                "metadata": {"title": "Launch plan"},
            }
        ],
    )

    client.db.upsert_text(
        collection="product-docs",
        documents=[
            {
                "id": "doc-2",
                "text": "Schift reduces vector migration downtime.",
                "metadata": {"title": "Overview"},
            }
        ],
        model="openai/text-embedding-3-small",
    )

    stats = client.db.collection_stats("product-docs")

Available methods:

• create_collection(name, dimension)
• list_buckets()
• get_collection(name)
• collection_stats(name)
• delete_collection(name)
• upsert(collection, vectors)
• upsert_text(collection, documents, model)

query

The query module is callable and supports hosted buckets or an external DB handle.

from schift import WorkspaceClient

with WorkspaceClient() as client:
    hosted = client.query(
        "vector migration rollback plan",
        bucket="product-docs",
        top_k=10,
        rerank=True,
        rerank_top_k=5,
    )

    passthrough = client.query(
        "incident response",
        db="prod-search-db",
        model="openai/text-embedding-3-small",
        top_k=5,
    )

rerank

Rerank a list of candidate documents with a cross-encoder style endpoint.

from schift import WorkspaceClient

with WorkspaceClient() as client:
    reranked = client.rerank(
        "incident response",
        documents=[
            {"id": "doc-1", "text": "..."},
            {"id": "doc-2", "text": "..."},
        ],
        top_k=2,
    )

usage

Fetch aggregate usage for billing or dashboards.

from schift import WorkspaceClient

with WorkspaceClient() as client:
    usage = client.usage.get(period="30d")

The granularity parameter is deprecated (the server ignores it) and is no longer sent. See docs/USAGE_CONTRACT.md for the returned keys.

Workflows

Build and run RAG pipelines as composable DAGs.

Quick Start

from schift import WorkspaceClient

with WorkspaceClient() as client:
    # Create from template
    wf = client.workflow.create_rag("Product Search")

    # Run with inputs
    result = client.workflow.run(wf.id, inputs={"query": "best laptop"})
    print(result.outputs["answer"])

CRUD

wf = client.workflow.create("My Pipeline", description="Custom RAG")
workflows = client.workflow.list()
wf = client.workflow.get(wf.id)
wf = client.workflow.update(wf.id, name="Renamed")
client.workflow.delete(wf.id)

Building a Graph

wf = client.workflow.create("Custom RAG")

# Add blocks
start = client.workflow.add_block(wf.id, "start", title="Start")
retriever = client.workflow.add_block(wf.id, "retriever", config={
    "bucket": "my-docs",
    "top_k": 5,
    "rerank": True,
    "rerank_top_k": 3,
})
prompt = client.workflow.add_block(wf.id, "prompt_template", config={
    "template": "Answer based on:\n{{results}}\n\nQuestion: {{query}}",
})
llm = client.workflow.add_block(wf.id, "llm", config={
    "model": "openai/gpt-4o-mini",  # or "anthropic/claude-sonnet-4-20250514", "gemini-2.5-flash"
    "temperature": 0.7,
})
end = client.workflow.add_block(wf.id, "end")

# Connect blocks
client.workflow.add_edge(wf.id, start["id"], retriever["id"])
client.workflow.add_edge(wf.id, retriever["id"], prompt["id"])
client.workflow.add_edge(wf.id, prompt["id"], llm["id"])
client.workflow.add_edge(wf.id, llm["id"], end["id"])

Multiple Inputs

The start node forwards all input variables to downstream blocks:

result = client.workflow.run(wf.id, inputs={
    "query": "maternity leave policy",
    "user_id": "u-123",
    "language": "ko",
})

Reference variables in prompt templates with {{variable}} syntax.

Validation & Run History

# Validate graph (cycles, missing connections, etc.)
v = client.workflow.validate(wf.id)
print(v.valid, v.errors)

# List past runs
runs = client.workflow.runs(wf.id)
for r in runs:
    print(r.run_id, r.status, r.outputs)

YAML Import / Export

# Export to YAML
yaml_str = client.workflow.to_yaml(wf.id, path="pipeline.yaml")

# Load from file
definition = client.workflow.from_yaml("pipeline.yaml")

# Push YAML to create on server
wf = client.workflow.push_yaml("pipeline.yaml")

Templates

Method	Template
create_rag(name)	Retriever -> Prompt -> LLM
create_doc_qa(name)	Document QA with sources
create_chat(name)	Conversational RAG
create_ocr_ingest(name)	OCR -> Chunk -> Embed

Workflow Runs

from schift import WorkspaceClient

with WorkspaceClient() as client:
    wf = client.workflow.create_rag("Workspace RAG")
    result = client.workflow.run(wf.id, inputs={"query": "hello"})

Hosted Workflow v2 Runs

Use workflows_v2 for stored and published Workflow v2 definitions. This calls POST /v2/workflows/{workflow_id}/run and keeps simulate/live, approvals, and mock binding payloads aligned with the hosted API.

from schift import WorkspaceClient

with WorkspaceClient() as client:
    run = client.workflows_v2.run(
        "wf_v2_123",
        inputs={"invoice": "T-001"},
        mode="live",
        approvals={"review_issue": True},
        mock_bindings={"tax_provider": {"status": "mocked"}},
    )
    print(run["status"])

Block Types

Category	Types
Control	start, end, conditional, loop
Retrieval	retriever (with rerank toggle), reranker
LLM	llm (OpenAI/Anthropic/Google), prompt_template, answer
Data	document_loader, chunker, embedder, text_processor
Integration	api_call, webhook, code_executor
Storage	vector_store, cache

API Reference

Method	Description
workflow.create(name)	Create workflow
workflow.get(id)	Get workflow
workflow.list()	List workflows
workflow.update(id, **kw)	Update workflow
workflow.delete(id)	Delete workflow
workflow.run(id, inputs)	Run workflow
workflow.runs(id)	List past runs
workflow.validate(id)	Validate graph
workflow.add_block(id, type, config)	Add block
workflow.add_edge(id, src, tgt)	Add edge
workflow.to_yaml(id)	Export as YAML
workflow.from_yaml(path)	Load YAML definition
workflow.push_yaml(path)	Import YAML to server

Projection Workflow

Projection creation still lives in the legacy Client API. It returns a local Projection object that can transform vectors without further API calls.

from schift import Client

legacy = Client(api_key="sch_your_key_here")

# Embed the same sample texts with both providers before fitting.
source_pairs = old_model_embeddings
target_pairs = new_model_embeddings

projection = legacy.fit(
    source=source_pairs,
    target=target_pairs,
    source_model="openai/text-embedding-3-small",
    target_model="google/gemini-embedding-001",
    project_name="openai-to-gemini",
)

converted = projection.transform(source_pairs[:10])
projection.save("./projection-openai-to-gemini")

You can later reload a saved projection:

from schift import Projection

projection = Projection.load("./projection-openai-to-gemini")

The legacy client also supports:

• fit(...)
• bench(...)
• list_projections()
• get_projection(project_id)

Local Migration With Adapters

The local migration engine reads vectors from a source adapter, applies a Projection, and writes to a sink adapter. This path does not depend on hosted collection APIs.

from schift import Projection
from schift.migrate import migrate
from schift.adapters.file import NpyAdapter

projection = Projection.load("./projection-openai-to-gemini")

source = NpyAdapter("old_embeddings.npy")
sink = NpyAdapter("new_embeddings.npy")

result = migrate(
    source=source,
    sink=sink,
    projection=projection,
    batch_size=2048,
    dry_run=True,
)

print(result)

The same engine is exposed through WorkspaceClient().migrate.run(...).

Built-in Adapters

NumPy files

from schift.adapters.file import NpyAdapter

adapter = NpyAdapter("embeddings.npy")

pgvector

Requires pip install "schift[postgres]".

from schift.adapters.pgvector import PgVectorAdapter

adapter = PgVectorAdapter(
    conninfo="postgresql://user:password@localhost/mydb",
    table="documents",
    embedding_column="embedding",
    id_column="id",
)

Qdrant

Requires pip install "schift[qdrant]".

from schift.adapters.qdrant import QdrantAdapter

adapter = QdrantAdapter(
    url="http://localhost:6333",
    collection="documents",
    api_key=None,
)

Registry Helpers

from schift.adapters import get_adapter, list_adapters

print(list_adapters())

adapter = get_adapter(
    {
        "type": "npy",
        "path": "embeddings.npy",
    }
)

Errors

Both client styles raise exceptions from schift.client:

• AuthError
• QuotaError
• SDKError

Example:

from schift import WorkspaceClient
from schift import AuthError, QuotaError, SDKError

try:
    with WorkspaceClient() as client:
        client.catalog.list()
except AuthError:
    ...
except QuotaError:
    ...
except SDKError:
    ...

Development

From clients/sdk/python/:

python -m pip install -e ".[dev]"
python -m ruff check .
python -m pytest

Source Layout

clients/sdk/python/
├── schift/
│   ├── schift_client.py   # modular SDK entry point
│   ├── client.py          # legacy projection client
│   ├── projection.py      # local projection object
│   ├── adapters/          # npy, pgvector, qdrant
│   ├── workflow.py         # workflow CRUD, blocks, edges, YAML, templates
│   └── *.py               # catalog, embed, routing, db, query, rerank, usage
└── docs/                  # request/response contracts and schemas