CrewAI CouchbaseFTSVectorSearchTool：为 Agent 接入 Couchbase 向量语义检索的完整实践

本文以 CrewAI 工具库中的 CouchbaseFTSVectorSearchTool 官方说明文档为核心，完整梳理其安装方式、环境准备（集群、Bucket/Scope/Collection、向量索引）、全部构造参数、将工具挂载到 Agent 的示例代码，并结合 工具实现源码 与 测试用例，深入讲解该工具的初始化校验机制、Scoped 与集群级索引两条检索路径，以及向量查询在底层的真实调用链。读完本文，你可以直接在自己的 CrewAI 项目中把 Couchbase 的 FTS 向量搜索能力作为 RAG 工具交给 Agent 使用。

工具定位：面向 Couchbase 的语义搜索工具

Couchbase 是一款具备向量搜索能力的 NoSQL 数据库，支持存储与查询向量嵌入（Embedding）。CouchbaseFTSVectorSearchTool 是 CrewAI 工具库（crewai-tools 包）中专为 Couchbase 打造的工具，其目标非常聚焦：给定一段查询文本，在 Couchbase 向量索引中做语义检索，找出与查询语义相似的文档，从而让 Agent 在回答问题时可以“先检索、再作答”。

从源码结构看，该工具继承自 crewai.tools 的 BaseTool，并遵循 CrewAI 工具的通用约定：

• 工具名固定为 "CouchbaseFTSVectorSearchTool"；
• 工具描述为 "A tool to search the Couchbase database for relevant information on internal documents."，该描述会进入 LLM 的工具说明中，直接影响 Agent 何时决定调用它；
• 输入 Schema 只有一个 query 字段（CouchbaseToolSchema），且在 Schema 定义 中特别注明“Pass only the query, not the question”，即要求模型传入提炼后的检索词而非原始问题；
• 对 cluster 与 embedding_function 两个非 Pydantic 原生类型的字段使用了 SkipValidation 包装，以支持传入任意外部对象（model_config = ConfigDict(arbitrary_types_allowed=True)）。

该工具已注册在包的两级导出中：crewai_tools 顶层 __init__ 与 tools 子包 __init__ 均导出了 CouchbaseFTSVectorSearchTool，因此可以直接 from crewai_tools import CouchbaseFTSVectorSearchTool。

安装与依赖

按照 README 说明，安装 crewai_tools 包的命令为：

uv pip install 'crewai[tools]'

此外，Couchbase 官方 Python SDK 是一个可选依赖，以 extras 的形式声明在 crewai-tools 的 pyproject.toml 中：

couchbase = [
    "couchbase>=4.6.0",
]

也就是说，SDK 版本要求为 couchbase>=4.6.0。源码中对 SDK 采用了“优雅降级”的导入方式（导入保护块）：先尝试导入 couchbase.cluster、couchbase.options、couchbase.search、couchbase.vector_search 等模块，成功则置 COUCHBASE_AVAILABLE = True，否则回退为 Any 占位，保证未安装 SDK 的环境中 crewai_tools 包本身仍可正常导入。真正的缺依赖处理发生在实例化阶段（见下文“初始化时的依赖检查”）。

环境准备：集群、集合与向量索引

在实例化工具之前，你需要先准备好 Couchbase 侧的资源，README 给出的准备步骤如下：

1. 准备一个 Couchbase 集群，两种方式任选其一：
   • 在 Couchbase Capella（Couchbase 的云数据库服务）上创建集群；
   • 部署一个本地 Couchbase Server。
2. 在集群上创建 bucket、scope 和 collection，随后按照 Couchbase Python SDK 的入门文档创建 Cluster 对象并把文档写入 collection。
3. 创建向量搜索索引（Vector Search Index），Capella 与本地 Server 各有对应的创建流程文档。
4. 保证索引的 Dimension 与嵌入模型匹配。例如 OpenAI 的 text-embedding-3-small 模型维度为 1536，则索引的 Dimension 字段必须配置为 1536。维度不一致是向量检索最常见的坑之一，README 对此做了显式提醒。

需要特别说明的是：这个工具只负责“查询”一侧。文档入库（embedding 计算、写入 collection、建立索引）不在本工具职责范围内，需要你在创建工具前用 Couchbase SDK 或其他管道完成。

完整参数说明

README 中 Arguments 一节列出的全部构造参数如下，结合 源码字段定义 补充了类型与默认值信息：

参数	类型	必填	默认值	说明
cluster	couchbase.cluster.Cluster	是	—	已初始化的 Couchbase Cluster 实例，连接到你目标的 Couchbase Server
bucket_name	str	是	—	要检索的 Couchbase bucket 名称
scope_name	str	是	—	bucket 内 scope 的名称
collection_name	str	是	—	scope 内 collection 的名称
index_name	str	是	—	搜索（向量）索引的名称
embedding_function	Callable[[str], list[float]]	是	—	将字符串转为浮点数列表（向量）的嵌入函数，用于在检索前对查询做嵌入
embedding_key	str | None	否	"embedding"	搜索索引中存放向量的字段名
scoped_index	bool	否	True	索引是否为 scope 级索引（True）还是集群级索引（False）
limit	int | None	否	3	返回的搜索结果最大条数

参数要点：

• embedding_function 的签名约定是“字符串进、浮点数列表出”，测试中的 mock 也按此约定实现：mock_embedding_function 直接返回 [0.1] * 10 这样的定长向量。
• embedding_key 必须与建索引时指定的向量字段名一致，默认为 'embedding'；如果你的文档中向量字段叫别的名字，务必显式传入。
• scoped_index 决定了后续检索走 scope.search 还是 cluster.search 路径，源码中初始化校验与 _run 执行两条路径都以此为分支条件（详见下文）。
• limit 在源码中被使用了两次：既传给 VectorQuery（构造处），又传给 SearchOptions(limit=self.limit, fields=["*"])，两处保持同一上限。

使用示例：把工具交给 Agent

README 给出的完整示例如下（保留原文结构，补充注释）：

from crewai_tools import CouchbaseFTSVectorSearchTool

# 从 Couchbase SDK 实例化一个 Cluster 对象
# （假设已按 SDK 入门文档完成 cluster = Cluster(...) 并写入文档、建好向量索引）

tool = CouchbaseFTSVectorSearchTool(
    cluster=cluster,                # Couchbase SDK 的 Cluster 实例
    collection_name="collection",   # scope 内的 collection 名
    scope_name="scope",            # bucket 内的 scope 名
    bucket_name="bucket",           # bucket 名
    index_name="index",            # 向量搜索索引名
    embedding_function=embed_fn    # str -> list[float] 的嵌入函数
)

# 将工具加入 Agent
rag_agent = Agent(
    name="rag_agent",
    role="You are a helpful assistant that can answer questions with the help of the CouchbaseFTSVectorSearchTool.",
    llm="gpt-4o-mini",
    tools=[tool],
)

这个示例展示了标准的 RAG 接入方式：工具实例化后挂到 Agent.tools，Agent 在推理过程中判断需要检索内部文档时，会自动调用该工具，工具返回 JSON 格式的检索结果，Agent 再基于结果组织回答。

初始化时的深度校验：fail-fast 设计

该工具最值得在源码层面学习的一点，是实例化阶段就完成了一整套“fail-fast”校验，而不是把错误推迟到第一次检索时才暴露。构造函数 在 couchbase SDK 可用时依次执行：

1. 连接解析：调用 cluster.bucket(bucket_name) → bucket.scope(scope_name) → scope.collection(collection_name)，缓存 _bucket、_scope、_collection 三个对象。任一环节异常会抛出 ValueError: "Error connecting to couchbase. Please check the connection and credentials"；
2. Bucket 存在性检查（_check_bucket_exists）：通过 bucket_manager.get_bucket(self.bucket_name) 探测，失败则抛出 Bucket {name} does not exist. Please create the bucket before searching.；
3. Scope/Collection 存在性检查（_check_scope_and_collection_exists）：遍历 bucket 下所有 scope 及其 collection，若 scope_name 或 collection_name 不在其中，分别抛出对应的 ValueError；
4. 索引存在性检查（_check_index_exists）：按 scoped_index 分两条路径——
   • scoped_index=True：从 scope.search_indexes().get_all_indexes() 取索引名列表；
   • scoped_index=False：先确认 cluster 已提供，再从 cluster.search_indexes().get_all_indexes() 取集群级索引名列表；
   • 索引名不在列表中则抛出 Index {name} does not exist. Please create the index before searching.。

这些错误分支全部有对应的测试覆盖，见 couchbase_tool_test.py：

• test_check_bucket_exists_fail：bucket 不存在时抛 Bucket test_bucket does not exist.；
• test_check_scope_exists_fail 与 test_check_collection_exists_fail：scope / collection 名错误时抛出匹配错误；
• test_check_index_exists_fail_scoped 与 test_check_index_exists_fail_global：分别验证 scope 级索引与集群级索引缺失时的报错；
• test_initialization_missing_required_args：逐个移除 6 个必填参数，验证每一种缺失都会触发 ValueError。

缺少 SDK 时的自动安装提示

当 COUCHBASE_AVAILABLE 为 False（即运行环境没有安装 couchbase 包）时，构造函数不会静默失败，而是进入交互式补救分支：

else:
    import click

    if click.confirm(
        "The 'couchbase' package is required to use the CouchbaseFTSVectorSearchTool. "
        "Would you like to install it?"
    ):
        import subprocess

        subprocess.run(["uv", "add", "couchbase"], check=True)
    else:
        raise ImportError(
            "The 'couchbase' package is required to use the CouchbaseFTSVectorSearchTool. "
            "Please install it with: uv add couchbase"
        )

即：先询问用户是否要安装，确认后自动执行 uv add couchbase；拒绝则抛出 ImportError 并附带安装命令提示。这一行为由 test_initialization_couchbase_unavailable 验证——测试中断言 click.confirm 被调用了一次，且用户拒绝时抛出匹配 The 'couchbase' package is required 的 ImportError。

检索执行流程：_run 的底层调用链

当 Agent 传入 query 后，_run 方法 按如下链路执行一次向量检索：

1. 查询嵌入：query_embedding = self.embedding_function(query)，把你配置的嵌入函数应用于查询串；
2. 构造向量查询：

   search_req = search.SearchRequest.create(
       VectorSearch.from_vector_query(
           VectorQuery(self.embedding_key, query_embedding, self.limit)
       )
   )
   

   其中 VectorQuery 的三要素依次是向量字段名（embedding_key）、查询向量、条数上限（limit），这与测试中的断言一致——test_run_success_scoped_index 验证了 VectorQuery 确实以 (embedding_key, 嵌入结果, limit) 三个参数被调用；
3. 按索引类型选择检索入口：

   if self.scoped_index:
       search_iter = self._scope.search(self.index_name, search_req,
                                        SearchOptions(limit=self.limit, fields=fields))
   else:
       search_iter = self.cluster.search(self.index_name, search_req,
                                         SearchOptions(limit=self.limit, fields=fields))
   

   fields=["*"] 表示返回文档的全部字段；两条路径各有测试佐证——scope 级路径下 cluster.search 断言 assert_not_called，集群级路径（scoped_index=False）下 scope.search 断言 assert_not_called（分别见 L282 与 L346）；
4. 收集结果并序列化：遍历 search_iter.rows()，把每行的 row.fields 收集进列表，最终 json.dumps(json_response, indent=2) 返回缩进格式化的 JSON 字符串供 LLM 阅读。test_run_success_scoped_index 的结果断言 验证了返回内容包含各文档字段且整体是合法 JSON 数组；
5. 异常兜底：检索过程中的任何异常不会向上抛出，而是被捕获并返回字符串 "Search failed with error: {e}"。这意味着工具永远不会让 Agent 崩溃，但调用方应检查结果内容判断是否真正检索成功——这是使用时的一个实际注意点。

Scoped 索引与集群级索引：scoped_index 的实际影响

scoped_index 参数在工具内贯穿始终，其影响集中在三处：

阶段	scoped_index=True（默认）	scoped_index=False
索引校验	在 scope.search_indexes() 中查找索引名	在 cluster.search_indexes() 中查找索引名
检索入口	scope.search(index_name, ...)	cluster.search(index_name, ...)
测试覆盖	scoped 成功用例 + scoped 索引缺失用例	global 成功用例 + global 索引缺失用例

从源码结构看，这一区分对应 Couchbase 中 scope 级 Search 索引与集群级（传统 FTS 命名空间）索引两种索引形态。如果你的向量索引创建在某个 scope 之下（Couchbase 较新版本建 scope 级向量索引的常见形态），保持默认 True 即可；若索引建在集群层，需要显式传 scoped_index=False，否则初始化阶段就会因“索引不存在”而报 ValueError。

适用前提与使用限制

• 版本与依赖前提：需要 crewai-tools 提供 CouchbaseFTSVectorSearchTool 导出的版本，并安装 couchbase>=4.6.0 SDK（见 pyproject.toml 的 extras 定义）。
• 数据准备责任外置：bucket/scope/collection、文档写入、向量索引创建均需在工具之外完成；索引 Dimension 必须与 embedding_function 所用嵌入模型的输出维度一致。
• 工具语义边界：该工具只做“查询—嵌入—向量检索—返回 JSON”这一件事，不具备写入、删除或管理索引的能力；collection 对象虽然在初始化时被缓存，但 _run 流程中只走 Search API，不直接读写 collection 文档。
• 失败可见性：检索失败时返回的是错误描述字符串而非异常（L227-L228），自动化流程中建议对返回值做前缀判断或结构化解析。

小结

CouchbaseFTSVectorSearchTool 是 CrewAI 将 Couchbase FTS 向量搜索封装为 Agent 可用工具的完整方案：文档层面它提供了从集群准备、索引配置到 Agent 集成的操作路径；实现层面它通过初始化期的 bucket/scope/collection/索引四重校验快速暴露配置错误，通过 scoped_index 分支适配两种索引形态，并以 JSON 字符串稳定地向 LLM 交付检索结果。如果你的 RAG 数据存放在 Couchbase 中，这套“README 操作 + 源码校验机制 + 测试用例佐证”的组合，就是把它接入 CrewAI Agent 的完整依据。