Vector Search Ops

vector-search-ops — Vector Search Operations

Covers day-to-day operations for the resources/vector_search.yml component scaffolded by AgentOps Stacks: endpoint health, index sync, retriever testing, and DLT pipeline changes.

When to use

• "my retriever isn't returning results"
• "how do I re-sync the index after adding documents?"
• "check vector search status"
• "the DLT pipeline failed"
• "test the retriever"
• "how do I update the embedding source?"

Key resources in the project

The index is named <catalog>.<schema>.<project>_vs_index and syncs from <catalog>.<schema>.<project>_chunked_docs.

Operations

1. Check endpoint and index status

from databricks.sdk import WorkspaceClient
w = WorkspaceClient()

# Endpoint
ep = w.vector_search_endpoints.get_endpoint("<project>_vs_endpoint")
print(ep.endpoint_status)  # ONLINE / OFFLINE / PROVISIONING

# Index
idx = w.vector_search_indexes.get_index(
    index_name="<catalog>.<schema>.<project>_vs_index"
)
print(idx.status)           # ONLINE / PROVISIONING / FAILED
print(idx.index_url)        # endpoint URL

Or via CLI:

databricks vector-search endpoints get --name <project>_vs_endpoint
databricks vector-search indexes get-index --index-name <catalog>.<schema>.<project>_vs_index

2. Trigger a manual sync

The index uses pipeline_type: TRIGGERED — it does NOT auto-sync on table writes. Trigger it explicitly after new documents are ingested:

from databricks.sdk import WorkspaceClient
w = WorkspaceClient()

w.vector_search_indexes.sync_index(
    index_name="<catalog>.<schema>.<project>_vs_index"
)
print("Sync triggered — check status with get_index()")

Wait for sync to complete before testing:

import time
while True:
    idx = w.vector_search_indexes.get_index(index_name="<catalog>.<schema>.<project>_vs_index")
    if idx.status.detailed_state in ("ONLINE", "ONLINE_NO_PENDING_UPDATE"):
        break
    print(f"  {idx.status.detailed_state} — waiting...")
    time.sleep(15)

3. Test the retriever

Run a quick retrieval test without starting the full agent:

from databricks.sdk import WorkspaceClient
from databricks_langchain import DatabricksVectorSearch
import os

w = WorkspaceClient()
vs_index = "<catalog>.<schema>.<project>_vs_index"

retriever = DatabricksVectorSearch(
    index_name=vs_index,
    columns=["content", "doc_uri"],
    workspace_client=w,
).as_retriever(search_kwargs={"k": 5})

docs = retriever.invoke("your test query here")
for doc in docs:
    print(f"[{doc.metadata.get('doc_uri', '?')}]\n{doc.page_content[:200]}\n")

4. Ingest new documents

New documents go into the UC Volume at /Volumes/<catalog>/<schema>/artifacts/raw_docs/.

After uploading:

1. Run the DLT pipeline (data_pipeline.py) to update the _chunked_docs table
2. Trigger the VS index sync (step 2 above)

# Upload a file to the volume
databricks fs cp my_doc.pdf dbfs:/Volumes/<catalog>/<schema>/artifacts/raw_docs/

# Run DLT pipeline via bundle
databricks bundle run -t dev data_pipeline

# Then trigger sync (Python snippet above or SDK)

5. Update the DLT pipeline (chunking logic)

Edit src/components/retriever/data_pipeline.py. The two DLT tables are:

• <project>_raw_docs — reads from the UC Volume
• <project>_chunked_docs — apply chunking here

Common changes:

• Different source format — swap cloudFiles.format in raw_docs()
• Custom chunking — replace the body of chunked_docs() (LangChain splitters, unstructured.io, etc.)
• Add metadata columns — extend .select() and columns_to_sync in vector_search.yml

After editing, redeploy:

databricks bundle deploy -t dev
databricks bundle run -t dev data_pipeline
# then trigger sync

6. Change the embedding model

In resources/vector_search.yml, update the embedding_model_endpoint variable or the embedding_model_endpoint_name field on the index. The built-in default is databricks-gte-large-en.

Note: Changing the embedding model requires rebuilding the index from scratch. Delete and recreate the index (or undeploy + redeploy the bundle resource).

databricks bundle destroy -t dev  # removes VS endpoint + index
databricks bundle deploy -t dev   # recreates with new model
databricks bundle run -t dev data_pipeline
# then sync

7. Increase k (number of retrieved chunks)

In src/agents/<name>/graph.py, get_relevant_chunks(query, k=5) — increase k for broader retrieval, decrease for more precise context.

Also adjustable per-call:

retriever.as_retriever(search_kwargs={"k": 10})

Troubleshooting

Environment variables (set in app.yaml)

env:
  - name: CATALOG
    value: <catalog>
  - name: SCHEMA
    value: <schema>

The retriever in graph.py reads os.environ["CATALOG"] and os.environ["SCHEMA"] to build the index name at runtime.