Couchbase
Define a Couchbase FTS vector index in code, insert a PDF, and search it with an agent.
Code
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.auth import PasswordAuthenticator
from couchbase.management.search import SearchIndex
from couchbase.options import ClusterOptions, KnownConfigProfiles
# 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)
# Define the vector search index
search_index = SearchIndex(
name="vector_search",
source_type="gocbcore",
idx_type="fulltext-index",
source_name="recipe_bucket",
plan_params={"index_partitions": 1, "num_replicas": 0},
params={
"doc_config": {
"docid_prefix_delim": "",
"docid_regexp": "",
"mode": "scope.collection.type_field",
"type_field": "type",
},
"mapping": {
"default_analyzer": "standard",
"default_datetime_parser": "dateTimeOptional",
"index_dynamic": True,
"store_dynamic": True,
"default_mapping": {"dynamic": True, "enabled": False},
"types": {
"recipe_scope.recipes": {
"dynamic": False,
"enabled": True,
"properties": {
"content": {
"enabled": True,
"fields": [
{
"docvalues": True,
"include_in_all": False,
"include_term_vectors": False,
"index": True,
"name": "content",
"store": True,
"type": "text",
}
],
},
"embedding": {
"enabled": True,
"dynamic": False,
"fields": [
{
"vector_index_optimized_for": "recall",
"docvalues": True,
"dims": 1536,
"include_in_all": False,
"include_term_vectors": False,
"index": True,
"name": "embedding",
"similarity": "dot_product",
"store": True,
"type": "vector",
}
],
},
"meta": {
"dynamic": True,
"enabled": True,
"properties": {
"name": {
"enabled": True,
"fields": [
{
"docvalues": True,
"include_in_all": False,
"include_term_vectors": False,
"index": True,
"name": "name",
"store": True,
"analyzer": "keyword",
"type": "text",
}
],
}
},
},
},
}
},
},
},
)
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=search_index,
embedder=OpenAIEmbedder(
dimensions=1536,
),
wait_until_index_ready=60,
overwrite=True,
)
knowledge = Knowledge(
name="Couchbase Knowledge Base",
description="This is a knowledge base that uses a Couchbase DB",
vector_db=vector_db,
)
knowledge.insert(
name="Recipes",
url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
metadata={"doc_type": "recipe_book"},
)
# Wait for the vector index to sync with KV
time.sleep(20)
agent = Agent(
knowledge=knowledge,
search_knowledge=True,
read_chat_history=True,
)
agent.print_response("List down the ingredients to make Massaman Gai", markdown=True)
vector_db.delete_by_name("Recipes")
# or
vector_db.delete_by_metadata({"doc_type": "recipe_book"})Usage
Set up your virtual environment
uv venv --python 3.12
source .venv/bin/activateStart Couchbase
docker run -d --name couchbase-server \
-p 8091-8096:8091-8096 \
-p 11210:11210 \
couchbase:latestOpen http://localhost:8091, choose Setup New Cluster, and create the Administrator account with password password. Enable Data, Search, and Query services (plus Index if creating SQL indexes), following the cluster initialization guide. Then create:
- Bucket:
recipe_bucket - Scope:
recipe_scope - Collection:
recipes
Install dependencies
uv pip install -U couchbase pypdf openai agnoSet environment variables
export COUCHBASE_USER="Administrator"
export COUCHBASE_PASSWORD="password"
export COUCHBASE_CONNECTION_STRING="couchbase://localhost"
export OPENAI_API_KEY=xxxRun Agent
python couchbase_db.py