第161回では、AIなどの後続処理へ渡すCSVを許可列だけに絞りました。入力を整えても、AIが返した結果を無検査でCSV更新や集計へ渡すことはできません。真偽値が文字列になっている、許可していない分類名が混ざる、同じIDが二度返る、といった問題があるためです。
今回は、AIの分類結果を1行1オブジェクトのJSONLとして受け取り、Python標準ライブラリだけで検査します。特定のAIサービスへ通信するコードではありません。実際のレスポンスをJSONLへ変換する部分は、利用するサービスの仕様に合わせて分離します。
今回固定する結果契約
各行のJSONオブジェクトは、次の4項目だけを持つものとします。
| 項目 | 条件 |
|---|---|
record_id |
空でない文字列。前後空白、U+0000～U+001F、U+007Fを認めない。項目、型、値の検査を通過した行同士で重複させない |
category |
ポリシーで許可した文字列のいずれか |
needs_review |
JSONの真偽値であるtrueまたはfalse。文字列は不可 |
reason |
空白だけではない文字列。ポリシーで決めた長さ以内 |
不足項目だけでなく、未知の項目が増えた場合も拒否します。PythonのJSONデコーダが既定で受け入れるNaNやInfinityも拒否します。同じオブジェクト内の重複キーも、後に書かれた値で上書きせずエラーにします。
長さの判定にはPythonのlen()を使います。結合文字や絵文字の並びを、画面上の見た目どおりの1文字として数える処理ではありません。上限は後続システムと同じ数え方で決める必要があります。
各行について記録するのは、最初に検出したエラー1件です。検査結果は新しいディレクトリへ出力し、元のJSONLは変更しません。
| 生成物 | 用途 |
|---|---|
valid.jsonl |
契約を通過した行を、項目順をそろえて再出力 |
errors.jsonl |
拒否した物理行番号とエラーコードを記録。元の値は複製しない |
summary.json |
件数、判定、入力とポリシーのSHA-256を記録 |
入力が0バイトの場合は、行ごとの判定を作れません。契約違反として終了コード2を返し、作成途中の出力ディレクトリを削除します。
1．検査ポリシーを用意する
次をUTF-8のresult_policy.jsonとして保存します。UTF-8 BOMが付いていても読み込めます。
{
"schema_version": 1,
"allowed_categories": ["配送", "請求", "その他"],
"max_record_id_chars": 64,
"max_reason_chars": 200,
"max_line_bytes": 8192
}64、200、8192は、この例の設定値です。AIサービス共通の上限ではありません。実データのID仕様、確認画面、後続システムの制約から決めます。max_line_bytesは1物理行のバイト数で、改行がある場合は改行コードも含みます。
2．拒否例を含むJSONLを用意する
次をUTF-8のai_results.jsonlとして保存します。JSONLなので、配列を囲む角括弧や行末のカンマは付けません。
{"record_id":"R-001","category":"配送","needs_review":false,"reason":"到着日の確認依頼"}
{"record_id":"R-002","category":"返金","needs_review":false,"reason":"許可されていない分類"}
{"record_id":"R-003","category":"請求","needs_review":"false","reason":"真偽値が文字列"}
1行目は契約どおりです。2行目の「返金」は許可カテゴリにありません。3行目のneeds_reviewは真偽値ではなく文字列です。
3．行単位の検証ゲートを実装する
次をvalidate_ai_jsonl.pyとして保存します。Python 3.10以上と標準ライブラリだけを使います。
from __future__ import annotations
import argparse
import hashlib
import json
import shutil
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, BinaryIO, Iterator
FIELDS = {"record_id", "category", "needs_review", "reason"}
class ContractError(Exception):
pass
class RowError(Exception):
def __init__(self, code: str):
super().__init__(code)
self.code = code
@dataclass(frozen=True)
class Policy:
allowed_categories: frozenset[str]
max_record_id_chars: int
max_reason_chars: int
max_line_bytes: int
def has_surrogate(text: str) -> bool:
return any(0xD800 <= ord(char) <= 0xDFFF for char in text)
def strict_object(pairs: list[tuple[str, object]]) -> dict[str, object]:
result: dict[str, object] = {}
for key, value in pairs:
if key in result:
raise RowError("DUPLICATE_KEY")
result[key] = value
return result
def reject_constant(_: str) -> object:
raise RowError("NON_STANDARD_NUMBER")
def load_policy(path: Path) -> tuple[Policy, str]:
raw = path.read_bytes()
digest = hashlib.sha256(raw).hexdigest()
try:
text = raw.decode("utf-8-sig")
except UnicodeDecodeError as exc:
raise ContractError("policy must be UTF-8") from exc
try:
data = json.loads(
text,
object_pairs_hook=strict_object,
parse_constant=reject_constant,
)
except (ValueError, RecursionError, RowError) as exc:
raise ContractError("policy is not valid strict JSON") from exc
expected = {
"schema_version",
"allowed_categories",
"max_record_id_chars",
"max_reason_chars",
"max_line_bytes",
}
if type(data) is not dict or set(data) != expected:
raise ContractError("policy fields do not match the required schema")
if type(data["schema_version"]) is not int or data["schema_version"] != 1:
raise ContractError("schema_version must be the integer 1")
categories = data["allowed_categories"]
if type(categories) is not list or not categories:
raise ContractError("allowed_categories must be a non-empty array")
if any(
type(item) is not str
or not item
or item != item.strip()
or has_surrogate(item)
for item in categories
):
raise ContractError("each category must be a trimmed Unicode string")
if len(set(categories)) != len(categories):
raise ContractError("allowed_categories contains duplicates")
limits: dict[str, int] = {}
for name in (
"max_record_id_chars",
"max_reason_chars",
"max_line_bytes",
):
value = data[name]
if type(value) is not int or value <= 0:
raise ContractError(f"{name} must be a positive integer")
limits[name] = value
return (
Policy(
allowed_categories=frozenset(categories),
max_record_id_chars=limits["max_record_id_chars"],
max_reason_chars=limits["max_reason_chars"],
max_line_bytes=limits["max_line_bytes"],
),
digest,
)
def limited_lines(
stream: BinaryIO,
limit: int,
digest: Any,
) -> Iterator[tuple[int, bytes | None]]:
line_number = 0
while True:
chunk = stream.readline(limit + 1)
if chunk == b"":
return
line_number += 1
digest.update(chunk)
too_long = len(chunk) > limit
if too_long and not chunk.endswith(b"\n"):
while True:
rest = stream.readline(limit + 1)
if rest == b"":
break
digest.update(rest)
if rest.endswith(b"\n"):
break
yield line_number, None if too_long else chunk
def validate_record(value: object, policy: Policy) -> dict[str, object]:
if type(value) is not dict:
raise RowError("NOT_OBJECT")
if set(value) != FIELDS:
raise RowError("FIELDS_MISMATCH")
record_id = value["record_id"]
if type(record_id) is not str:
raise RowError("RECORD_ID_TYPE")
if not record_id or record_id != record_id.strip():
raise RowError("RECORD_ID_FORMAT")
if len(record_id) > policy.max_record_id_chars:
raise RowError("RECORD_ID_TOO_LONG")
if has_surrogate(record_id):
raise RowError("RECORD_ID_UNICODE")
if any(ord(char) < 32 or ord(char) == 127 for char in record_id):
raise RowError("RECORD_ID_CONTROL_CHAR")
category = value["category"]
if type(category) is not str:
raise RowError("CATEGORY_TYPE")
if category not in policy.allowed_categories:
raise RowError("CATEGORY_NOT_ALLOWED")
needs_review = value["needs_review"]
if type(needs_review) is not bool:
raise RowError("NEEDS_REVIEW_TYPE")
reason = value["reason"]
if type(reason) is not str:
raise RowError("REASON_TYPE")
if not reason.strip():
raise RowError("REASON_EMPTY")
if len(reason) > policy.max_reason_chars:
raise RowError("REASON_TOO_LONG")
if "\x00" in reason or has_surrogate(reason):
raise RowError("REASON_UNICODE")
return {
"record_id": record_id,
"category": category,
"needs_review": needs_review,
"reason": reason,
}
def json_line(value: object) -> bytes:
text = json.dumps(value, ensure_ascii=False, separators=(",", ":"))
return (text + "\n").encode("utf-8")
def write_rejection(
stream: BinaryIO,
line_number: int,
code: str,
) -> None:
stream.write(json_line({"line": line_number, "code": code}))
def process(
input_path: Path,
output_dir: Path,
policy: Policy,
policy_sha256: str,
) -> dict[str, object]:
input_digest = hashlib.sha256()
seen_ids: set[str] = set()
total_lines = 0
valid_records = 0
rejected_records = 0
with (
input_path.open("rb") as source,
(output_dir / "valid.jsonl").open("xb") as valid_stream,
(output_dir / "errors.jsonl").open("xb") as error_stream,
):
for line_number, raw in limited_lines(
source,
policy.max_line_bytes,
input_digest,
):
total_lines += 1
error_code: str | None = None
record: dict[str, object] | None = None
if raw is None:
error_code = "LINE_TOO_LONG"
else:
try:
encoding = "utf-8-sig" if line_number == 1 else "utf-8"
text = raw.decode(encoding)
except UnicodeDecodeError:
error_code = "INVALID_UTF8"
else:
if not text.strip():
error_code = "BLANK_LINE"
else:
try:
value = json.loads(
text,
object_pairs_hook=strict_object,
parse_constant=reject_constant,
)
record = validate_record(value, policy)
except (ValueError, RecursionError):
error_code = "JSON_SYNTAX"
except RowError as exc:
error_code = exc.code
if error_code is None and record is not None:
record_id = str(record["record_id"])
if record_id in seen_ids:
error_code = "DUPLICATE_RECORD_ID"
else:
seen_ids.add(record_id)
valid_stream.write(json_line(record))
valid_records += 1
if error_code is not None:
write_rejection(error_stream, line_number, error_code)
rejected_records += 1
if total_lines == 0:
raise ContractError("input JSONL is empty")
summary: dict[str, object] = {
"schema_version": 1,
"status": "ok" if rejected_records == 0 else "rejected",
"input_file": input_path.name,
"input_sha256": input_digest.hexdigest(),
"policy_sha256": policy_sha256,
"total_lines": total_lines,
"valid_records": valid_records,
"rejected_records": rejected_records,
}
with (output_dir / "summary.json").open("xb") as stream:
stream.write(json_line(summary))
return summary
def emit(stream: Any, value: object) -> None:
text = json.dumps(value, ensure_ascii=False, separators=(",", ":"))
stream.write(text + "\n")
def main() -> int:
parser = argparse.ArgumentParser(
description="Validate provider-neutral AI result JSONL."
)
parser.add_argument("--input", required=True, type=Path)
parser.add_argument("--policy", required=True, type=Path)
parser.add_argument("--output-dir", required=True, type=Path)
args = parser.parse_args()
created_output = False
try:
if not args.input.is_file():
raise ContractError("input must be an existing file")
if not args.policy.is_file():
raise ContractError("policy must be an existing file")
if args.output_dir.exists():
raise ContractError("output directory already exists")
if not args.output_dir.parent.is_dir():
raise ContractError("output parent directory does not exist")
policy, policy_sha256 = load_policy(args.policy)
try:
args.output_dir.mkdir()
except FileExistsError as exc:
raise ContractError("output directory already exists") from exc
created_output = True
marker = args.output_dir / ".incomplete"
marker.write_bytes(b"incomplete\n")
summary = process(
args.input,
args.output_dir,
policy,
policy_sha256,
)
marker.unlink()
result = {
"status": summary["status"],
"total_lines": summary["total_lines"],
"valid_records": summary["valid_records"],
"rejected_records": summary["rejected_records"],
"output_dir": str(args.output_dir),
}
emit(sys.stdout, result)
return 0 if summary["status"] == "ok" else 1
except ContractError as exc:
if created_output:
shutil.rmtree(args.output_dir, ignore_errors=True)
emit(
sys.stderr,
{
"status": "error",
"code": "CONTRACT_ERROR",
"message": str(exc),
},
)
return 2
except KeyboardInterrupt:
if created_output:
shutil.rmtree(args.output_dir, ignore_errors=True)
emit(sys.stderr, {"status": "error", "code": "INTERRUPTED"})
return 130
except OSError as exc:
if created_output:
shutil.rmtree(args.output_dir, ignore_errors=True)
emit(
sys.stderr,
{
"status": "error",
"code": "IO_ERROR",
"message": str(exc),
},
)
return 3
except Exception:
if created_output:
shutil.rmtree(args.output_dir, ignore_errors=True)
emit(
sys.stderr,
{"status": "error", "code": "UNEXPECTED_ERROR"},
)
return 3
if __name__ == "__main__":
raise SystemExit(main())
処理順と失敗時の扱い
- 入力ファイル、ポリシー、未作成の出力先を確認する
- ポリシーJSONの重複キー、項目、型、値を確認する
- 出力ディレクトリを排他的に作り、
.incompleteを置く - 入力をバイト列で1行ずつ読み、行サイズ、UTF-8、JSON構文、項目、型、値、ID重複を確認する
- 合格行とエラー情報を分けて書き、最後に
summary.jsonを作る - 完了後だけ
.incompleteを削除し、判定に応じた終了コードを返す
行が上限を超えた場合は、その行全体をメモリへためず、改行まで読み捨てて次の行へ進みます。入力全体もメモリへ載せません。ただし、重複検査用のrecord_id集合は保持するため、ユニークID数に応じてメモリ使用量が増えます。
通常の例外とCtrl+Cでは、この実行が作った出力ディレクトリの削除を試みます。プロセスの強制終了や電源断では、途中のディレクトリが残る場合があります。同じ出力先へ複数のプロセスを同時実行しない運用も必要です。
4．実行して終了コードを確認する
次はbash系シェルでの実行例です。
python3 validate_ai_jsonl.py \
--input ai_results.jsonl \
--policy result_policy.json \
--output-dir checked
echo $?標準出力と終了コードは次のようになります。
{"status":"rejected","total_lines":3,"valid_records":1,"rejected_records":2,"output_dir":"checked"}
1終了コード1は、検査処理は完了したものの、少なくとも1行を拒否したことを表します。後続のCSV更新や登録処理は止めます。
| 終了コード | 意味 |
|---|---|
| 0 | 全行が契約を通過 |
| 1 | 1行以上を拒否。検査結果は出力済み |
| 2 | 引数、入力、ポリシー、出力先の契約違反 |
| 3 | ファイル操作または予期しない実行エラー |
| 130 | Ctrl+Cによる中断 |
argparseが検出する必須引数の不足や未知の引数も終了コード2です。この場合はスクリプト独自のJSONではなく、使用法が標準エラー出力へ表示されます。
5．合格行とエラーを確認する
checked/valid.jsonlには1行目だけが入ります。
{"record_id":"R-001","category":"配送","needs_review":false,"reason":"到着日の確認依頼"}checked/errors.jsonlには、元の理由文やカテゴリ値を複製せず、位置とコードだけを記録します。
{"line":2,"code":"CATEGORY_NOT_ALLOWED"}
{"line":3,"code":"NEEDS_REVIEW_TYPE"}summary.jsonには入力とポリシーのSHA-256も入ります。ハッシュは同じ内容かを照合するための値であり、暗号化や匿名化ではありません。
主なエラーコードは次のとおりです。
JSON_SYNTAX：JSONとして解析できないDUPLICATE_KEY：同じオブジェクト内に同名キーがあるNON_STANDARD_NUMBER：NaNやInfinityがあるFIELDS_MISMATCH：必須項目の不足または未知項目の追加CATEGORY_NOT_ALLOWED：許可されていない分類NEEDS_REVIEW_TYPE：真偽値の型が違うDUPLICATE_RECORD_ID：検査を通過した行のIDが重複LINE_TOO_LONG、INVALID_UTF8：行サイズまたは文字コードの違反
6．全行が合格する経路を確認する
次をai_results_ok.jsonlとして保存します。
{"record_id":"R-001","category":"配送","needs_review":false,"reason":"到着日の確認依頼"}
{"record_id":"R-002","category":"請求","needs_review":true,"reason":"請求内容を担当者が確認する必要あり"}
既存ディレクトリを上書きしないため、別の出力先を指定します。
python3 validate_ai_jsonl.py \
--input ai_results_ok.jsonl \
--policy result_policy.json \
--output-dir checked-ok
echo $?この入力では次の結果になります。
{"status":"ok","total_lines":2,"valid_records":2,"rejected_records":0,"output_dir":"checked-ok"}
0自動処理へ組み込む場合は、終了コードが0で、.incompleteがなく、summary.jsonのstatusがokであることを確認してから次へ進めます。終了コード1で作られたvalid.jsonlは調査用の部分結果です。完全な結果として登録してはいけません。
このゲートが判定しないこと
- 分類や理由が業務上正しいか
reasonに事実誤認、機密情報、不適切な文言がないか- 依頼したIDがすべて返ったか、依頼していないIDが追加されたか
needs_reviewの判断基準が妥当か- AIサービス固有の利用規約、保存条件、データ送信可否
このスクリプトが確認するのは、構文と決めたデータ契約です。意味の正しさは保証しません。反映前には依頼時と結果のID一覧を突合し、欠落と余分なIDを検出します。そのうえで、needs_reviewが真の行や業務上重要な分類を人が確認します。
導入前の確認
正常な1行、未知カテゴリの1行、型違いの1行を用意し、終了コード0と1の両方を確認します。確認できたら、「AI結果を受け取る処理」と「CSVや業務システムへ反映する処理」の間に、このゲートを置きます。