用生成式 AI 在 AWS 上自动化合同智能：Doczy.ai 的实践与可复用架构

aws dynamodb create-table \
  --table-name ContractIntelligence \
  --attribute-definitions \
    AttributeName=contractId,AttributeType=S \
    AttributeName=uploadDate,AttributeType=S \
  --key-schema \
    AttributeName=contractId,KeyType=HASH \
    AttributeName=uploadDate,KeyType=RANGE \
  --billing-mode PAY_PER_REQUEST \
  --region us-east-1

# 创建 bucket
aws s3api create-bucket \
  --bucket my-contract-intelligence-bucket \
  --region us-east-1

# 添加 EventBridge 通知配置
aws s3api put-bucket-notification-configuration \
  --bucket my-contract-intelligence-bucket \
  --notification-configuration '{
    "EventBridgeConfiguration": {}
  }'

import json
import boto3
import uuid
from datetime import datetime

s3_client = boto3.client("s3")
bedrock_client = boto3.client("bedrock-runtime")
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("ContractIntelligence")

EXTRACTION_SCHEMA = {
    "contract_title": "合同标题",
    "effective_date": "生效日期",
    "expiration_date": "到期日期",
    "counterparty": "对方当事人名称",
    "contract_value": "合同金额及币种",
    "termination_clause_summary": "终止条款摘要",
    "key_obligations": "主要义务列表",
    "risk_flags": "风险标记（如违约金、竞业限制等）",
}

def build_prompt(raw_text: str) -> str:
    schema_desc = json.dumps(EXTRACTION_SCHEMA, ensure_ascii=False, indent=2)
    return f"""你是一名合同分析专家。请从以下合同文本中提取关键信息，严格按照 JSON schema 输出。
只输出 JSON，不要添加任何解释文字。如果某字段在文本中找不到，填 null。

Schema:
{schema_desc}

合同文本:
{raw_text}"""

def lambda_handler(event, context):
    for record in event.get("Records", []):
        bucket = record["s3"]["bucket"]["name"]
        key = record["s3"]["object"]["key"]

        # 1. 从 S3 读取合同文本（假设已预处理为纯文本）
        response = s3_client.get_object(Bucket=bucket, Key=key)
        raw_text = response["Body"].read().decode("utf-8")

        # 2. 调用 Bedrock Claude 3 Sonnet 提取
        prompt = build_prompt(raw_text)
        bedrock_response = bedrock_client.invoke_model(
            modelId="anthropic.claude-3-sonnet-20240229-v1:0",
            body=json.dumps({
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens": 4096,
                "messages": [{"role": "user", "content": prompt}],
            }),
            contentType="application/json",
            accept="application/json",
        )
        result_body = json.loads(bedrock_response["body"].read())
        extracted_json = json.loads(result_body["content"][0]["text"])

        # 3. 写入 DynamoDB
        contract_id = str(uuid.uuid4())
        upload_date = datetime.utcnow().strftime("%Y-%m-%d")
        item = {
            "contractId": contract_id,
            "uploadDate": upload_date,
            "sourceFile": key,
            **extracted_json,
        }
        table.put_item(Item=item)

        print(f"Contract {key} extracted → contractId={contract_id}")

    return {"statusCode": 200, "body": json.dumps("Processed")}

Comment: "Contract intelligence pipeline"
StartAt: ExtractText
States:
  ExtractText:
    Type: Task
    Resource: arn:aws:states:::textract:startDocumentAnalysis
    Parameters:
      S3Bucket.$: "$.bucket"
      S3Key.$: "$.key"
      FeatureTypes:
        - TABLES
        - FORMS
    Next: RunLLMExtraction
    Retry:
      - ErrorEquals:
          - States.ALL
        IntervalSeconds: 10
        MaxAttempts: 3
  RunLLMExtraction:
    Type: Task
    Resource: arn:aws:lambda:us-east-1:123456789012:function:ContractLLMExtract
    Next: ValidateOutput
  ValidateOutput:
    Type: Task
    Resource: arn:aws:lambda:us-east-1:123456789012:function:ValidateContractSchema
    End: true
    Catch:
      - ErrorEquals:
          - States.ALL
        Next: LogValidationFailure
  LogValidationFailure:
    Type: Task
    Resource: arn:aws:lambda:us-east-1:123456789012:function:LogFailure
    End: true