サブエージェントの受け渡しを壊さないスキーマ進化の設計

境界で弾くバリデーション

契約は、下流エージェントが仕事を始める前に評価します。受け取った直後に検証し、不適合なら下流を一切起動しない。これが「サイレント破壊」を「明示的な失敗」に変える唯一の場所です。

import jsonschema
 
def accept_handoff(payload: dict, schema: dict) -> dict:
    """下流が成果物を受け取る境界。不適合なら即座に止める。"""
    try:
        jsonschema.validate(instance=payload, schema=schema)
    except jsonschema.ValidationError as e:
        # どのフィールドで、何を期待し、何が来たのかを必ず残す
        path = "/".join(str(p) for p in e.absolute_path) or "(root)"
        raise HandoffError(
            f"ハンドオフ契約違反 at '{path}': {e.message}"
        ) from e
    return payload
 
 
class HandoffError(Exception):
    """上流の成果物が下流の入力契約を満たさないときに投げる。"""

ここで大切なのは、エラーメッセージに「どのフィールドが」「なぜ」落ちたかを残すことです。

サブエージェントの失敗ログは後から人が読みます。validation failed とだけ書かれたログを深夜に眺めるのは、想像以上に消耗します。path を添えるだけで、原因特定が数分から数十秒へ、体感で10倍ほど速くなります。

スキーマを進化させる現実的な手順

運用が始まると、必ず「findings に信頼度スコアを足したい」のような要望が出ます。ここで additionalProperties: False の壁にぶつかります。

私が採っているのは、schema_version を上げて新旧を併存させる方式です。破壊的変更を一気に当てず、移行期間を設けます。

1. 新フィールドを 任意（required に入れない） で追加した V2 スキーマを定義する
2. 下流のバリデータを「V1 も V2 も受け付ける」状態にする（anyOf で両対応）
3. 上流を V2 出力に切り替える
4. 全フローが V2 に移ったことをログで確認してから、V1 受付を外す

RESEARCH_HANDOFF_V2 = {
    **RESEARCH_HANDOFF_V1,
    "properties": {
        **RESEARCH_HANDOFF_V1["properties"],
        "schema_version": {"const": 2},
        "findings": {
            "type": "array", "minItems": 1,
            "items": {
                "type": "object",
                "required": ["claim", "source_url"],   # confidence は required にしない
                "properties": {
                    "claim": {"type": "string", "minLength": 10},
                    "source_url": {"type": "string", "format": "uri"},
                    "confidence": {"type": "number", "minimum": 0, "maximum": 1},
                },
                "additionalProperties": False,
            },
        },
    },
}
 
ACCEPTED = {"anyOf": [RESEARCH_HANDOFF_V1, RESEARCH_HANDOFF_V2]}

この順番を守ると、上流と下流のデプロイタイミングがずれても受け渡しは壊れません。新フィールドをいきなり required にすると、切り替えの一瞬で既存フローが全滅します。追加は任意から、削除は最後に。この原則だけは譲らないことをお勧めします。

本番で実際に起きた症状

実運用で遭遇したのは、次のような形でした。

• 下流が空の成果物を返すのに、どのエージェントも例外を投げない
• 再実行すると直るときと直らないときがある（上流のデプロイ進行と噛み合っていた）
• App Store 向けのリリースノート生成と AdMob 収益レポートの要約で、説明文だけがまるごと欠落する

最後のものは特に怖い症状でした。Google Play と App Store の両方へ配信する文面を生成する流れで、下流が「タイトルは来たが本文キーが無い」状態を素通りさせていたのです。境界バリデーションを入れる前は、配信直前のレビューで人間が気づくしかありませんでした。

契約検証を境界に置いてからは、この種の欠落はパイプラインの最初の段で止まり、対処はその場の再生成だけで済むようになりました。失敗の場所が「配信直前」から「受け渡し直後」に前倒しされる。これが設計で買える最大の安心だと感じています。

再発を防ぐ3つの仕掛け

最後に、同じ事故を繰り返さないために組み込んでいる仕掛けを3つ挙げます。

第一に、契約スキーマを上流・下流が 同じファイルから import すること。プロンプト内に形を二重で書くと、必ず片方だけが更新されてずれます。

第二に、schema_version を成果物そのものに埋めること。後からログを掘るとき、その成果物がどの契約で作られたのかが一目で分かります。

第三に、バリデーション失敗を「リトライ可能エラー」と「契約違反エラー」に分けること。前者はネットワーク等の一時障害で、再試行に意味があります。後者は形そのものの不一致で、再試行しても無駄なので即座に人へ上げます。

個人開発で複数のエージェントを束ねていると、賢いオーケストレーションよりも、地味な境界検証のほうが夜の安眠に効きます。型付きの受け渡しは、エージェントを増やすほど価値が複利で効いてくる投資だと考えています。

同じ「沈黙する下流」に悩んでいる方が、境界に一枚の検証を置くきっかけになれば嬉しく思います。