Couchbase Agent 知识库

设置

本地设置（Docker）

使用 Docker 在本地运行 Couchbase：

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. 访问 Couchbase UI：http://localhost:8091
2. 使用用户名：Administrator 和密码：password 登录
3. 创建一个名为 recipe_bucket 的 bucket，一个名为 recipe_scope 的 scope，以及一个名为 recipes 的 collection

托管设置（Capella）

对于托管集群，请使用 Couchbase Capella：

• 遵循 Capella 的 UI 来创建数据库、bucket、scope 和 collection

环境变量

设置您的环境变量：

export COUCHBASE_USER="Administrator"
export COUCHBASE_PASSWORD="password" 
export COUCHBASE_CONNECTION_STRING="couchbase://localhost"
export OPENAI_API_KEY="<your-openai-api-key>"

对于 Capella，将 COUCHBASE_CONNECTION_STRING 设置为您的 Capella 连接字符串。

安装依赖

pip install couchbase

示例

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 连接设置
username = os.getenv("COUCHBASE_USER")
password = os.getenv("COUCHBASE_PASSWORD")
connection_string = os.getenv("COUCHBASE_CONNECTION_STRING")

# 创建带有身份验证的集群选项
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
    ),
)

# 加载知识库
knowledge_base.load(recreate=True)

# 等待向量索引与 KV 同步
time.sleep(20)

# 创建并使用 Agent
agent = Agent(knowledge=knowledge_base, show_tool_calls=True)
agent.print_response("How to make Thai curry?", markdown=True)

异步支持 ⚡

Couchbase 也支持异步操作，能够实现并发并带来更好的性能。

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 连接设置
username = os.getenv("COUCHBASE_USER")
password = os.getenv("COUCHBASE_PASSWORD")
connection_string = os.getenv("COUCHBASE_CONNECTION_STRING")

# 创建带有身份验证的集群选项
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
    ),
)

# 创建并使用 Agent
agent = Agent(knowledge=knowledge_base, show_tool_calls=True)

async def run_agent():
    await knowledge_base.aload(recreate=True)
    time.sleep(5)  # 等待向量索引与 KV 同步
    await agent.aprint_response("How to make Thai curry?", markdown=True)

if __name__ == "__main__":
    # 首次运行后注释掉
    asyncio.run(run_agent())

在 asyncio.run() 中使用 aload() 和 aprint_response() 方法，以实现高吞吐量应用中的非阻塞操作。

关键配置说明

连接配置文件

对于本地和云部署，请使用 KnownConfigProfiles.WanDevelopment 来适当处理网络延迟和超时。

Couchbase 参数

开发者资源

• 查看 Cookbook (Sync)
• 查看 Cookbook (Async)