문서에서 구조화 요약 생성하기

소개

운영팀은 짧은 사고 보고서의 구조화 요약이 필요합니다. 모델은 그럴듯한 글을 만들지만 애플리케이션은 예측 가능한 필드를 요구하며 잘못된 출력을 거부해야 합니다. Bedrock Converse로 JSON 요약을 요청하고 Python으로 파싱하며 명확히 표시된 로컬 사례로 거부 경로를 테스트합니다.

Python 기초, JSON, 이전 랩의 Converse 요청을 알아야 합니다. 각 새 VM은 자체 문서, 설정된 ID, 추론 한도를 제공합니다. 이전 VM의 파일이나 자격 증명은 필요하지 않습니다.

자격증 관련성

문서와 필수 필드 확인하기

이 단계에서는 제공된 합성 보고서를 확인하고 요약을 사용할 애플리케이션의 구조를 정의합니다.

작업 공간에서 시작합니다:

cd /home/labex/project

문서는 일반 텍스트입니다. 파일을 출력하는 cat으로 읽습니다:

cat incident.txt

사고는 INC-204입니다. 결제 설정 오류로 30분 동안 결제가 실패했습니다. 팀은 설정을 되돌리고 서비스를 복구했습니다. 후속 설정 테스트는 예정되어 있지만 완료되지 않았습니다. 요약은 이 사실의 범위에 머물러야 합니다.

애플리케이션은 정확히 세 필드인 incident_id, impact, next_action을 기대하며 각각 비어 있지 않은 문자열이어야 합니다. 스키마는 이 형태를 설명합니다. 목록이나 누락된 필드, 숫자도 유효한 JSON일 수 있으므로 JSON 유효성만으로 충분하지 않습니다. 제공된 스키마 참조를 읽습니다:

cat summary-schema.json

참조는 원하는 출력을 설명합니다. 일반 Python으로 필드를 강제 검증하며, 모델이 자동으로 준수한다는 약속은 아닙니다. JSON 요청과 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"와 세 필수 문자열을 가진 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

같은 거부를 보고해야 합니다. 완성되어 보이는 객체도 모델이 토큰 한도에 도달했다면 거부해야 합니다. 이 경계를 테스트합니다:

/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여야 합니다. 유효한 실제 응답과 제어된 실패 입력을 모두 테스트했습니다. 파서 사례를 추론 성공의 증거와 혼동하지 마세요. 완료 요청과 원본 응답이 별도로 입증합니다.

직접 만든 연습 산출물 삭제하기

이 단계에서는 기능 확인 후 자신의 스크립트와 두 결과 파일을 삭제합니다.

문서, 스키마, 파서 사례는 새 환경에 제공된 자료입니다. 백엔드 서비스와 함께 보존하세요. rm은 지정된 세 학습자 파일만 삭제합니다:

rm summary.py summary.json response.json

제공된 입력이 남아 있는지 확인합니다:

ls

Converse는 삭제할 영구 클라우드 서버를 만들지 않습니다. 로컬 파일 삭제는 credits를 환불하거나 AWS View의 실제 기록을 지우지 않습니다.

요약

실제 Bedrock Converse 응답으로 구조화 요약을 생성하고 필드명과 타입, 종료 이유를 확인했으며 잘못된 응답을 거부했습니다. 사실 검토를 스키마 검증과 분리하고 직접 만든 파일만 삭제했습니다.

자세한 내용은 AWS 공식 문서를 참고하세요: Converse 응답 내용, 중지 사유 및 사용량.