清理

如果我们计划使用上面创建的知识库来构建我们的 RAG 应用程序，请确保注释以下部分。 如果我们只是想尝试使用 SDK 创建 KB，请确保删除所有创建的资源，因为我们将被收取存储文档在 OSS 索引中的费用。

删除知识库并在完成所有笔记本后删除资源。

%store -r

import boto3
import time

boto3_session = boto3.Session()
bedrock_agent_client = boto3_session.client('bedrock-agent', region_name=boto3_session.region_name)
aoss_client = boto3.client('opensearchserverless')
s3_client = boto3_session.client('s3', region_name=boto3_session.region_name)
iam_client = boto3.client("iam")

删除 Bedrock 知识库数据源

# 获取与 kb 关联的数据源
response = bedrock_agent_client.list_data_sources(knowledgeBaseId=kb_id)
data_sources_list = [ds_summary for ds_summary in response['dataSourceSummaries']]

for idx, ds in enumerate(data_sources_list):
    bedrock_agent_client.delete_data_source(dataSourceId = ds["dataSourceId"], knowledgeBaseId=ds["knowledgeBaseId"])
    time.sleep(10)

删除知识库和 OpenSearch 集合

# 获取 kb 执行角色
response = bedrock_agent_client.get_knowledge_base(knowledgeBaseId=kb_id)
kb_role_name = response['knowledgeBase']['roleArn'].split("/")[-1]

# 获取与 kb 执行角色关联的所有策略
kb_attached_role_policies_response = iam_client.list_attached_role_policies(
    RoleName=kb_role_name)
kb_attached_role_policies = kb_attached_role_policies_response['AttachedPolicies']

bedrock_agent_client.delete_knowledge_base(knowledgeBaseId=kb_id)
aoss_client.delete_collection(id=collection['createCollectionDetail']['id'])
aoss_client.delete_access_policy(type="data", name=access_policy['accessPolicyDetail']['name'])
aoss_client.delete_security_policy(type="network", name=network_policy['securityPolicyDetail']['name'])
aoss_client.delete_security_policy(type="encryption", name=encryption_policy['securityPolicyDetail']['name'])

删除角色和策略

for policy in kb_attached_role_policies:
    iam_client.detach_role_policy(
            RoleName=kb_role_name,
            PolicyArn=policy['PolicyArn']
    )

iam_client.delete_role(RoleName=kb_role_name)

for policy in kb_attached_role_policies:
    iam_client.delete_policy(PolicyArn=policy['PolicyArn'])

删除 S3 对象

objects = s3_client.list_objects(Bucket=bucket_name)
if 'Contents' in objects:
    for obj in objects['Contents']:
        s3_client.delete_object(Bucket=bucket_name, Key=obj['Key'])
s3_client.delete_bucket(Bucket=bucket_name)