CrewAI BedrockKBRetrieverTool 实战详解：为 Agent 接入 Amazon Bedrock 知识库检索

本文基于 CrewAI 仓库中 crewai-tools 包的 Bedrock 知识库工具文档（lib/crewai-tools/src/crewai_tools/aws/bedrock/knowledge_base/README.md）及其对应源码实现撰写。读完后，你将掌握 BedrockKBRetrieverTool 的安装与配置、全部构造参数的取值规则与校验逻辑、在 CrewAI Agent 中的接入方式，以及该工具底层对 bedrock-agent-runtime.retrieve() 的调用链与响应解析机制。

1. 工具定位与核心能力

BedrockKBRetrieverTool 让 CrewAI 智能体能够用自然语言查询（natural language query）从 Amazon Bedrock 知识库中检索信息，是典型的 RAG（检索增强生成）接入点：Agent 负责推理与任务编排，知识库检索负责提供来自企业私有数据的事实依据。

工具实现在 retriever_tool.py，类定义继承自 crewai.tools.BaseTool，并暴露给 LLM 的输入 Schema 只有一个字段：

class BedrockKBRetrieverToolInput(BaseModel):
    """Input schema for BedrockKBRetrieverTool."""
    query: str = Field(
        ..., description="The query to retrieve information from the knowledge base"
    )

也就是说，Agent 侧只需要生成一句查询语句，工具内部完成其余所有工作。工具在包内的导出路径为 knowledge_base/init.py，并向上聚合到 bedrock/init.py 与 aws/init.py，因此既可以从 crewai_tools.aws.bedrock.knowledge_base 导入，也可以直接从 crewai_tools.aws 导入。

2. 安装与环境要求

2.1 安装

pip install 'crewai[tools]'

2.2 前置条件

• 已配置 AWS 凭据（环境变量或 AWS CLI 均可）；
• 依赖 boto3 与 python-dotenv 包。源码层面，工具声明了 package_dependencies: list[str] = Field(default_factory=lambda: ["boto3"])，且模块导入时执行 load_dotenv()，因此 .env 文件中的配置会自动被加载；
• 具备目标 Amazon Bedrock 知识库的访问权限。

2.3 环境变量

BEDROCK_KB_ID=your-knowledge-base-id  # 可作为 knowledge_base_id 参数的替代
AWS_REGION=your-aws-region            # 默认 us-east-1
AWS_ACCESS_KEY_ID=your-access-key     # AWS 鉴权所需
AWS_SECRET_ACCESS_KEY=your-secret-key # AWS 鉴权所需

从源码看，__init__ 中的参数回退逻辑为：

self.knowledge_base_id = knowledge_base_id or os.getenv("BEDROCK_KB_ID")

即构造参数优先，未显式传入时才回退到 BEDROCK_KB_ID 环境变量（见 retriever_tool.py#L59-L60）。Region 的解析链为 AWS_REGION → AWS_DEFAULT_REGION → 兜底 us-east-1（见 retriever_tool.py#L194-L200）。

3. 在 CrewAI Agent 中的完整用法

以下示例继承自工具 README，展示"初始化工具 → 挂载到 Agent → 定义 Task → 运行 Crew"的完整链路：

from crewai import Agent, Task, Crew
from crewai_tools.aws.bedrock.knowledge_base.retriever_tool import BedrockKBRetrieverTool

# 初始化工具
kb_tool = BedrockKBRetrieverTool(
    knowledge_base_id="your-kb-id",
    number_of_results=5
)

# 创建使用该工具的 CrewAI Agent
researcher = Agent(
    role='Knowledge Base Researcher',
    goal='Find information about company policies',
    backstory='I am a researcher specialized in retrieving and analyzing company documentation.',
    tools=[kb_tool],
    verbose=True
)

# 为 Agent 创建任务
research_task = Task(
    description="Find our company's remote work policy and summarize the key points.",
    agent=researcher
)

# 创建包含该 Agent 的 Crew
crew = Crew(
    agents=[researcher],
    tasks=[research_task],
    verbose=2
)

# 运行
result = crew.kickoff()
print(result)

工具初始化后，其 description 会被动态改写为 Retrieves information from Amazon Bedrock Knowledge Base '{knowledge_base_id}' given a query，这样 Agent 在决策是否调用工具时能明确感知具体查询的是哪个知识库。

4. 参数详解与校验规则

4.1 构造参数

参数	类型	必填	默认值	说明
knowledge_base_id	str	是*	None	知识库唯一标识（0–10 位字母数字字符）；可用环境变量 BEDROCK_KB_ID 替代
number_of_results	int	否	5	返回的最大结果数
retrieval_configuration	dict	否	None	自定义知识库查询配置
guardrail_configuration	dict	否	None	内容过滤（Guardrail）设置
next_token	str	否	None	分页游标，用于获取下一批结果

* 若通过 BEDROCK_KB_ID 环境变量提供，则构造参数可省略。

4.2 参数校验逻辑（源码级）

_validate_parameters() 在 __init__ 末尾被调用，校验失败会抛出 BedrockValidationError（定义于 exceptions.py，继承自 BedrockError 基类）。具体约束（见 retriever_tool.py#L89-L124）：

• knowledge_base_id：非空、必须是字符串、长度 ≤ 10、仅允许字母数字字符；
• next_token：若非空，必须是字符串、长度在 1–2048 之间、不能包含空格；
• number_of_results：必须是整数且 > 0。

4.3 retrieval_configuration 的自动生成

如果没有显式传入 retrieval_configuration，工具会根据 number_of_results 自动构造：

def _build_retrieval_configuration(self) -> dict[str, Any]:
    vector_search_config = {}
    if self.number_of_results is not None:
        vector_search_config["numberOfResults"] = self.number_of_results
    return {"vectorSearchConfiguration": vector_search_config}

即默认等价于 {"vectorSearchConfiguration": {"numberOfResults": 5}}。

5. 高级用法：自定义检索配置与 Guardrail

5.1 混合检索（HYBRID）

显式传入 retrieval_configuration 时，工具会原样使用该配置（不再自动构造）：

kb_tool = BedrockKBRetrieverTool(
    knowledge_base_id="your-kb-id",
    retrieval_configuration={
        "vectorSearchConfiguration": {
            "numberOfResults": 10,
            "overrideSearchType": "HYBRID"
        }
    }
)

policy_expert = Agent(
    role='Policy Expert',
    goal='Analyze company policies in detail',
    backstory='I am an expert in corporate policy analysis with deep knowledge of regulatory requirements.',
    tools=[kb_tool]
)

5.2 内容过滤（Guardrail）与分页

guardrail_configuration 与 next_token 分别映射到 Bedrock API 的 guardrailConfiguration 与 nextToken 字段，用于对检索内容做合规过滤和分批拉取：

kb_tool = BedrockKBRetrieverTool(
    knowledge_base_id="kb123",
    guardrail_configuration={
        "guardrailIdentifier": "your-guardrail-id",
        "guardrailVersion": "DRAFT",
        "trace": "ENABLED",
    }
)

响应中的 guardrailAction 字段会回传 Guardrail 的处理动作，便于在 Agent 侧感知内容是否被干预。

6. 底层调用链：_run() 的完整执行流程

_run(query) 是 Agent 调用工具时的实际入口，执行流程可拆解为六步（见 retriever_tool.py#L183-L250）：

1. 懒加载 boto3：import boto3 失败时抛出 ImportError，提示执行 uv add boto3；

2. 创建客户端：boto3.client("bedrock-agent-runtime", region_name=...)，AWS 鉴权由 SDK 自动从环境读取；

3. 组装请求参数：

   retrieve_params = {
       "knowledgeBaseId": self.knowledge_base_id,
       "retrievalQuery": {"text": query},
   }
   if self.retrieval_configuration:
       retrieve_params["retrievalConfiguration"] = self.retrieval_configuration
   if self.guardrail_configuration:
       retrieve_params["guardrailConfiguration"] = self.guardrail_configuration
   if self.next_token:
       retrieve_params["nextToken"] = self.next_token
   

4. 发起检索：bedrock_agent_runtime.retrieve(**retrieve_params)；

5. 逐条后处理：对 response["retrievalResults"] 中每条记录调用 _process_retrieval_result() 标准化；

6. 序列化返回：结果为空时返回 {"message": "No results found for the given query."}，否则输出 results 列表；如响应中存在 nextToken、guardrailAction 则一并透传，最终以 json.dumps(..., indent=2) 返回给 Agent。

异常处理分两层：botocore.exceptions.ClientError 会被解包出 Code 与 Message 后抛出 BedrockKnowledgeBaseError("Error ({code}): {message}")；其他异常统一包装为 BedrockKnowledgeBaseError("Unexpected error: ...")。这两个异常类型均定义在 exceptions.py 中，可在业务代码中精确捕获知识库错误而不影响其他工具调用。

7. 响应格式与来源映射机制

7.1 标准化响应

工具对外返回 JSON 字符串：

{
  "results": [
    {
      "content": "Retrieved text content",
      "content_type": "text",
      "source_type": "S3",
      "source_uri": "s3://bucket/document.pdf",
      "score": 0.95,
      "metadata": {
        "additional": "metadata"
      }
    }
  ],
  "nextToken": "pagination-token",
  "guardrailAction": "NONE"
}

字段说明：

• content / content_type：检索片段文本及其类型（默认 text）；
• source_type / source_uri：数据源类型与定位地址（见 7.2 的映射表）；
• score：相关性分数，仅当 API 返回时存在；
• metadata：知识库附加元数据，仅当返回时存在；
• 此外，_process_retrieval_result() 还会把二进制片段映射为 byte_content（来自 byteContent）、结构化行映射为 row_content（来自 row），用于 SQL 等非文本数据源。

7.2 八种数据源的 URI 映射

_process_retrieval_result() 内置了一张 Bedrock 位置类型 → 工具输出类型 → URI 取字段的映射表（见 retriever_tool.py#L144-L153）：

Bedrock 位置字段	输出 source_type	URI 取值字段
s3Location	S3	uri
confluenceLocation	Confluence	url
salesforceLocation	Salesforce	url
sharePointLocation	SharePoint	url
webLocation	Web	url
customDocumentLocation	CustomDocument	id
kendraDocumentLocation	KendraDocument	uri
sqlLocation	SQL	query

这意味着该工具天然覆盖 Bedrock 知识库支持的多种数据源：Amazon S3、Confluence、Salesforce、SharePoint、网页、自定义文档位置、Amazon Kendra 与 SQL 数据库，Agent 拿到的每条结果都带有可追溯的来源信息。

8. 典型应用场景

工具 README 归纳了五类落地场景，均围绕"让 Agent 的推理扎根于企业真实数据"展开：

• 企业知识集成：Agent 直接访问组织私有知识而不暴露敏感数据，基于内部政策、流程与文档做决策；
• 领域专家知识：无需微调模型，即可让 Agent 接入法律、医疗、技术等垂直领域知识库，复用 AWS 环境中已有的知识资产；
• 数据驱动决策：以实际业务数据为回答依据，减少幻觉；
• 可扩展的信息访问：无需将数 TB 级知识嵌入模型，按任务动态检索相关片段；
• 合规与治理：Agent 的回答对齐已批准的内部文档，且 source_uri / metadata 可形成可审计的信息来源记录，配合 Guardrail 进一步控制访问边界。

9. 小结

BedrockKBRetrieverTool 的实现非常聚焦：一个 query 输入 Schema、五条构造参数、一次 retrieve() 调用、一套八源映射的结果标准化逻辑。它的工程价值在于把 Bedrock 知识库检索封装成 CrewAI 工具协议（BaseTool）下的普通成员——Agent 通过 tools=[kb_tool] 即可挂载，无需感知 boto3 细节；而参数校验、错误包装、来源映射等健壮性逻辑保证了它在生产链路中可预测、可调试。

相关源码索引：

• 工具实现：retriever_tool.py
• 异常定义：exceptions.py
• 模块文档：README.md