Amazon Connect Customerで音声・チャット終了後にリアルタイム文字起こしを要約してコンタクト属性へ保存してみた

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "InvokeBedrockModel",
      "Effect": "Allow",
      "Action": "bedrock:InvokeModel",
      "Resource": "*"
    },
    {
      "Sid": "ReadAndUpdateConnectContact",
      "Effect": "Allow",
      "Action": [
        "connect:ListRealtimeContactAnalysisSegments",
        "connect:ListRealtimeContactAnalysisSegmentsV2",
        "connect:UpdateContactAttributes"
      ],
      "Resource": "arn:aws:connect:ap-northeast-1:111111111111:instance/3ff2093d-af96-43fd-b038-3c07cdd7609c/contact/*"
    }
  ]
}

import json
import os
import time

import boto3

MODEL_ID = os.environ["MODEL_ID"]

MAX_RESULTS = 100
CHAT_OUTPUT_TYPE = "Raw"
CHAT_SEGMENT_TYPES = ["Transcript"]

EMPTY_RESULT_RETRY_COUNT = 3
EMPTY_RESULT_RETRY_SECONDS = 1

connect_lens_client = boto3.client("connect-contact-lens")
connect_client = boto3.client("connect")
bedrock_client = boto3.client("bedrock-runtime")

def validate_and_extract_ids(detail: dict) -> tuple[str, str, str, str]:
    contact_id = detail.get("contactId")
    initial_contact_id = detail.get("initialContactId", contact_id)
    instance_arn = detail.get("instanceArn")
    channel = detail.get("channel")

    if not contact_id or not initial_contact_id or not instance_arn or not channel:
        raise ValueError("Event detail is missing required fields.")

    try:
        instance_id = instance_arn.rsplit("/", maxsplit=1)[-1]
    except IndexError as error:
        raise ValueError(f"Invalid instanceArn format: {instance_arn}") from error

    return contact_id, initial_contact_id, instance_id, channel.upper()

def extract_transcript(segment: dict, sort_key) -> dict | None:
    transcript = segment.get("Transcript")

    if not transcript:
        return None

    content = transcript.get("Content", "").strip()

    if not content:
        return None

    return {
        "participant": transcript.get("ParticipantRole", "UNKNOWN"),
        "content": content,
        "sort_key": sort_key,
    }

def fetch_voice_transcripts(instance_id: str, contact_id: str) -> list[dict]:
    transcripts = []
    next_token = None

    print(f"Fetching voice transcripts. ContactId={contact_id}")

    while True:
        params = {
            "InstanceId": instance_id,
            "ContactId": contact_id,
            "MaxResults": MAX_RESULTS,
        }

        if next_token:
            params["NextToken"] = next_token

        response = connect_lens_client.list_realtime_contact_analysis_segments(
            **params
        )

        for segment in response.get("Segments", []):
            transcript = segment.get("Transcript", {})
            sort_key = transcript.get("BeginOffsetMillis", 0)

            normalized_transcript = extract_transcript(segment, sort_key)

            if normalized_transcript:
                transcripts.append(normalized_transcript)

        next_token = response.get("NextToken")

        if not next_token:
            break

    return sorted(transcripts, key=lambda item: item["sort_key"])

def fetch_chat_transcripts(instance_id: str, contact_id: str) -> list[dict]:
    transcripts = []
    next_token = None

    print(f"Fetching chat transcripts. ContactId={contact_id}")

    while True:
        params = {
            "InstanceId": instance_id,
            "ContactId": contact_id,
            "MaxResults": MAX_RESULTS,
            "OutputType": CHAT_OUTPUT_TYPE,
            "SegmentTypes": CHAT_SEGMENT_TYPES,
        }

        if next_token:
            params["NextToken"] = next_token

        response = connect_client.list_realtime_contact_analysis_segments_v2(
            **params
        )

        for segment in response.get("Segments", []):
            transcript = segment.get("Transcript", {})
            absolute_time = transcript.get("Time", {}).get("AbsoluteTime", "")

            normalized_transcript = extract_transcript(segment, absolute_time)

            if normalized_transcript:
                transcripts.append(normalized_transcript)

        next_token = response.get("NextToken")

        if not next_token:
            break

    return sorted(transcripts, key=lambda item: item["sort_key"])

def fetch_transcripts_with_retry(
    channel: str,
    instance_id: str,
    contact_id: str,
) -> list[dict]:
    for attempt in range(EMPTY_RESULT_RETRY_COUNT):
        if channel == "VOICE":
            transcripts = fetch_voice_transcripts(instance_id, contact_id)
        elif channel == "CHAT":
            transcripts = fetch_chat_transcripts(instance_id, contact_id)
        else:
            raise ValueError(f"Unsupported channel: {channel}")

        if transcripts:
            return transcripts

        if attempt < EMPTY_RESULT_RETRY_COUNT - 1:
            print(
                "No transcripts found. "
                f"Retrying after {EMPTY_RESULT_RETRY_SECONDS} second(s)."
            )
            time.sleep(EMPTY_RESULT_RETRY_SECONDS)

    return []

def generate_summary(full_text: str) -> str:
    prompt = f"""
以下の会話ログを要約してください。

<conversation>
{full_text}
</conversation>

要約の条件:
1. 「【問い合わせ】... / 【対応】... / 【結論】...」の形式で出力すること。
2. 改行を含めず、1行で出力すること。
3. 簡潔にまとめること。
4. 会話ログにない内容を補完しないこと。
"""

    response = bedrock_client.converse(
        modelId=MODEL_ID,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "text": prompt
                    }
                ],
            }
        ],
        inferenceConfig={
            "maxTokens": 1000,
            "temperature": 0.0,
        },
    )

    content_list = response["output"]["message"]["content"]

    if not content_list:
        raise RuntimeError("Bedrock returned an empty response.")

    summary = content_list[0].get("text", "").replace("\n", " ").strip()

    if not summary:
        raise RuntimeError("Generated summary is empty.")

    return summary

def update_contact_attributes(
    instance_id: str,
    initial_contact_id: str,
    summary: str,
) -> None:
    connect_client.update_contact_attributes(
        InstanceId=instance_id,
        InitialContactId=initial_contact_id,
        Attributes={
            "realtimesummary": summary
        },
    )

    print(
        "Contact attributes updated. "
        f"InitialContactId={initial_contact_id}"
    )

def lambda_handler(event, context):
    print(json.dumps(event, ensure_ascii=False))

    detail = event.get("detail", {})

    contact_id, initial_contact_id, instance_id, channel = validate_and_extract_ids(
        detail
    )

    print(
        "Processing contact. "
        f"ContactId={contact_id}, "
        f"InitialContactId={initial_contact_id}, "
        f"Channel={channel}"
    )

    transcripts = fetch_transcripts_with_retry(
        channel=channel,
        instance_id=instance_id,
        contact_id=contact_id,
    )

    if not transcripts:
        print("No transcripts found.")
        return {
            "statusCode": 200,
            "body": "No transcripts found.",
        }

    full_text = "\n".join(
        f"[{item['participant']}] {item['content']}"
        for item in transcripts
    )

    summary = generate_summary(full_text)

    print("Generated summary:")
    print(summary)

    update_contact_attributes(
        instance_id=instance_id,
        initial_contact_id=initial_contact_id,
        summary=summary,
    )

    return {
        "statusCode": 200,
        "body": "Success",
    }

{
  "Channel": "CHAT",
  "Status": "COMPLETED",
  "Segments": [
    {
      "Transcript": {
        "Id": "11111111-2222-3333-4444-555555555555",
        "ParticipantId": "participant-id-1",
        "ParticipantRole": "CUSTOMER",
        "DisplayName": "Customer Name",
        "Content": "こんにちは、パスワードをリセットしたいのですが。",
        "ContentType": "text/plain",
        "Time": {
          "AbsoluteTime": "2026-01-14T23:51:30.000Z"
        },
        "Sentiment": "NEUTRAL"
      }
    },
    {
      "Transcript": {
        "Id": "22222222-3333-4444-5555-666666666666",
        "ParticipantId": "agent-id-1",
        "ParticipantRole": "AGENT",
        "DisplayName": "Support Agent",
        "Content": "かしこまりました。ご本人確認のため、生年月日をお願いします。",
        "ContentType": "text/plain",
        "Time": {
          "AbsoluteTime": "2026-01-14T23:51:35.000Z"
        },
        "Sentiment": "POSITIVE"
      }
    }
  ]
}

[
    {
        "participant": "CUSTOMER",
        "content": "こんにちは、パスワードをリセットしたいのですが。",
        "sort_key": "2026-01-14T23:51:30.000Z"
    },
    {
        "participant": "AGENT",
        "content": "かしこまりました。ご本人確認のため、生年月日をお願いします。",
        "sort_key": "2026-01-14T23:51:35.000Z"
    }
]

[CUSTOMER] こんにちは、パスワードをリセットしたいのですが。
[AGENT] かしこまりました。ご本人確認のため、生年月日をお願いします。

{
  "source": [
    "aws.connect"
  ],
  "detail-type": [
    "Amazon Connect Contact Event"
  ],
  "detail": {
    "eventType": [
      "DISCONNECTED"
    ],
    "channel": [
      "VOICE",
      "CHAT"
    ],
    "instanceArn": [
      "arn:aws:connect:ap-northeast-1:111111111111:instance/3ff2093d-af96-43fd-b038-3c07cdd7609c"
    ]
  }
}

【問い合わせ】パスワードリセットメールのリンクをクリックしても「リンクが無効です」というエラーが表示されログインできない。複数デバイス・ブラウザで試したが同じエラーが発生。 / 【対応】エージェントがパスワードリセットリンクを再発行し、新しいリンクを顧客に送信。 / 【結論】新しいリンクでログインに成功し、問題解決。