使用 crewAI DB2VectorSearchTool：在 IBM DB2 中落地原生向量语义检索

使用 crewAI DB2VectorSearchTool：在 IBM DB2 中落地原生向量语义检索

【免费下载链接】crewAIFramework for orchestrating role-playing, autonomous AI agents. By fostering collaborative intelligence, CrewAI empowers agents to work together seamlessly, tackling complex tasks.项目地址: https://gitcode.com/GitHub_Trending/cr/crewAI

导读

本文围绕 crewAI Tools 提供的DB2VectorSearchTool，系统讲解如何让 CrewAI Agent 直接对IBM DB2 原生 VECTOR 数据类型执行语义检索。你将掌握该工具的安装配置、连接串与环境变量设定、OpenAI/自定义嵌入函数的选用、元数据过滤、距离度量选择与 JSON 结果解析，并通过源码级拆解理解其背后的 SQL 构造、SQL 注入防护与连接生命周期设计，最终能把它接入自己的 Crew 工作流。

该工具在仓库中以独立模块存在，核心实现位于 db2_search_tool.py，公开导出类为DB2VectorSearchTool，可直接从crewai_tools顶层导入。

一、工具定位：检索专用（retrieval-only）

DB2VectorSearchTool是 CrewAI 系列向量检索工具中的一员，专门承担从 DB2 中读取与查询最相似文档的职责。其功能边界如下：

• 生成查询文本的嵌入向量；
• 对 DB2 中的 VECTOR 列执行向量相似度搜索（使用VECTOR_DISTANCE）；
• 按需应用元数据过滤；
• 返回结构化的规范化 JSON 结果。

按照官方 README 的架构说明，该工具与 QdrantVectorSearchTool、WeaviateVectorSearchTool 采用同一套工具架构约定。特别需要注意的是：该工具只做检索，不做文档入库（ingestion）。向量数据的写入、批量嵌入、索引维护需要由独立的入库流程完成，这是使用前必须接受的前提。

二、安装与依赖

工具本体随crewai-tools发布，DB2 相关能力依赖两个可选第三方包。官方 README 给出的安装命令是：

uv add ibm_db openai

其中ibm_db用于建立与 DB2 的原生连接与执行 SQL，openai仅在走默认 OpenAI 嵌入路径时才需要（若完全使用自定义嵌入函数，可不必安装openai）。

从源码看，这两项依赖被登记在工具的元数据中（见 db2_search_tool.py#L86-L91）：

package_dependencies: list[str] = Field( default_factory=lambda: [ "ibm_db", "openai", # Optional openai is used for embeddings ] )

同时，tool.specs.json（约 L6034 起）同样记录了package_dependencies: ["ibm_db", "openai"]，便于自动化的依赖清单生成。

需要特别说明工具的“运行时动态导入（runtime dynamic imports）”特性：ibm_db、ibm_db_dbi乃至openai都不会在模块加载时被强制导入，而是在首次真正使用时才通过importlib惰性加载（见 db2_search_tool.py#L160-L172 的_resolve_db2_packages与 db2_search_tool.py#L211-L220 的_get_openai_client）。这带来一个直接收益：即使开发机尚未安装ibm_db，也可以安全地 import 并实例化该工具——只有在真正发起检索并连接 DB2 时才会报缺包错误。测试代码也充分利用了这一点：测试套件在 import 阶段预先注入ibm_db/ibm_db_dbi的桩模块，使全部用例无需真实 DB2 即可运行（见 test_db2_search_tool.py#L19-L50）。

三、环境变量与连接串

工具声明了两个环境变量（见 db2_search_tool.py#L93-L106）：

OPENAI_API_KEY=your_openai_key DB2_CONNECTION_STRING=DATABASE=TESTDB;HOSTNAME=localhost;PORT=50000;PROTOCOL=TCPIP;UID=db2user;PWD=password;

两者的角色并不相同：

• OPENAI_API_KEY：仅在采用默认 OpenAI 嵌入路径时需要。源码中若检测不到该环境变量且未提供自定义嵌入函数，会直接抛出"OPENAI_API_KEY environment variable is missing. Required for default embeddings."（见 db2_search_tool.py#L213-L217）。因此它被声明为“非必需（required=False）”，是因为存在custom_embedding_fn这一替代方案；
• DB2_CONNECTION_STRING：作为连接串的环境变量备选。实际上连接串更推荐的传参方式是构造时的connection_string字段，它被声明为必填（required）。

connection_string字段的取值格式（见 db2_search_tool.py#L108-L114）支持两种写法：

1. 标准键值对串：DATABASE=mydb;HOSTNAME=localhost;PORT=50000;PROTOCOL=TCPIP;UID=user;PWD=pass;，其中PROTOCOL=TCPIP是 TCP 连接协议，PORT默认常为50000；
2. 本地数据库名简写：仅传数据库名（如"TESTDB"），适用于本机已配置的本地连接。

四、快速上手：基础检索

官方 README 提供了最小可运行示例。假设 DB2 中已有一张名为documents的表，其默认约定为：文本列名为content，向量列名为embedding（类型为 DB2VECTOR）：

from crewai_tools import DB2VectorSearchTool tool = DB2VectorSearchTool( connection_string="DATABASE=TESTDB;HOSTNAME=localhost;PORT=50000;PROTOCOL=TCPIP;UID=db2user;PWD=password;", table_name="documents", ) result = tool.run( query="What is machine learning?", ) print(result)

DB2VectorSearchTool继承自crewai.tools.BaseTool（见 db2_search_tool.py#L12、db2_search_tool.py#L62-L72），因此它拥有标准的name、description、args_schema，可以被 Agent 当作普通工具调度。一次run的内部执行链路为：

1. 校验查询文本非空（空串、纯空白或None都会返回错误 JSON，见 db2_search_tool.py#L269-L276）；
2. 为查询文本生成嵌入向量；
3. 建立 DB2 连接（失败时清理并返回"Failed to connect to DB2: ..."）；
4. 校验距离度量与所有表/列标识符；
5. 构造VECTOR_DISTANCESQL 并参数化执行；
6. 按max_distance后置过滤、组装结果、断开连接；
7. 返回规范化 JSON 字符串。

成功响应示例

run返回的是 JSON 字符串，整体结构如下：

{ "success": true, "results": [ { "distance": 0.12, "data": { "content": "machine learning is ..." } } ] }

每个结果条目中，distance恒为数值型的距离分数，data则由“返回列名 → 行值”的映射构成。

五、元数据过滤

当需要按业务元数据缩小检索范围时，同时传入filter_by（列名）与filter_value（过滤值）：

result = tool.run( query="AI papers", filter_by="category", filter_value="AI", )

两点使用约束来自入参 schemaDB2ToolSchema的校验器（见 db2_search_tool.py#L30-L59）：

• filter_by与filter_value必须成对出现，只传其一会抛出"filter_by and filter_value must be provided together."；
• filter_by不允许为空白字符串，否则抛出"filter_by must be a non-empty column name."。

这些行为均被单元测试覆盖，例如 test_db2_search_tool.py#L111-L131。

底层实现上，过滤会拼接为参数化WHERE子句：WHERE {filter_by} = ?，过滤值通过绑定参数传入（而非字符串拼接），从源头上杜绝值注入（见 db2_search_tool.py#L306-L309）。测试 test_db2_search_tool.py#L447-L461 验证了过滤值确实进入了execute的参数元组，test_db2_search_tool.py#L485-L494 则验证了带过滤时 SQL 中确实包含WHERE dept = ?。

六、全部配置字段与默认值

工具暴露了比 README 更丰富的可调参数。下表汇总了这些字段、默认值与约束（均可从 db2_search_tool.py#L108-L137 以及 tool.specs.json L5852-L6076 的init_params_schema交叉印证）：

组合示例：多返回列 + 距离阈值

下面示例演示检索时同时返回多列、收紧返回条数并过滤掉低相关度结果：

tool = DB2VectorSearchTool( connection_string="DATABASE=TESTDB;HOSTNAME=localhost;PORT=50000;PROTOCOL=TCPIP;UID=db2user;PWD=password;", table_name="papers", vector_column="embedding", return_columns=["title", "abstract", "year"], limit=5, distance_metric="COSINE", max_distance=0.35, ) result = tool.run( query="multi-agent reinforcement learning", filter_by="year", filter_value=2024, )

支持的相似度度量

源码通过类变量维护了一份距离度量白名单（见 db2_search_tool.py#L74-L84），与 DB2VECTOR_DISTANCE内置函数能力对齐：

• COSINE（默认，余弦相似度语义）
• EUCLIDEAN（欧氏距离）
• EUCLIDEAN_SQUARED（平方欧氏距离）
• DOT（点积）
• HAMMING（汉明距离）
• MANHATTAN（曼哈顿距离）

传入白名单之外的度量（例如大写后仍不匹配的字符串）会抛出"Invalid distance metric: ..."，测试见 test_db2_search_tool.py#L371-L383。这层白名单并非仅为了提示，更是一道安全闸门：度量值会直接拼进 SQL 的函数名位置，白名单机制杜绝了在该位置注入任意 SQL 的可能。

七、嵌入策略：自定义函数优先，否则 OpenAI

查询向量生成遵循“自定义函数优先、OpenAI 兜底”的策略，实现在 db2_search_tool.py#L222-L235：

def _generate_embedding(self, text: str) -> list[float]: if self.custom_embedding_fn: return self.custom_embedding_fn(text) result = ( self._get_openai_client() .embeddings.create( input=[text], model=self.embedding_model, ) .data[0] .embedding ) return list(result)

自定义嵌入函数

如果希望接入自有嵌入服务、本地模型或企业级向量底座，可直接传入可调用对象：

def my_embedder(text: str) -> list[float]: # 调用任意模型服务，返回一维 float 列表 return [0.1, 0.2, ...] tool = DB2VectorSearchTool( connection_string="DATABASE=TESTDB;HOSTNAME=localhost;PORT=50000;PROTOCOL=TCPIP;UID=db2user;PWD=password;", table_name="documents", custom_embedding_fn=my_embedder, )

其字段类型为ImportString[Callable[[str], list[float]]]（见 db2_search_tool.py#L150-L153），即支持传函数对象，也可传一个可导入路径字符串。相关行为测试见 test_db2_search_tool.py#L279-L290。

OpenAI 兜底路径

未提供自定义函数时，工具从环境变量读取OPENAI_API_KEY，用embedding_model指定的模型（默认text-embedding-3-large）生成查询向量。OpenAI 客户端对象会缓存复用，避免每次查询都重复初始化（测试 test_db2_search_tool.py#L306-L319 断言构造函数只被调用一次）。

一个关键的工程提示

从源码可以推断出一个与入库强相关的约束：SQL 中向量维度取自查询向量的实际长度（vector_dimension = len(query_vector)），并以此维度对存储向量做VECTOR(...)转换（见 db2_search_tool.py#L237-L260、db2_search_tool.py#L300-L301）。因此入库阶段使用的嵌入函数/模型必须与检索阶段保持一致，否则存储向量与查询向量的维度或语义空间不匹配，会导致转换错误或相似度失去意义。这也是“入库与检索分离”架构下最容易踩的坑。

八、底层 SQL 与执行机制

_build_sql（db2_search_tool.py#L237-L260）负责把各参数拼装成一条完整的向量检索语句，典型形态为：

SELECT {return_columns...}, VECTOR_DISTANCE({vector_column}, VECTOR(CAST(? AS CLOB), {vector_dimension}, FLOAT32), {distance_metric}) AS distance FROM {table_name} [WHERE {filter_by} = ?] ORDER BY distance ASC FETCH FIRST {limit} ROWS ONLY;

理解这条语句的几个关键点：

1. 查询向量通过占位符?绑定，先转CLOB再以FLOAT32与指定维度转成VECTOR，作为VECTOR_DISTANCE的比对对象；
2. distance列始终被追加在 SELECT 的最后一列，这正是结果解析时row[-1]取距离的约定来源（见 db2_search_tool.py#L326）；
3. 结果按距离升序排列，即最相似者排在最前；
4. FETCH FIRST {limit} ROWS ONLY在数据库侧就限制返回条数；
5. 向量字符串与过滤值都走参数绑定，返回列清单来自已通过校验的return_columns。

filter_by/filter_value会额外追加WHERE子句（无过滤时整段省略，测试见 test_db2_search_tool.py#L496-L505）。max_distance阈值则在 SQL 返回后于 Python 侧做二次过滤（见 db2_search_tool.py#L328-L329），测试用例 test_db2_search_tool.py#L434-L445 演示了“过远文档被剔除、邻近文档保留”。

九、安全设计：标识符校验与注入防护

工具在源码 docstring 中自称 “fortified”（加固型），其安全设计可以从三层机制得到印证：

1. 标识符白名单正则校验

任何进入 SQL 的表名、向量列名、返回列名、过滤列名，都会先经过_validate_identifier（见 db2_search_tool.py#L193-L209）的严格校验：

• 简单标识符必须以字母开头，仅允许字母、数字、下划线（^[A-Za-z][A-Za-z0-9_]*$）；
• table_name额外开启allow_period=True，允许myschema.mytable形式的恰好一段点号分隔；
• 任何不匹配的名字都会抛出"Security Alert: Invalid database identifier detected: ..."。

测试 test_db2_search_tool.py#L230-L273 用一批恶意样本（'; DROP TABLE documents; --、table--、col OR 1=1、schema..table、纯点号串等）验证了该校验的有效性。

2. 度量白名单

如第六节所述，distance_metric仅允许六个预置值（见 db2_search_tool.py#L74-L84），防止度量被注入为任意 SQL。

3. 参数化绑定 + schema 级双保险

过滤值一律通过?占位符绑定，杜绝值注入；filter_by/filter_value的成对性与非空约束又在DB2ToolSchema层提前拦截（见 db2_search_tool.py#L53-L59）。两条防线共同作用的结果是：即使用户在filter_by中传入col; DROP TABLE ...这类载荷，也会在校验阶段被拒绝并返回错误 JSON（测试见 test_db2_search_tool.py#L552-L591）。

十、结果序列化与错误语义

类型安全的 JSON 编码

DB2 返回的行可能包含Decimal、datetime、bytes等无法直接被标准json序列化的类型。工具为此内置了DB2JSONEncoder（见 db2_search_tool.py#L17-L27）：

对应测试见 test_db2_search_tool.py#L200-L223，含 Decimal 金额列返回场景的端到端验证（test_db2_search_tool.py#L507-L516）。

统一的错误返回格式

工具的异常处理策略是不抛裸异常、以 JSON 形式返回错误（见 db2_search_tool.py#L278-L362）：

{ "success": false, "error": "具体的错误信息", "error_type": "ValueError" }

连接失败时错误信息带Failed to connect to DB2前缀；空白查询则返回专门的"Query cannot be empty or contain only whitespace."提示。应用层在调用后应首先检查success字段，再读取results。error_type字段使用异常类型名（如ValueError、RuntimeError），便于上层进行精确的分类处理。

十一、连接生命周期：一次查询一个连接

工具采用了“每次检索独立建连、显式释放”的短连接生命周期策略（见 db2_search_tool.py#L174-L191）：

1. _connect()惰性解析db2_package/db2_dbi_package（支持默认None或显式字符串如"ibm_db"），随后建立底层连接、包装 DBI 连接并取得 cursor；
2. _run()无论成功或异常，都会在退出前调用_disconnect()，关闭 cursor、DBI 连接与底层连接，并把三个句柄全部复位为None（幂等，重复调用安全）；
3. 对象析构函数__del__同样调用_disconnect()，作为最后的兜底清理（见 db2_search_tool.py#L364-L365）。

测试 test_db2_search_tool.py#L609-L620 证明连续两次_connect()会建立两个新连接，test_db2_search_tool.py#L518-L545 则验证了成功路径与异常路径都会触发_disconnect。这种策略对 Agent 多轮调用场景是友好的：不会因为 Agent 空闲而长期占用 DB2 连接数。

十二、在 Crew 中接入 Agent

作为标准化的 CrewAI 工具，最自然的用法是把它挂到 Agent 的tools列表中，让 Agent 根据任务自行决定何时检索：

from crewai import Agent, Crew, Process, Task from crewai_tools import DB2VectorSearchTool search_tool = DB2VectorSearchTool( connection_string="DATABASE=TESTDB;HOSTNAME=localhost;PORT=50000;PROTOCOL=TCPIP;UID=db2user;PWD=password;", table_name="documents", return_columns=["title", "content"], limit=5, ) researcher = Agent( role="知识库研究员", goal="基于 DB2 向量知识库检索信息并回答问题", backstory="你擅长把用户的提问转化为精准的语义检索。", tools=[search_tool], ) task = Task( description="检索并总结：什么是机器学习？", expected_output="一段基于检索结果的准确总结", agent=researcher, ) crew = Crew(agents=[researcher], tasks=[task], process=Process.sequential) crew.kickoff()

DB2VectorSearchTool已通过 crewai_tools 顶层__init__.py（约 L67、L255）对外导出，因此from crewai_tools import DB2VectorSearchTool, DB2ToolSchema是官方支持的标准导入方式（验证见 test_db2_search_tool.py#L701-L707）。

十三、无真实数据库的测试验证

该工具的质量保障依赖一套纯单元测试，其设计思路（见 test_db2_search_tool.py#L1-L56）是：在 import 工具模块之前，先向sys.modules注入ibm_db/ibm_db_dbi的 MagicMock 桩模块，从而在没有真实 IBM DB2 实例、甚至没安装驱动的情况下覆盖全部逻辑分支。

测试矩阵覆盖了：schema 校验、默认值与字段约束（limit越界、max_distance为负、空return_columns）、标识符注入防护、嵌入函数优先级与 OpenAI 客户端缓存、空白查询防护、连接失败、非法度量、成功路径的 JSON 结构、多返回列映射、max_distance过滤、SQL 中的度量与WHERE/FETCH FIRST断言、断开连接的生命周期等等。若你希望本地复跑验证，在 lib/crewai-tools 目录下执行 pytest 针对该文件即可（crewai_tools.tools.db2_search_tool为正确导入名）。

结语

DB2VectorSearchTool将 IBM DB2 从传统的关系型数据库延伸为 CrewAI Agent 可感知的语义检索底座：它封装了查询嵌入、VECTOR_DISTANCE相似度计算、元数据过滤与规范化 JSON 输出，同时在标识符校验、度量白名单、参数化绑定与连接生命周期上做了工程化加固。投入使用前请再次确认两点：表结构符合约定（VECTOR列 + 返回列），且入库时的嵌入模型与检索时保持一致。关于完整的构造参数与运行时入参 schema，可查阅工具生成的 tool.specs.json 中DB2VectorSearchTool条目。

【免费下载链接】crewAIFramework for orchestrating role-playing, autonomous AI agents. By fostering collaborative intelligence, CrewAI empowers agents to work together seamlessly, tackling complex tasks.项目地址: https://gitcode.com/GitHub_Trending/cr/crewAI