文書から構造化要約を生成する

はじめに

運用チームは短いインシデント報告の構造化要約を必要としています。モデルはもっともらしい文章を生成できますが、アプリケーションには予測可能なフィールドが必要で、無効な出力は拒否しなければなりません。Bedrock Converse で JSON 要約を要求し、Python で解析し、明示されたローカルテストケースで拒否経路を確認します。

Python 入門、JSON、前のラボの Converse リクエストを理解している必要があります。各新規 VM は専用の文書、設定済み ID、推論枠を提供します。前の VM のファイルや認証情報は不要です。

認定試験との関連

文書と必須フィールドを確認する

このステップでは提供された合成報告を確認し、要約を使うアプリケーションが期待する構造を定義します。

ワークスペースに移動します：

cd /home/labex/project

文書はプレーンテキストです。ファイルを表示する cat で読みます：

cat incident.txt

文書のインシデントは INC-204 です。チェックアウト設定の誤りで 30 分間処理が失敗しました。チームは設定を戻してサービスを復旧しました。追加の設定テストは予定されていますが、未完了です。要約はこの事実の範囲に収めます。

アプリケーションは正確に 3 つのフィールド、incident_id、impact、next_action を期待し、各値は空でない文字列でなければなりません。スキーマ はこの形を示します。有効な JSON だけでは不十分で、リスト、欠けたフィールド、数値でも有効な JSON になり得ます。提供されたスキーマ参照を読みます：

cat summary-schema.json

参照は希望する出力を説明します。通常の Python でフィールドを強制検証しますが、モデルが従うことを自動的に保証するものではありません。JSON を要求することと検証することは別の責任です。

構造化要約を生成して検証する

このステップでは要約を要求し、成功を表示する前に返されたテキストを検証する小さな Python コマンドを作ります。

boto3 は AWS の Python SDK です。提供されたランタイムにインストール済みで、設定済みの認証情報とサービスエンドポイントを使います。コードは CLI と同じ操作の converse を使います。maxTokens で出力を制限し、曖昧な失敗後に別の推論を黙って送らないよう自動再試行を無効にします。

次のコードで summary.py を作ります。parse_response は終了理由を確認し、json.loads で解析し、正確なキーと型を検証します。ValueError は制御された拒否を示します。ClientError や接続エラーは、認証情報や完全な例外を表示せず報告します。任意の --response-file モードはローカル応答を読んで解析器をテストし、推論は送信しません：

cat > summary.py <<'EOF'
import argparse
import json
from pathlib import Path
import boto3
from botocore.config import Config
from botocore.exceptions import BotoCoreError, ClientError


def parse_response(response):
    if not isinstance(response, dict):
        raise ValueError("invalid_response")
    if response.get("stopReason") != "end_turn":
        raise ValueError("incomplete_response")
    try:
        text = response["output"]["message"]["content"][0]["text"]
        value = json.loads(text)
        expected = {"incident_id", "impact", "next_action"}
        if not isinstance(value, dict) or set(value) != expected:
            raise ValueError("invalid_fields")
        if any(not isinstance(v, str) or not v.strip() for v in value.values()):
            raise ValueError("invalid_field_type")
        if value["incident_id"] != "INC-204":
            raise ValueError("unexpected_incident")
        return value
    except (KeyError, IndexError, TypeError, json.JSONDecodeError) as error:
        raise ValueError("invalid_response") from error


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--response-file", type=Path)
    args = parser.parse_args()
    try:
        if args.response_file:
            response = json.loads(args.response_file.read_text())
        else:
            client = boto3.client("bedrock-runtime", config=Config(
                connect_timeout=5, read_timeout=150,
                retries={"total_max_attempts": 1}))
            response = client.converse(
                modelId="labex.text-v1:0",
                system=[{"text": "Summarize only facts supplied in the incident document. Return exactly one JSON object with string fields incident_id, impact and next_action. No Markdown fences, extra keys or surrounding prose. Do not turn planned work into completed work."}],
                messages=[{"role": "user", "content": [
                    {"text": Path("incident.txt").read_text()}]}],
                inferenceConfig={"maxTokens": 768, "temperature": 0})
            Path("response.json").write_text(json.dumps(response, indent=2))
        value = parse_response(response)
    except (ValueError, KeyError, TypeError):
        print(json.dumps({"status": "rejected", "reason": "invalid_model_output"}))
        return 2
    except (BotoCoreError, ClientError):
        print(json.dumps({"status": "unavailable", "reason": "inference_failed"}))
        return 3
    print(json.dumps({"status": "ok", "summary": value}, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
EOF

インストール済みの AWS Python ランタイムでスクリプトを実行します。最後の > summary.json は表示される結果を保存します：

/opt/labex/aws/venv/bin/python summary.py > summary.json

プロンプトを待ち、結果を確認します：

cat summary.json

成功した実行には "status": "ok" と、3 つの必須文字列を含む summary オブジェクトがあります。元の Converse 応答も読みます：

python3 -m json.tool response.json

response.json は実際のモデル出力、終了理由、使用量を記録します。上部の AWS View で完了リクエストを展開してテキストを比較します。表現は変わる場合があります。impact がチェックアウト失敗、next_action が予定された作業を示すか確認します。スキーマに合うだけでは内容の正確さは証明できません。

拒否が報告されたら、再試行前に元の応答を読みます。モデルは出力を切り詰めたり、形式を無視したりする場合があります。成功結果を固定値で置き換えないでください。失敗した推論でも枠を消費する場合があります。

無効な応答の境界をテストする

このステップでは壊れた JSON、誤った型、切り詰められた応答が、追加のモデル呼び出しなしで拒否されることを確認します。

fixtures/ には意図的な解析器テスト入力があります。ローカルテストデータであり、推論結果ではありません。一覧にします：

ls fixtures

壊れた JSON のケースでスクリプトを実行します：

/opt/labex/aws/venv/bin/python summary.py --response-file fixtures/malformed.json

期待する出力は {"status": "rejected", "reason": "invalid_model_output"} です。終了コード 2 はこの拒否を示します。echo $? は直前のコマンドの終了コードを表示します：

echo $?

有効な JSON でもスキーマに違反できます。impact が数値のケースをテストします：

/opt/labex/aws/venv/bin/python summary.py --response-file fixtures/wrong-type.json

同じ拒否になるはずです。モデルがトークン上限に達した場合、完全に見える JSON オブジェクトでも拒否すべきです。この境界をテストします：

/opt/labex/aws/venv/bin/python summary.py --response-file fixtures/truncated.json

stopReason が max_tokens のため拒否になるはずです。これらは成功要約を表示せず、AWS View にリクエストを追加しません。

最後に保存された実際の応答をもう一度解析します：

/opt/labex/aws/venv/bin/python summary.py --response-file response.json

引き続き status: ok になるはずです。有効な実際の応答と、制御された失敗入力の両方をテストしました。解析器ケースを推論成功の証拠と混同しないでください。完了リクエストと元の応答がそれを別に証明します。

自分で作成した演習成果物を削除する

このステップでは機能確認後にスクリプトと 2 つの結果ファイルを削除します。

提供された文書、スキーマ、テストケースは新規環境のものです。バックエンドとともに残してください。rm は指定された 3 つの成果物のみを削除します：

rm summary.py summary.json response.json

提供された入力が残っているか確認します：

ls

Converse は削除すべき永続的なクラウドサーバーを作りません。ローカルファイルを消しても credits は戻らず、AWS View の実際の推論記録も消えません。

まとめ

実際の Bedrock Converse 応答から構造化要約を作り、フィールド名と型を検証し、終了理由を確認して無効な応答を拒否しました。また、事実の確認をスキーマ検証と分け、自分のファイルのみを削除しました。

詳しくは、AWS 公式ドキュメントを参照してください：Converse の応答内容、停止理由、使用量。