根据文档生成结构化摘要

PythonBeginner
立即练习

介绍

运维团队需要一份简短事件报告的结构化摘要。模型可以生成看似合理的文字,但应用需要可预测的字段,并且必须拒绝无效输出。你将通过 Bedrock Converse 请求 JSON 摘要,用 Python 解析,并使用明确标注的本地测试样例检查拒绝路径。

你应了解 Python 入门知识、JSON 和上一引导式实验中的 Converse 请求。每个 fresh VM 提供自己的文档、已配置身份和推理额度,不需要之前 VM 的文件或凭据。

认证关联

认证 考试任务 练习内容
AI Practitioner (AIF-C01) 任务 3.2 指定输出格式,并在应用使用模型响应前验证它。

使用 draw.io 制作的概念图:文档进入 Bedrock Converse,经过解析和验证后得到可用摘要或受控拒绝。

查看文档和必需字段

本步骤中,你将查看提供的合成事件报告,并明确使用摘要的应用所要求的结构。

先进入工作区:

cd /home/labex/project

提供的文档是纯文本。使用打印文件内容的 cat 读取它:

cat incident.txt

文档描述事件 INC-204:结账配置错误导致结账失败,持续 30 分钟。团队回退了配置并恢复服务。后续配置测试已计划,但尚未完成。这些事实限定了摘要内容。

应用要求恰好三个字段:incident_id、impact 和 next_action,每个字段都必须是非空字符串。Schema 描述这种结构;仅有有效 JSON 不够,因为列表、缺失字段的对象或数字也可能是有效 JSON。读取提供的 schema 参考文件:

cat summary-schema.json

参考文件解释期望的输出。你将在普通 Python 代码中强制检查这些字段;它并不自动保证模型会遵守要求。提示模型输出 JSON 与验证 JSON 是两项独立责任。

生成并验证结构化摘要

本步骤中,你将构建一个小型 Python 命令,请求摘要,并在打印成功前验证返回的文本。

boto3 是 AWS 的 Python SDK。提供的 Python 运行环境已安装它,并使用现有凭据和服务端点。代码调用 converse,与 CLI 中练习的请求操作相同。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 记录实际模型输出、结束原因和 token 用量。在上方 AWS View 中展开已完成请求并比较文本。生成的措辞可能变化。检查 impact 是否反映结账失败,next_action 是否描述计划中的工作。通过 schema 检查并不证明这些陈述准确。

来自 VM 的实际 AWS View:已完成请求包含 JSON 事件摘要、token 用量和剩余额度。你的措辞、耗时和额度可能不同。

如果脚本报告拒绝,重试前先读取原始响应。模型可能截断输出或不遵循要求的格式;不要用写死的成功结果替换摘要。失败的推理也可能消耗额度。

测试无效响应边界

本步骤中,你将确认格式错误、类型错误和被截断的响应会被拒绝,而且不发送新的模型请求。

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,但仍违反 schema。测试数字类型的 impact 样例:

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

如果模型达到 token 上限,即使 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。你已经测试了可用的实际响应和受控失败输入。不要将解析器样例误当成推理成功的证据;已完成请求和原始响应会单独证明这一点。

删除本次练习产物

本步骤中,你将在功能检查通过后删除自己创建的脚本和两个生成的结果文件。

提供的文档、schema 和解析器样例属于新的环境。保留它们和后端服务。rm 只删除三个指定的学习者产物:

rm summary.py summary.json response.json

确认提供的输入仍然存在:

ls

Converse 不创建需要删除的持久云服务器。删除本地文件不会退还已消耗的 credits,也不会抹去 AWS View 保留的实际推理记录。

总结

你使用实际 Bedrock Converse 响应生成了结构化事件摘要,强制验证字段名和类型,检查完成原因,并拒绝受控的无效响应。你还将事实含义审阅与 schema 验证分开,并只清理自己创建的文件。

进一步阅读 AWS 官方资料:Converse 响应内容、停止原因与用量。