Couchbase Agent Knowledge

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="<your-openai-api-key>"

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.embedder.openai import OpenAIEmbedder
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
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 = PDFUrlKnowledgeBase(
    urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
    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.load(recreate=True)

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

# Create and use the agent
agent = Agent(knowledge=knowledge_base, show_tool_calls=True)
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.embedder.openai import OpenAIEmbedder
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
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 = PDFUrlKnowledgeBase(
    urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
    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, show_tool_calls=True)

async def run_agent():
    await knowledge_base.aload(recreate=True)
    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__":
    # Comment out after the first run
    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

Developer Resources

• View Cookbook (Sync)
• View Cookbook (Async)