用 Bedrock 知识库把监管问询从数天压缩到数小时——Amazon FinTech 的做法

# 创建团队专属 S3 bucket
aws s3 mb s3://regulatory-kb-team-alpha --region us-east-1

# 上传合规文档
aws s3 cp ./compliance-docs/ s3://regulatory-kb-team-alpha/docs/ \
  --recursive --include "*.pdf" --include "*.docx"

# 确认文件已上传
aws s3 ls s3://regulatory-kb-team-alpha/docs/ --recursive

import boto3
import time
import uuid

bedrock_agent = boto3.client("bedrock-agent", region_name="us-east-1")

KB_NAME = "regulatory-kb-team-alpha"
S3_URI = "s3://regulatory-kb-team-alpha/docs/"
# 使用 Bedrock 自带的 Titan Embedding 模型
EMBED_MODEL_ID = "amazon.titan-embed-text-v1"

# 1. 创建 Knowledge Base
create_resp = bedrock_agent.create_knowledge_base(
    name=KB_NAME,
    description="Team Alpha 专属监管合规文档知识库",
    roleArn="arn:aws:iam::<YOUR_ACCOUNT_ID>:role/BedrockKBRole",  # 需预创建，赋予 S3 读取权限
    knowledgeBaseConfiguration={
        "type": "VECTOR",
        "vectorKnowledgeBaseConfiguration": {
            "embeddingModelArn": f"arn:aws:bedrock:us-east-1::foundation-model/{EMBED_MODEL_ID}"
        }
    },
    storageConfiguration={
        "type": "OPENSEARCH_SERVERLESS",  # Bedrock 自动管理向量索引
        "opensearchServerlessConfiguration": {
            "collectionArn": "<COLLECTION_ARN>",  # 可通过 aoss client 创建，或让 Bedrock 自动创建
            "fieldMapping": {
                "vectorField": "vector",
                "textField": "text",
                "metadataField": "metadata"
            }
        }
    }
)

kb_id = create_resp["knowledgeBase"]["knowledgeBaseId"]
print(f"Knowledge Base ID: {kb_id}")

# 2. 添加 S3 数据源
ds_resp = bedrock_agent.create_data_source(
    knowledgeBaseId=kb_id,
    name="team-alpha-s3-docs",
    description="合规文档 S3 数据源",
    dataSourceConfiguration={
        "type": "S3",
        "s3Configuration": {
            "bucketArn": "arn:aws:s3:::regulatory-kb-team-alpha",
            "prefix": "docs/"
        }
    },
    vectorIngestionConfiguration={
        "chunkingConfiguration": {
            "chunkingStrategy": "FIXED_SIZE",
            "fixedSizeChunkingConfiguration": {
                "maxTokens": 500,       # 每块最大 token 数
                "overlapPercentage": 20  # 块间重叠，避免关键信息被截断
            }
        }
    }
)

ds_id = ds_resp["dataSource"]["dataSourceId"]
print(f"Data Source ID: {ds_id}")

# 3. 触发同步（将 S3 文档向量化并写入索引）
sync_resp = bedrock_agent.start_ingestion_job(
    knowledgeBaseId=kb_id,
    dataSourceId=ds_id
)
job_id = sync_resp["ingestionJob"]["ingestionJobId"]

# 等待同步完成
while True:
    status = bedrock_agent.get_ingestion_job(
        knowledgeBaseId=kb_id,
        dataSourceId=ds_id,
        ingestionJobId=job_id
    )["ingestionJob"]["status"]
    print(f"同步状态: {status}")
    if status in ("COMPLETE", "FAILED"):
        break
    time.sleep(10)

print(f"知识库同步完成，状态: {status}")

bedrock_runtime = boto3.client("bedrock-runtime", region_name="us-east-1")
bedrock_agent_runtime = boto3.client("bedrock-agent-runtime", region_name="us-east-1")

MODEL_ID = "anthropic.claude-3-sonnet-20240229"

def answer_regulatory_query(question: str, kb_id: str) -> dict:
    """检索知识库并生成监管问询回复"""

    # 1. 从 Knowledge Base 检索相关文档片段
    retrieve_resp = bedrock_agent_runtime.retrieve(
        knowledgeBaseId=kb_id,
        retrievalQuery={"text": question},
        maxResults=5  # 返回最相关的 5 个片段
    )

    # 2. 组装检索结果为上下文
    context_parts = []
    citations = []
    for result in retrieve_resp["retrievalResults"]:
        text = result["content"]["text"]
        source = result["location"]["s3Location"]["uri"]
        context_parts.append(text)
        citations.append({"text": text, "source": source})

    context = "\n\n---\n\n".join(context_parts)

    # 3. 构造 prompt，要求模型引用来源
    prompt = f"""你是一名合规分析师，正在回复监管机构的问询。
请基于以下内部文档内容回答问题。如果文档中没有相关信息，明确说明"现有文档未覆盖此问题"。
每条结论必须标注来源文档路径。

内部文档内容：
{context}

监管问询问题：
{question}

请给出结构化回复，包含：
1. 直接回答
2. 依据的文档段落（引用原文）
3. 来源文档路径
"""

    # 4. 调用 Bedrock 模型生成回复
    body = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 2048,
        "messages": [{"role": "user", "content": prompt}]
    }

    model_resp = bedrock_runtime.invoke_model(
        modelId=MODEL_ID,
        body=json.dumps(body),
        contentType="application/json",
        accept="application/json"
    )

    response_body = json.loads(model_resp["body"].read())
    answer = response_body["content"][0]["text"]

    return {"answer": answer, "citations": citations}

# 使用示例
result = answer_regulatory_query(
    question="我们在欧洲市场的反洗钱（AML）审查流程是什么？",
    kb_id=kb_id
)

print("=== 回复 ===")
print(result["answer"])
print("\n=== 检索来源 ===")
for c in result["citations"]:
    print(f"  文档: {c['source']}")
    print(f"  片段: {c['text'][:100]}...")