pyvespa Skill

Use this skill when the user is working with Vespa from Python — defining application packages programmatically, deploying to Docker or Vespa Cloud, feeding documents, querying, or using the query builder DSL.

Deeper references — load docs/package-api.md for the full configuration class reference (Schema, Field, RankProfile, etc.) and docs/application-api.md for the Vespa connection, feed, query, and deployment APIs.

Overview

pyvespa is the official Python client for Vespa.ai. Install with:

pip install pyvespa

Requires Python >=3.10, <3.14. Zero native dependencies for querying/feeding; Docker SDK needed only for local deployment.

Core Modules

Quick Start

from vespa.package import (
    ApplicationPackage,
    Schema,
    Document,
    Field,
    FieldSet,
    RankProfile,
    HNSW,
    FirstPhaseRanking,
)
from vespa.deployment import VespaDocker

# 1. Define the application
app_package = ApplicationPackage(name="myapp")

app_package.schema.add_fields(
    Field(name="title", type="string", indexing=["index", "summary"], index="enable-bm25"),
    Field(name="body", type="string", indexing=["index", "summary"], index="enable-bm25"),
    Field(
        name="embedding",
        type="tensor<float>(x[384])",
        indexing=["index", "attribute"],
        ann=HNSW(distance_metric="angular"),
    ),
)
app_package.schema.add_field_set(FieldSet(name="default", fields=["title", "body"]))
app_package.schema.add_rank_profile(
    RankProfile(
        name="hybrid",
        inputs=[("query(q)", "tensor<float>(x[384])")],
        first_phase="bm25(title) + bm25(body) + closeness(field, embedding)",
        match_features=["bm25(title)", "bm25(body)", "closeness(field, embedding)"],
    ),
)

# 2. Deploy locally
vespa_docker = VespaDocker(port=8080)
app = vespa_docker.deploy(app_package)

# 3. Feed a document
app.feed_data_point(
    schema="myapp",
    data_id="doc-1",
    fields={"title": "Vespa intro", "body": "Vespa is a serving engine", "embedding": [0.1] * 384},
)

# 4. Query
with app.syncio() as sess:
    response = sess.query(
        body={
            "yql": "select * from sources * where userQuery() or ({targetHits:10}nearestNeighbor(embedding, q))",
            "query": "serving engine",
            "ranking": "hybrid",
            "input.query(q)": [0.1] * 384,
        }
    )
    for hit in response.hits:
        print(hit["fields"]["title"], hit["relevance"])

ApplicationPackage

ApplicationPackage(
    name: str,                              # Lowercase, [a-z0-9], starts with letter, max 20 chars
    schema: Optional[List[Schema]] = None,  # Defaults to one Schema matching name
    stateless_model_evaluation: bool = False,
    components: Optional[List[Component]] = None,
    auth_clients: Optional[List[AuthClient]] = None,
)

When created with defaults, an ApplicationPackage auto-creates a single Schema and Document with the same name. Access via app_package.schema.

Key methods:

• app_package.schema — returns the single schema (asserts exactly one)
• app_package.get_schema(name) — get a schema by name
• app_package.add_schema(schema) — add additional schemas
• app_package.to_files(root) — write to disk as a standard Vespa application directory
• app_package.to_zip() — serialize to a deployable zip (BytesIO)

Multiple Schemas

from vespa.package import Schema, Document

schema_a = Schema(name="article", document=Document())
schema_b = Schema(name="comment", document=Document())

app_package = ApplicationPackage(
    name="myapp",
    schema=[schema_a, schema_b],
    create_schema_by_default=False,
)

Field

The most important configuration class.

Field(
    name: str,
    type: str,                  # "string", "int", "long", "float", "double", "bool",
                                # "tensor<float>(x[384])", "array<string>", "weightedset<string>", etc.
    indexing: list | tuple | str,  # ["index", "summary"] → pipe-separated; tuple → multiline block
    index: str | dict = None,     # "enable-bm25", or dict for predicate arity etc.
    attribute: list = None,       # ["fast-search", "paged"]
    ann: HNSW = None,             # For tensor fields with ANN index
    match: list = None,           # [("exact", None)] or ["word"]
    bolding: True = None,
    summary: Summary = None,
    rank: str = None,             # "filter" for filter-only fields
    struct_fields: list = None,
)

Indexing directive formats:

• ["index", "summary"] → indexing: index | summary
• ("input title | index | summary",) → multiline indexing { ... } block (used for embed expressions)

Common field patterns:

# Text field with BM25
Field(name="title", type="string", indexing=["index", "summary"], index="enable-bm25")

# Attribute for filtering/sorting
Field(name="price", type="float", indexing=["attribute", "summary"])

# Fast-search attribute
Field(name="category", type="string", indexing=["attribute", "summary"], attribute=["fast-search"])

# Tensor with HNSW
Field(
    name="embedding",
    type="tensor<float>(x[384])",
    indexing=["index", "attribute"],
    ann=HNSW(distance_metric="angular", max_links_per_node=16, neighbors_to_explore_at_insert=200),
)

# Embedder integration (multiline indexing)
Field(
    name="embedding",
    type="tensor<float>(x[384])",
    indexing=('input title . " " . input body | embed e5 | index | attribute',),
    ann=HNSW(distance_metric="angular"),
)

# Array field
Field(name="tags", type="array<string>", indexing=["attribute", "summary"])

# Weighted set
Field(name="labels", type="weightedset<string>", indexing=["attribute", "summary"])

# Filter-only field
Field(name="internal_id", type="long", indexing=["attribute"], rank="filter")

HNSW

HNSW(
    distance_metric: str = "euclidean",  # euclidean|angular|dotproduct|prenormalized-angular|hamming|geodegrees
    max_links_per_node: int = 16,
    neighbors_to_explore_at_insert: int = 200,
)

Use angular or prenormalized-angular for cosine similarity. Use dotproduct for maximum inner product.

RankProfile

RankProfile(
    name: str,
    first_phase: str | FirstPhaseRanking = None,
    second_phase: SecondPhaseRanking = None,
    global_phase: GlobalPhaseRanking = None,
    inherits: str = None,
    inputs: list = None,           # [("query(q)", "tensor<float>(x[384])")]
    functions: list = None,        # [Function(name, expression)]
    constants: dict = None,
    summary_features: list = None,
    match_features: list = None,
)

Ranking phases:

from vespa.package import (
    RankProfile,
    FirstPhaseRanking,
    SecondPhaseRanking,
    GlobalPhaseRanking,
    Function,
)

RankProfile(
    name="hybrid",
    inputs=[("query(q)", "tensor<float>(x[384])")],
    functions=[
        Function(name="text_score", expression="bm25(title) + 0.5 * bm25(body)"),
        Function(name="vector_score", expression="closeness(field, embedding)"),
    ],
    first_phase=FirstPhaseRanking(expression="text_score + vector_score", keep_rank_count=1000),
    second_phase=SecondPhaseRanking(expression="text_score * 0.4 + vector_score * 0.6", rerank_count=100),
    match_features=["bm25(title)", "bm25(body)", "closeness(field, embedding)"],
)

Deployment

Two deployment targets: VespaDocker (local, needs 4 GB Docker memory) and VespaCloud (tenant + application + auth mode). Typical flow: vespa_docker.deploy(app_package) returns a Vespa instance you use for feeding/querying.

For Docker reconnect, Vespa Cloud token/mTLS auth modes, and get_application(endpoint_type=...), load docs/deployment-auth.md.

Feeding Documents

Single Document

with app.syncio() as sess:
    response = sess.feed_data_point(
        schema="myschema",
        data_id="doc-1",
        fields={"title": "Hello", "count": 42},
    )
    assert response.is_successful()

Bulk Feed

docs = [{"id": str(i), "fields": {"title": f"Doc {i}"}} for i in range(10000)]


def callback(response, doc_id):
    if not response.is_successful():
        print(f"Failed: {doc_id} — {response.status_code}")


# Sync (thread pool)
app.feed_iterable(docs, schema="myschema", callback=callback, max_workers=8)

# Async (HTTP/2, preferred for throughput)
app.feed_async_iterable(docs, schema="myschema", callback=callback, max_workers=64)

Feed data format — each dict must have "id" (str) and "fields" (dict). Optional "groupname".

Update and Delete

with app.syncio() as sess:
    # Partial update
    sess.update_data(schema="myschema", data_id="doc-1", fields={"count": {"increment": 1}}, create=True)

    # Delete
    sess.delete_data(schema="myschema", data_id="doc-1")

# Bulk update/delete
app.feed_iterable(docs, schema="myschema", operation_type="update")
app.feed_iterable(ids, schema="myschema", operation_type="delete")

Querying

Direct Query

with app.syncio() as sess:
    response = sess.query(
        body={
            "yql": "select * from sources * where title contains 'vespa'",
            "ranking": "bm25",
            "hits": 10,
        }
    )
    print(response.hits)  # List of hit dicts
    print(response.number_documents_retrieved)  # Total matches

For the pyvespa Query Builder DSL (vespa.querybuilder), batch queries (app.query_many), grouping DSL, and document visiting (app.visit), load docs/query-builder-dsl.md.

Gotchas

1. App name restrictions — ApplicationPackage(name=...) must be lowercase [a-z0-9], start with letter, max 20 chars. No hyphens or underscores.
2. Tuple vs list for indexing — use a tuple ("input title | embed e5 | index",) for multiline/embed expressions; use a list ["index", "summary"] for pipe-separated.
3. Namespace defaults to schema name — if you don't pass namespace to feed/query methods, it defaults to the schema parameter value.
4. Docker memory — allocate at least 4 GB to Docker for local deployments.
5. feed_async_iterable vs feed_iterable — async is better for throughput (HTTP/2 multiplexing over 1 connection); sync uses a thread pool with multiple connections.
6. Context managers — always use app.syncio() or app.asyncio() for production code to manage connections properly.
7. is_successful() — use this (not the deprecated misspelled is_successfull()).
8. Cloud token auth — set VESPA_CLOUD_SECRET_TOKEN env var and pass auth_client_token_id to VespaCloud, then use endpoint_type="token" in get_application().

For deeper detail, load docs/package-api.md, docs/application-api.md, docs/deployment-auth.md, or docs/query-builder-dsl.md from this skill's directory as needed. External docs: https://vespa-engine.github.io/pyvespa/