Skip to main content

​
Setup

​
Local Setup (Docker)

Run Couchbase locally using Docker: 
docker run -d --name couchbase-server \
  -p 8091-8096:8091-8096 \
  -p 11210:11210 \
  -e COUCHBASE_ADMINISTRATOR_USERNAME=Administrator \
  -e COUCHBASE_ADMINISTRATOR_PASSWORD=password \
  couchbase:latest
  1. Access the Couchbase UI at: http://localhost:8091
  2. Login with username: Administrator and password: password
  3. 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

pip install couchbase

​
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
from couchbase.management.search import SearchIndex

# 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,
        overwrite=True
    ),
)

# Load the knowledge base
knowledge_base.add_content(
    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)

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
from couchbase.management.search import SearchIndex

# 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,
        overwrite=True
    ),
)

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

async def run_agent():
    await knowledge_base.add_content_async(
      url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
    )
    time.sleep(5)  # Wait for the vector index to sync with KV
    await agent.aprint_response("How to make Thai curry?", markdown=True)

if __name__ == "__main__":
    asyncio.run(run_agent())
Use aload() 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

ParameterTypeDescriptionDefault
bucket_namestrName of the Couchbase bucketRequired
scope_namestrName of the scope within the bucketRequired
collection_namestrName of the collection within the scopeRequired
couchbase_connection_stringstrCouchbase cluster connection stringRequired
cluster_optionsClusterOptionsOptions for configuring the Couchbase cluster connectionRequired
search_indexUnion[str, SearchIndex]Search index configuration, either as index name or SearchIndex definitionRequired
embedderEmbedderEmbedder instance for generating embeddingsOpenAIEmbedder()
overwriteboolWhether to overwrite existing collectionFalse
is_global_level_indexboolWhether the search index is at global levelFalse
wait_until_index_readyOptional[float]Time in seconds to wait until the index is readyNone
batch_limitintMaximum number of documents to process in a single batch (applies to both sync and async operations)500

​
Developer Resources