用 Amazon Bedrock 构建多智能体向量提示文档分类系统

S3 文档
  │
  ├── OCR / 文本抽取 ──> 文本分析智能体
  │
  ├── 页面渲染 ────────> Titan Multimodal Embeddings ──> 向量检索
  │
  └────────────────────> 决策智能体 ──> 类别、理由、置信度

{
  "document_id": "sample-policy-001",
  "label": "insurance_policy",
  "page": 1,
  "embedding_model": "amazon.titan-embed-image-v1",
  "metadata": {
    "language": "en",
    "source": "human_reviewed"
  }
}

python -m venv .venv
source .venv/bin/activate
pip install strands-agents boto3
export AWS_REGION=us-east-1
export BEDROCK_TEXT_MODEL_ID=your-enabled-haiku-model-id
export BEDROCK_VISION_MODEL_ID=amazon.titan-embed-image-v1

import json
import os
from pathlib import Path

import boto3
from strands import Agent
from strands.models import BedrockModel

REGION = os.getenv("AWS_REGION", "us-east-1")
TEXT_MODEL_ID = os.environ["BEDROCK_TEXT_MODEL_ID"]
VISION_MODEL_ID = os.getenv("BEDROCK_VISION_MODEL_ID", "amazon.titan-embed-image-v1")

runtime = boto3.client("bedrock-runtime", region_name=REGION)

text_model = BedrockModel(
    model_id=TEXT_MODEL_ID,
    region_name=REGION,
)

text_agent = Agent(
    model=text_model,
    system_prompt=(
        "You classify insurance documents from OCR text. "
        "Return JSON with label, evidence, and confidence. "
        "Allowed labels: insurance_policy, affidavit, endorsement, unknown."
    ),
)

decision_agent = Agent(
    model=text_model,
    system_prompt=(
        "You are the final insurance-document classifier. "
        "Combine text evidence and visual-search evidence. "
        "Return strict JSON: label, confidence, reasons, needs_human_review. "
        "If evidence conflicts or confidence is below 0.80, request human review."
    ),
)


def embed_image(image_path: str) -> list[float]:
    """Create a Titan multimodal embedding for one page image.

    Confirm the request schema and model ID in the target AWS Region before use.
    """
    image_bytes = Path(image_path).read_bytes()
    body = json.dumps({
        "inputImage": image_bytes.hex(),
        "embeddingConfig": {"outputEmbeddingLength": 1024},
    })

    response = runtime.invoke_model(
        modelId=VISION_MODEL_ID,
        body=body,
        contentType="application/json",
        accept="application/json",
    )
    payload = json.loads(response["body"].read())
    return payload["embedding"]


def search_visual_examples(embedding: list[float]) -> list[dict]:
    """Replace this adapter with OpenSearch Serverless, Aurora, or another vector store."""
    # 生产实现中应使用向量数据库的 k-NN 查询，并返回 metadata 与 score。
    return [
        {
            "sample_id": "sample-affidavit-004",
            "label": "affidavit",
            "score": 0.91,
            "evidence": "signature and notary block are visually similar",
        }
    ]


def classify_document(ocr_text: str, first_page_image: str) -> dict:
    text_result = text_agent(
        f"OCR text:\n{ocr_text}\n\nReturn JSON only."
    )

    embedding = embed_image(first_page_image)
    visual_result = search_visual_examples(embedding)

    decision_input = {
        "text_analysis": str(text_result),
        "visual_search": visual_result,
        "policy": {
            "human_review_threshold": 0.80,
            "conflict_rule": "text and visual evidence conflict => review",
        },
    }
    final_result = decision_agent(json.dumps(decision_input, ensure_ascii=False))
    return {"result": str(final_result), "visual_matches": visual_result}


if __name__ == "__main__":
    result = classify_document(
        ocr_text="AFFIDAVIT OF LOSS ... subscribed and sworn before me ...",
        first_page_image="page-001.jpg",
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))

You classify one insurance document.

Text evidence:
- OCR keywords: "subscribed", "sworn", "notary"
- Text candidate: affidavit
- Text confidence: 0.86

Visual evidence:
- Top match: sample-affidavit-004
- Similarity: 0.91
- Visual cues: signature area and notary block

Rules:
1. Choose only from the approved label set.
2. Do not treat similarity alone as proof.
3. If evidence conflicts or confidence < 0.80, set needs_human_review=true.
4. Return valid JSON only.