> ## Documentation Index
> Fetch the complete documentation index at: https://ikun.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Couchbase Agent 知识库

## 设置

### 本地设置（Docker）

使用 Docker 在本地运行 Couchbase：

```shell theme={null}
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](http://localhost:8091)
2. 使用用户名：`Administrator` 和密码：`password` 登录
3. 创建一个名为 `recipe_bucket` 的 bucket，一个名为 `recipe_scope` 的 scope，以及一个名为 `recipes` 的 collection

### 托管设置（Capella）

对于托管集群，请使用 [Couchbase Capella](https://cloud.couchbase.com/)：

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

### 环境变量

设置您的环境变量：

```shell theme={null}
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 连接字符串。

### 安装依赖

```shell theme={null}
pip install couchbase
```

## 示例

```python agent_with_knowledge.py theme={null}
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)
```

<Card title="异步支持 ⚡">
  <div className="mt-2">
    <p>
      Couchbase 也支持异步操作，能够实现并发并带来更好的性能。
    </p>

    ```python async_couchbase.py theme={null}
    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())
    ```

    <Tip className="mt-4">
      在 `asyncio.run()` 中使用 <code>aload()</code> 和 <code>aprint\_response()</code> 方法，以实现高吞吐量应用中的非阻塞操作。
    </Tip>
  </div>
</Card>

## 关键配置说明

### 连接配置文件

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

## Couchbase 参数

<Snippet file="vectordb_couchbase_params.mdx" />

## 开发者资源

* 查看 [Cookbook (Sync)](https://github.com/agno-agi/agno/blob/main/cookbook/agent_concepts/knowledge/vector_dbs/couchbase/couchbase_db.py)
* 查看 [Cookbook (Async)](https://github.com/agno-agi/agno/blob/main/cookbook/agent_concepts/knowledge/vector_dbs/couchbase/async_couchbase_db.py)