先说结论：JSON能解析，不代表结果能用

模型返回下面这样的内容，json.loads()不会报错：

{"summary":"接口已经恢复","events":[],"needs_review":false}

但如果原文里有三条观测记录，这个结果至少有两个问题：事件被漏掉了，“已经恢复”也可能只是把一条带问号的备注当成了事实。

我更愿意把Claude放在管道中间，而不是放在管道终点：

原始文本
   ↓ 计算哈希、提取可核对事实
Claude结构化输出
   ↓ 字段/类型/枚举检查
   ↓ 数量、数值、哈希一致性检查
PASS ─────────→ 交给后续程序
FAIL ─────────→ HOLD，人工处理

这篇文章用一段很短的监控文本做演示。重点不是监控，而是一个可迁移的原则：让模型负责整理，让程序负责证明它没有随意改动输入。

1. Prompt要求JSON，为什么仍然不够

常见写法是：

请把下面的文本提取成JSON，不要输出多余内容。

它解决的是“希望输出长什么样”，没有解决以下问题：

1. 必填字段是否存在，数组里每个元素的类型是否一致；
2. 原文有三条记录时，模型有没有只提取两条；
3. 原文中的4.2%有没有被改成42%或变成一句描述；
4. 原文写着“recovered? unconfirmed”时，模型有没有把它写成确定事实；
5. 响应是否因为达到max_tokens而只剩半个JSON。

Anthropic当前文档把这类需求分成两层：JSON输出可以约束响应格式，应用侧仍要做业务语义校验；如果响应被截断或拒答，也不能按正常结果处理。这里的“结构化”是第一道门，不是事实鉴定器。

2. 示例输入：故意放一个容易被误读的问号

2026-08-27 09:00 api-gateway 502_rate=0.8% requests=12000
2026-08-27 09:05 api-gateway 502_rate=4.2% requests=11800
2026-08-27 09:10 api-gateway 502_rate=1.1% requests=11900
note: service recovered? unconfirmed

目标不是让Claude计算峰值，也不是让它判断系统是否真的恢复，而是让它整理出三条原始观测，并把“恢复”标成未确认。

我给输出规定了这些字段：

字段	用途	校验方式
source_hash	证明结果对应哪一份原文	必须等于Python现场计算的SHA-256
events	保留原文中的观测记录	时间戳集合和源文本一致
error_rate	保留源数据中的数值	与源文本逐条精确比较
status	区分observed和unconfirmed	只允许枚举值
needs_review	标记是否要人工看	出现未确认信息时必须为true

这里有一个取舍：摘要可以由模型生成，但数字和状态不能只相信摘要。后续程序应该使用events，而不是从summary里重新猜数字。

3. 完整Python代码

下面代码有两个入口：

• --dry-run不调用网络，用固定候选结果跑完整校验链；
• --live才会读取ANTHROPIC_API_KEY和ANTHROPIC_MODEL调用Claude。

没有密钥时也能先验证本地逻辑。代码不会把密钥写进文件，也没有发送消息、修改文件或调用外部写入接口。

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
from typing import Any


SOURCE_TEXT = """2026-08-27 09:00 api-gateway 502_rate=0.8% requests=12000
2026-08-27 09:05 api-gateway 502_rate=4.2% requests=11800
2026-08-27 09:10 api-gateway 502_rate=1.1% requests=11900
note: service recovered? unconfirmed
"""

EVENT_RE = re.compile(
    r"^(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}) "
    r"(?P<component>[\w-]+) 502_rate=(?P<rate>\d+(?:\.\d+)?)% "
    r"requests=(?P<requests>\d+)$"
)

OUTPUT_SCHEMA: dict[str, Any] = {
    "type": "object",
    "additionalProperties": False,
    "properties": {
        "source_hash": {"type": "string"},
        "summary": {"type": "string"},
        "events": {
            "type": "array",
            "items": {
                "type": "object",
                "additionalProperties": False,
                "properties": {
                    "timestamp": {"type": "string"},
                    "component": {"type": "string"},
                    "error_rate": {"type": ["number", "null"]},
                    "status": {"type": "string", "enum": ["observed", "unconfirmed"]},
                    "evidence": {"type": "string"},
                },
                "required": ["timestamp", "component", "error_rate", "status", "evidence"],
            },
        },
        "unknowns": {"type": "array", "items": {"type": "string"}},
        "needs_review": {"type": "boolean"},
    },
    "required": ["source_hash", "summary", "events", "unknowns", "needs_review"],
}

REQUIRED_KEYS = set(OUTPUT_SCHEMA["required"])
EVENT_KEYS = {"timestamp", "component", "error_rate", "status", "evidence"}


def source_hash(source_text: str) -> str:
    return hashlib.sha256(source_text.encode("utf-8")).hexdigest()


def source_facts(source_text: str) -> dict[str, float]:
    """Extract numeric facts locally; do not ask the model to recalculate them."""
    facts: dict[str, float] = {}
    for line in source_text.splitlines():
        match = EVENT_RE.match(line.strip())
        if match:
            facts[match.group("timestamp")] = float(match.group("rate"))
    if not facts:
        raise ValueError("源文本没有匹配到任何观测记录")
    return facts


def build_messages(source_text: str, digest: str) -> tuple[str, str]:
    system = """你是一个事实整理器，不是监控告警器。
只依据 <source_text> 中明确出现的内容，不补写原因、恢复结论或未出现的数字。
把每条带时间戳的观测原样整理到 events；note 中带问号或 unconfirmed 的内容只能进入 unknowns，不能当作已确认事实。
source_hash 必须原样复制给定的哈希。只输出符合JSON Schema的JSON对象，不要输出Markdown围栏或解释。"""
    user = f"""<task>把源文本整理成结构化结果。不要计算新的指标。</task>
<source_hash>{digest}</source_hash>
<source_text>
{source_text}
</source_text>"""
    return system, user


def validate_result(payload: Any, source_text: str) -> dict[str, Any]:
    """Schema-like and semantic checks. Any failure stops the pipeline."""
    if not isinstance(payload, dict):
        raise ValueError("模型结果必须是JSON对象")
    if set(payload) != REQUIRED_KEYS:
        raise ValueError(f"顶层字段不匹配: {sorted(set(payload) ^ REQUIRED_KEYS)}")
    if payload["source_hash"] != source_hash(source_text):
        raise ValueError("source_hash不匹配，结果可能对应另一份输入")
    if not isinstance(payload["summary"], str) or not payload["summary"].strip():
        raise ValueError("summary必须是非空字符串")
    if not isinstance(payload["events"], list):
        raise ValueError("events必须是数组")
    if not isinstance(payload["unknowns"], list) or not all(
        isinstance(item, str) for item in payload["unknowns"]
    ):
        raise ValueError("unknowns必须是字符串数组")
    if not isinstance(payload["needs_review"], bool):
        raise ValueError("needs_review必须是布尔值")

    expected_facts = source_facts(source_text)
    seen_timestamps: set[str] = set()
    has_unconfirmed = False
    for event in payload["events"]:
        if not isinstance(event, dict) or set(event) != EVENT_KEYS:
            raise ValueError("events中存在字段缺失或多余的对象")
        timestamp = event["timestamp"]
        if timestamp not in expected_facts:
            raise ValueError(f"事件时间戳不在源文本中: {timestamp}")
        if timestamp in seen_timestamps:
            raise ValueError(f"事件重复: {timestamp}")
        seen_timestamps.add(timestamp)
        if not isinstance(event["component"], str) or not event["component"].strip():
            raise ValueError("component必须是非空字符串")
        if event["status"] not in {"observed", "unconfirmed"}:
            raise ValueError("status不是允许的枚举值")
        rate = event["error_rate"]
        if not isinstance(rate, (int, float)) or isinstance(rate, bool):
            raise ValueError("error_rate必须是数字")
        if not 0 <= rate <= 100:
            raise ValueError("error_rate超出百分比范围")
        if rate != expected_facts[timestamp]:
            raise ValueError(f"{timestamp}的error_rate与源文本不一致")
        if not isinstance(event["evidence"], str) or not event["evidence"].strip():
            raise ValueError("evidence必须保留证据说明")
        has_unconfirmed |= event["status"] == "unconfirmed"

    if seen_timestamps != set(expected_facts):
        raise ValueError("events没有覆盖源文本中的全部观测记录")
    if (has_unconfirmed or payload["unknowns"]) and not payload["needs_review"]:
        raise ValueError("存在未确认信息时，needs_review必须为true")
    return payload


def dry_run(source_text: str) -> dict[str, Any]:
    digest = source_hash(source_text)
    candidate = {
        "source_hash": digest,
        "summary": "09:05的502错误率升至4.2%；备注中的恢复状态仍未确认。",
        "events": [
            {
                "timestamp": "2026-08-27 09:00",
                "component": "api-gateway",
                "error_rate": 0.8,
                "status": "observed",
                "evidence": "源文本记录502_rate=0.8%。",
            },
            {
                "timestamp": "2026-08-27 09:05",
                "component": "api-gateway",
                "error_rate": 4.2,
                "status": "observed",
                "evidence": "源文本记录502_rate=4.2%。",
            },
            {
                "timestamp": "2026-08-27 09:10",
                "component": "api-gateway",
                "error_rate": 1.1,
                "status": "observed",
                "evidence": "源文本记录502_rate=1.1%。",
            },
        ],
        "unknowns": ["service recovered"],
        "needs_review": True,
    }
    return validate_result(candidate, source_text)


def call_claude(source_text: str) -> dict[str, Any]:
    try:
        from anthropic import Anthropic
    except ImportError as exc:
        raise RuntimeError("实时模式需要先安装 anthropic：pip install anthropic") from exc

    api_key = os.getenv("ANTHROPIC_API_KEY")
    model = os.getenv("ANTHROPIC_MODEL")
    if not api_key or not model:
        raise RuntimeError("实时模式需要设置 ANTHROPIC_API_KEY 和 ANTHROPIC_MODEL")

    system, user = build_messages(source_text, source_hash(source_text))
    client = Anthropic(api_key=api_key)
    response = client.messages.create(
        model=model,
        max_tokens=1600,
        system=system,
        messages=[{"role": "user", "content": user}],
        output_config={
            "format": {"type": "json_schema", "schema": OUTPUT_SCHEMA}
        },
    )
    if getattr(response, "stop_reason", None) != "end_turn":
        raise RuntimeError(f"模型没有正常结束: {getattr(response, 'stop_reason', None)}")
    text_parts = [
        block.text for block in response.content
        if getattr(block, "type", None) == "text"
    ]
    if not text_parts:
        raise ValueError("响应中没有文本JSON")
    return validate_result(json.loads("".join(text_parts)), source_text)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--dry-run", action="store_true", help="不调用Claude，只测试校验器")
    parser.add_argument("--live", action="store_true", help="调用Claude API")
    args = parser.parse_args()
    if args.dry_run == args.live:
        parser.error("二选一：--dry-run 或 --live")

    result = dry_run(SOURCE_TEXT) if args.dry_run else call_claude(SOURCE_TEXT)
    print(json.dumps(result, ensure_ascii=False, indent=2))
    print("pipeline=PASS")
    print(f"next_action={'HUMAN_REVIEW' if result['needs_review'] else 'READ_ONLY'}")


if __name__ == "__main__":
    main()

4. 运行结果：通过的不是“模型很聪明”，而是证据链完整

先运行不联网模式：

python claude_contract_pipeline.py --dry-run

输出会包含：

"events": [
  {"timestamp": "2026-08-27 09:00", "error_rate": 0.8, "status": "observed"},
  {"timestamp": "2026-08-27 09:05", "error_rate": 4.2, "status": "observed"},
  {"timestamp": "2026-08-27 09:10", "error_rate": 1.1, "status": "observed"}
],
"unknowns": ["service recovered"],
"needs_review": true
pipeline=PASS
next_action=HUMAN_REVIEW

干跑只验证本地校验器，不伪装成一次真实API调用。要测试实时路径，再安装SDK并设置密钥：

python -m pip install anthropic

# Linux / macOS
export ANTHROPIC_API_KEY="你的密钥"
export ANTHROPIC_MODEL="你的模型标识"
python claude_contract_pipeline.py --live

Windows PowerShell：

$env:ANTHROPIC_API_KEY = "你的密钥"
$env:ANTHROPIC_MODEL = "你的模型标识"
python .\claude_contract_pipeline.py --live

注意，实时模式的“成功”必须同时满足两件事：API返回正常结束，并且本地validate_result()通过。前者只说明响应完成，后者才说明它仍然对应当前这份原文。

5. 故意制造三个失败测试

只测成功路径没有意义。把下面几种结果送入validate_result()，都应该阻断：

故障	应该被哪道门拦住
source_hash来自上一份输入	哈希一致性
09:05的error_rate被写成42.0	数值一致性和百分比范围
只返回两条events	时间戳集合完整性
status="recovered"	枚举值
needs_review=false，但unknowns里有未确认事项	业务语义规则
响应stop_reason="max_tokens"	API完成状态

这也是为什么代码里没有“校验失败就删掉问题字段再继续”的逻辑。删掉字段会让程序看起来更顺，却把错误藏起来。数据管道应该明确返回FAIL或HOLD，由上游决定是否重试、换输入或交给人看。

6. output_config和本地校验各自负责什么

Anthropic的Structured Outputs可以让Claude按给定JSON Schema输出，适合解决字段缺失、类型不一致和JSON语法错误这类问题。但它不能替应用回答“这个数字是否来自当前文件”“这句话是否把不确定改成了确定”。

可以把职责分成三层：

1. 模型层：根据上下文完成抽取和归纳，返回结构化对象；
2. 程序层：做哈希、字段、枚举、范围、数量和逐条数字对比；
3. 业务层：决定是否允许继续、是否需要人工复核，以及结果能不能进入下游系统。

尤其要注意stop_reason。如果响应因为达到令牌上限而截断，JSON可能不完整；如果模型拒答，响应也不是业务结果。两种情况都应该退出正常处理路径，而不是对半截文本做容错拼接。

7. 这套代码的边界

7.1 哈希只能证明“对应哪份输入”

哈希能防止结果串到另一份文本，却不能证明文本本身真实。原始数据的可信度仍然要由采集来源、签名、权限和审计记录保证。

7.2 Schema通过不等于语义通过

模型可以返回完全符合Schema的summary，但摘要仍可能遗漏上下文。因此关键数字要保留结构化字段和原文证据，不要让后续程序从自然语言摘要反向提取数字。

7.3 不要把示例直接当生产告警器

示例只覆盖固定格式的三条记录。真实日志可能有时区、重复事件、科学计数法、缺失值和字段版本变化。上线前应补充契约测试、重试上限、超时、日志脱敏和输入版本管理。

7.4 密钥和原文都不能随便进日志

密钥只从环境变量或密钥管理系统读取；生产日志不要打印完整Prompt、完整响应和敏感原文。本文的监控文本是脱敏测试数据，不代表任何真实系统状态。

8. 总结

这次实践真正有用的不是“让Claude输出JSON”这一句Prompt，而是把信任边界画清楚：

• JSON Schema负责让输出可解析；
• Python数据契约负责检查结构和输入事实是否对得上；
• 业务规则负责决定PASS、HOLD还是人工复核。

如果只记住一句话：模型可以帮你整理信息，但不要让它自己证明信息是真的。

参考资料

1. Claude Platform Docs：Structured outputs
2. Claude Platform Docs：Using the Messages API

本文示例只做结构化抽取和本地校验，不执行告警发送、文件写入或其他外部动作。实时API的参数和可用模型请以官方文档及当前账户配置为准。