Couchbase Vector Database

Setup

Local Setup (Docker)

Run Couchbase locally using Docker:

docker run -d --name couchbase-server \
  -p 8091-8096:8091-8096 \
  -p 11210:11210 \
  couchbase:latest

Access the Couchbase UI at: http://localhost:8091
Choose Setup New Cluster, create the Administrator account with password password, and enable Data, Search, and Query services (plus Index if creating SQL indexes). Follow the cluster initialization guide.
Create a bucket named recipe_bucket, a scope recipe_scope, and a collection recipes

Managed Setup (Capella)

For a managed cluster, use Couchbase Capella:

Follow Capella's UI to create a database, bucket, scope, and collection

Environment Variables

Set up your environment variables:

export COUCHBASE_USER="Administrator"
export COUCHBASE_PASSWORD="password"
export COUCHBASE_CONNECTION_STRING="couchbase://localhost"
export OPENAI_API_KEY=xxx

For Capella, set COUCHBASE_CONNECTION_STRING to your Capella connection string.

Install Dependencies

uv pip install -U couchbase pypdf openai agno

Example

agent_with_knowledge.py

import os
import time
from agno.agent import Agent
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.couchbase import CouchbaseSearch
from couchbase.options import ClusterOptions, KnownConfigProfiles
from couchbase.auth import PasswordAuthenticator

# Couchbase connection settings
username = os.getenv("COUCHBASE_USER")
password = os.getenv("COUCHBASE_PASSWORD")
connection_string = os.getenv("COUCHBASE_CONNECTION_STRING")

# Create cluster options with authentication
auth = PasswordAuthenticator(username, password)
cluster_options = ClusterOptions(auth)
cluster_options.apply_profile(KnownConfigProfiles.WanDevelopment)

knowledge_base = Knowledge(
    vector_db=CouchbaseSearch(
        bucket_name="recipe_bucket",
        scope_name="recipe_scope",
        collection_name="recipes",
        couchbase_connection_string=connection_string,
        cluster_options=cluster_options,
        search_index="vector_search_fts_index",
        embedder=OpenAIEmbedder(
            id="text-embedding-3-large",
            dimensions=3072,
            api_key=os.getenv("OPENAI_API_KEY")
        ),
        wait_until_index_ready=60,
    ),
)

# Load the knowledge base
knowledge_base.insert(
    url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
)

# Wait for the vector index to sync with KV
time.sleep(20)

# Create and use the agent
agent = Agent(knowledge=knowledge_base)
agent.print_response("How to make Thai curry?", markdown=True)

Passing search_index as a string requires an FTS index with that name to already exist on the collection. To define and create the index in code, pass a SearchIndex object instead. See Couchbase usage.

Async Support ⚡

Couchbase also supports asynchronous operations, enabling concurrency and leading to better performance.

async_couchbase.py

import asyncio
import os
import time
from agno.agent import Agent
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.couchbase import CouchbaseSearch
from couchbase.options import ClusterOptions, KnownConfigProfiles
from couchbase.auth import PasswordAuthenticator

# Couchbase connection settings
username = os.getenv("COUCHBASE_USER")
password = os.getenv("COUCHBASE_PASSWORD")
connection_string = os.getenv("COUCHBASE_CONNECTION_STRING")

# Create cluster options with authentication
auth = PasswordAuthenticator(username, password)
cluster_options = ClusterOptions(auth)
cluster_options.apply_profile(KnownConfigProfiles.WanDevelopment)

knowledge_base = Knowledge(
    vector_db=CouchbaseSearch(
        bucket_name="recipe_bucket",
        scope_name="recipe_scope",
        collection_name="recipes",
        couchbase_connection_string=connection_string,
        cluster_options=cluster_options,
        search_index="vector_search_fts_index",
        embedder=OpenAIEmbedder(
            id="text-embedding-3-large",
            dimensions=3072,
            api_key=os.getenv("OPENAI_API_KEY")
        ),
        wait_until_index_ready=60,
    ),
)

# Create and use the agent
agent = Agent(knowledge=knowledge_base)

async def run_agent():
    await knowledge_base.ainsert(
      url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
    )
    await asyncio.sleep(5)  # Demo indexing delay; verify index readiness for your deployment
    await agent.aprint_response("How to make Thai curry?", markdown=True)

if __name__ == "__main__":
    asyncio.run(run_agent())

Use ainsert() and aprint_response() methods with asyncio.run() for non-blocking operations in high-throughput applications.

Key Configuration Notes

Connection Profiles

Use KnownConfigProfiles.WanDevelopment for both local and cloud deployments to handle network latency and timeouts appropriately.

Couchbase Params

Parameter	Type	Description	Default
`bucket_name`	`str`	Name of the Couchbase bucket	Required
`scope_name`	`str`	Name of the scope within the bucket	Required
`collection_name`	`str`	Name of the collection within the scope	Required
`couchbase_connection_string`	`str`	Couchbase cluster connection string	Required
`cluster_options`	`ClusterOptions`	Options for configuring the Couchbase cluster connection	Required
`search_index`	`Union[str, SearchIndex]`	Search index configuration, either as index name or SearchIndex definition	Required
`embedder`	`Embedder`	Embedder instance for generating embeddings	`OpenAIEmbedder()`
`overwrite`	`bool`	Whether to overwrite existing collection	`False`
`is_global_level_index`	`bool`	Whether the search index is at global level	`False`
`wait_until_index_ready`	`float`	Time in seconds to wait until the index is ready	`0`
`batch_limit`	`int`	Maximum number of documents to process in a single batch (applies to both sync and async operations)	`500`

Async Support ⚡

On this page