Pythonで業務用語の定義漏れをCIで止める

昨日、LLM に渡す売上指標の用語集を JSON にしていて、純売上 の計算に必要な キャンセル額 がどこにも定義されていないことに気付いた。プロンプトなら「不足している用語は質問して」と書ける。でも実行のたびに守らせるより、マージ前に落としたほうが早い。

業務用語集は説明資料ではなく、エージェントが参照するデータ契約として扱うのがいい。必須項目の欠落、存在しない依存先、指標どうしの循環参照を CI で止める小さな linter を書いた。Python 3.10 以降、外部パッケージなしで動く。

8月3日に公開された AWS Japan の Context Ontology Accelerator の記事も、データだけでは業務の意味が足りない点を扱っている。自分はまず、立派なグラフ基盤より前に「その用語を誰が定義し、何を元に計算するか」を機械的に検査できる状態を作りたい。

用語集を JSON で固定する

例として、受注明細から作る売上系の指標を置く。id はプログラム用、label と definition は人と LLM が読むための情報だ。source と owner がない用語は、数字が合わなくなったときに戻る先がなくて困る。

owner は部署名だけでも最初は十分だった。定義を変える PR に名前のない指標が混ざると、レビューで正しさを決める人もいなくなる。ここは技術の問題に見えて、実際は変更の受け皿を置く作業でもある。

[
  {
    "id": "gross_sales",
    "label": "売上総額",
    "definition": "キャンセル前の受注明細合計",
    "source": "orders",
    "owner": "販売企画"
  },
  {
    "id": "cancelled_amount",
    "label": "キャンセル額",
    "definition": "キャンセル済み受注明細の合計",
    "source": "orders",
    "owner": "販売企画"
  },
  {
    "id": "net_sales",
    "label": "純売上",
    "definition": "売上総額からキャンセル額を引いた金額",
    "source": "orders",
    "owner": "販売企画",
    "depends_on": ["gross_sales", "cancelled_amount"]
  }
]

計算式そのものを文字列で解析するのは、SQL 方言や別名で急に難しくなる。ここでは依存先を depends_on に分けた。集計クエリやセマンティック層は別途レビューし、この JSON は「どの言葉がどの言葉に頼るか」の台帳に絞る。役割を欲張らないほうが CI で壊れにくい。

ここで用語集をマージ前に検査する。

Python の用語集 linter

次のスクリプトは、必須フィールド、ID の重複と形式、未定義の依存先、循環参照を調べる。循環参照の検出は深さ優先探索で、探索中のノードをもう一度踏んだら経路を出している。

#!/usr/bin/env python3
"""JSON の業務用語集を CI 向けに検査する。Python 3.10+。"""
from __future__ import annotations

import argparse
import json
from pathlib import Path
import re
import sys

REQUIRED = ("id", "label", "definition", "source", "owner")
IDENTIFIER = re.compile(r"^[a-z][a-z0-9_]*$")

def lint(path: Path) -> list[str]:
    try:
        terms = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        return [f"{path}: cannot read JSON: {exc}"]

    if not isinstance(terms, list):
        return ["top level must be a JSON array"]

    errors: list[str] = []
    by_id: dict[str, dict] = {}
    for index, term in enumerate(terms, 1):
        prefix = f"terms[{index}]"
        if not isinstance(term, dict):
            errors.append(f"{prefix}: must be an object")
            continue
        for key in REQUIRED:
            if not isinstance(term.get(key), str) or not term[key].strip():
                errors.append(f"{prefix}: {key} is required")
        term_id = term.get("id")
        if isinstance(term_id, str):
            if not IDENTIFIER.fullmatch(term_id):
                errors.append(f"{prefix}: invalid id {term_id!r}")
            elif term_id in by_id:
                errors.append(f"{prefix}: duplicate id {term_id!r}")
            else:
                by_id[term_id] = term
        depends_on = term.get("depends_on", [])
        if not isinstance(depends_on, list) or not all(
            isinstance(item, str) for item in depends_on
        ):
            errors.append(f"{prefix}: depends_on must be an array of strings")

    for term_id, term in by_id.items():
        for dependency in term.get("depends_on", []):
            if dependency not in by_id:
                errors.append(f"{term_id}: unknown dependency {dependency!r}")

    visiting: set[str] = set()
    visited: set[str] = set()

    def visit(term_id: str, trail: list[str]) -> None:
        if term_id in visiting:
            start = trail.index(term_id)
            errors.append("cycle: " + " -> ".join(trail[start:]))
            return
        if term_id in visited:
            return
        visiting.add(term_id)
        for dependency in by_id[term_id].get("depends_on", []):
            if dependency in by_id:
                visit(dependency, trail + [dependency])
        visiting.remove(term_id)
        visited.add(term_id)

    for term_id in by_id:
        visit(term_id, [term_id])
    return errors

def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("terms", type=Path)
    args = parser.parse_args()
    errors = lint(args.terms)
    if errors:
        print(*errors, sep="\n", file=sys.stderr)
        raise SystemExit(1)
    count = len(json.loads(args.terms.read_text(encoding="utf-8")))
    print(f"OK: {count} terms, 0 errors")

if __name__ == "__main__":
    main()

手元では上の 3 件の JSON に対して、次の出力を確認した。

$ python glossary_lint.py glossary.json
OK: 3 terms, 0 errors

adjustment を定義せず、gross_sales と net_sales を相互参照させた fixture では終了コード 1 になった。

net_sales: unknown dependency 'adjustment'
cycle: gross_sales -> net_sales -> gross_sales

CI では、たとえば GitHub Actions のテスト手順にこれを一行足せばいい。

- run: python glossary_lint.py data/business_terms.json

ここで止められること、止められないこと

この linter は「純売上の SQL が本当に正しいか」までは証明しない。キャンセル日の基準、消費税を含めるか、返品をいつ差し引くかは definition をレビューして、実データに対するテストも別に持つ必要がある。

それでも、参照先が消えたまま用語だけ残る事故や、指標 A が B を参照し B が A を参照する事故は、モデルの回答を眺めていても見つけにくい。用語集の構造なら差分の時点で決まる。ここを静かに弾けるだけで、回答がもっともらしいまま数字の由来を見失う回数はかなり減る。

おわりに

業務用語を LLM に渡すなら、説明文を増やす前に壊れない参照関係を作っておきたい。definition、source、owner、depends_on の4つが揃うと、回答の根拠をたどる入口になる。

自分なら最初は売上や在庫のように揉めやすい指標だけを 10 件ほど JSON 化して、この検査を CI に入れる。用語集が増えてからでは、欠けた一語がどのプロンプトに効いたのか追いにくい。先に止める場所を作っておくと、エージェント側の評価もやりやすくなる。

参考: https://zenn.dev/aws_japan/articles/context-ontology-accelerator-deploy