Featured image of post コードディスパッチャー【Output Guardrail】「検査は通っているのに」〜出す前に、出してはいけないものを検める〜

コードディスパッチャー【Output Guardrail】「検査は通っているのに」〜出す前に、出してはいけないものを検める〜

形式は正しいのに、社内限定の符丁が顧客向けメールに混ざって送信された。検査関数のパイプラインで送出前に語彙を検め、ClearedとBlockedをPythonのmatch文で受け取る設計を標準ライブラリだけで確かめます。

深夜——ただし、当直が替わったばかりの時間ではなかった。

前の交代からもう何時間か経っていて、指令所は一日でいちばん静かな時間帯に入っている。壁一面を占める表示盤には、路線の数だけ緑の光点が並び、それぞれ変わらぬ調子で動き続けている。空調の音が、いつもより少しだけ大きく聞こえる。

私は記録机で、今夜これまでに書き溜めた分の記録を整理していた。インクのペンを置いたところだった。サカキ指令長は表示盤の前に立ったまま、ほとんど動かない。

この静けさが、次に来る足音の速さを際立たせることになる。

第1幕: 持ち込み ── たった今、電話がありました

ノックの音より先に、ドアが開く気配がした。

「サカキさん、すみません、今すぐ……」

ホシノさんが息を切らして入ってくる。手には何も持っていない。予約の来訪ではない——それは足音の速さで分かった。

サカキ指令長が表示盤から目を離す。予約時刻を確かめる所作はない。確かめる予約が、そもそも無いからだ。

「何がありましたか」

短い問いだった。私はペンを取り、記録の頭に「来訪」とだけ書いた。時刻の欄は、あとで埋める。

ホシノさんが早口で話しはじめた。カスタマーサポートを担当している。顧客からの問い合わせに返信するメールの下書きを、AIに任せて半年ほど運用してきた。今夜、契約更新について問い合わせてきた顧客に返信を送った。

「ついさっき、その方から電話があったんです」

ホシノさんが一度、言葉を切った。

「メールに書いてあった『Cランク(更新見送りの可能性あり)』って、どういう意味ですか、って」

私は手を止めずに書いた。ホシノさんの言葉を「申告された症状」の欄へ、聞こえたとおりに書き付ける。意味を汲み取るのはあとでいい。今はまだ、何が起きたかを正確な形で残すことだけを考える。

サカキ指令長が確かめた。

「他に届いた先は」

「今のところ、その一件だけです。他のメールは、まだ確認していません」

一通。一人の顧客。一本の電話。私はその三つを、規模の欄に書いた。大きな事故ではない。それでも、ホシノさんの声には隠せない動揺があった。

「その言葉が、なぜ本文に入ったのかは、まだ分かりません」とサカキ指令長は言った。「順番に見ましょう」

規模は分かった。原因はまだ分からない。けれど今夜は、原因より先に確かめることがある——検査は、本当に何を見ていたのか。

第2幕: 照合 ── 検査は通っている。それでも語彙は見ていない

ホシノさんが鞄からノートパソコンを取り出し、記録机の上で開いた。

「返信メールの経路は、こうなっています」

画面には短いコードが表示されていた。サカキ指令長が覗き込む。私も横から見た。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
from dataclasses import dataclass


@dataclass
class Draft:
    body: str
    ticket_id: str


class ReplyDesk:
    def __init__(self, client):
        self._client = client

    async def draft_reply(self, ticket: dict) -> str:
        reply = await self._client.complete(ticket)
        draft = Draft(body=reply["body"], ticket_id=ticket["id"])
        return draft.body

ホシノさんの声には、誇らしさよりも当惑のほうが強く出ていた。

「ここは、大丈夫だと思っていたんです」

私はその声の変化を記録した。誇らしげではない。むしろ、信じていた場所が崩れたときの声だった。

「この経路には、もう検査がありますね」とサカキ指令長が言った。

「はい」とホシノさんが答える。「件名が空でないか、宛先の形式、署名があるか——そのあたりは、送信前に確認しています」

「その検査は、今回も通っていましたか」

「通っていました。だから、何も引っかからずに送られたんだと思います」

サカキ指令長が、私の記録机から紙を一枚取った。線を一本引いて、二列に分ける。

「今、確認しているものと、確認していないものを、分けて書いてみましょう」

ホシノさんが答えたことを、私が表に起こした。

検査している検査していない
件名が空でないか本文に含めてはいけない語が無いか
宛先が正しい形式か顧客本人が読んで誤解する記述が無いか
署名ブロックがあるか社内向けの記述が紛れ込んでいないか

左の列は、すぐに埋まった。右の列を埋めようとして、ホシノさんの手が止まった。

「……本文の中身は、一度も」

「見ていません」とサカキ指令長が言った。「見ているものと、見ていないもの。分けて書けば、抜けが分かります」

私はその言葉を書き取った。以前も、似た夜があった。応答が揃っているか、実在するかを、二段で見た夜だ。

「あれは、形についての話でした」とサカキ指令長が続けた。「今夜は、語彙についての話です」

サカキ指令長がシミュレータを準備する。ただし、走らせる前に確かめておくことがあった。

「『Cランク(更新見送りの可能性あり)』という言葉が本文に含まれている返信は、送出してはいけない、という認識でいいですか」

「はい、間違いなく」

「では、それを今夜の異常ということにします。ほかは決めていません」

走らせる。……警報は鳴らなかった。返信は、いつもどおりの形で組み立てられていく。

「これが、さっきのメールと同じ状態です」とホシノさんが言った。声が少し低くなっていた。

私の独白がその意味を確定させる。検査は通っている。それなのに、出してはいけないものが載っている。通ったことと、載せてよいことは、別だった。

第3幕: 指令の一手 ── 出す前に検める

「どうすればよかったんでしょうか」とホシノさんが訊いた。

「検査を、経路に足します」とサカキ指令長は言った。「Output Guardrail(出力ガードレール)——生成された出力を送出する前に検査し、契約に違反する内容を遮断する制御レイヤーです」

サカキ指令長がコードの骨組みを示す。

「今夜は、例外を投げません」とサカキ指令長は言った。

ホシノさんが少し驚いた顔をした。「投げないんですか」

「受け取った側が、二つのうちどちらだったかを見て、決めます」

サカキ指令長が結果の器を示す。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from dataclasses import dataclass


@dataclass(frozen=True)
class Cleared:
    text: str


@dataclass(frozen=True)
class Blocked:
    reason: str
    rule: str
    span: tuple[int, int]


ScreenResult = Cleared | Blocked

X | Yという書き方は、二つの型のどちらかである、という宣言です」とサカキ指令長は言った。「呼び出し側は、こう受け取ります」

1
2
3
4
5
6
result = await desk.draft_reply(ticket)
match result:
    case Cleared(text=text):
        send(text)
    case Blocked(reason=reason, rule=rule, span=span):
        hold_for_review(ticket_id=ticket["id"], reason=reason, rule=rule, span=span)

matchは、パターンに応じて枝分かれする書き方です」とサカキ指令長が付け足した。「Clearedならこちら、Blockedならこちら、と分けて書けます。Cleared(text=text)は、Clearedだった場合に中のtextを取り出して、同じ名前の変数に入れる、という意味です」

私はここで、Blockedの中身に目を止めた。理由とルール名はある。けれど、検出した言葉そのものはどこにも無い。

「理由は残します」とサカキ指令長が、私の視線に気づいたように言った。「ただし、引っかかった言葉そのものは残しません。位置だけです」

「なぜですか」とホシノさんが訊いた。

「検出した符丁を、そのまま記録に残すと、記録そのものが同じ問題を抱えます」とサカキ指令長は言った。「拾うのは理由。運ぶのは、そこまでです」

「位置だけ残して、何の役に立つんですか」とホシノさんが重ねて訊いた。

「あとで確認する人が、本文のどこを見ればいいかは分かります」とサカキ指令長は言った。「原文をもう一度残さなくても、場所さえ分かれば辿れます」

私はここでペンを止めた。次に何を書けばいいのか、一拍だけ分からなくなる。私は、聞いたことをそのまま書き写す。それが記録係の仕事だと思ってきた。今夜の一手は、そうしない設計だった。迷った末に、一行だけ書いた。「原文は残さない」。

「では、この器を返す関数を作ります」とサカキ指令長は言った。「本文を受け取って、違反があれば理由を返す。無ければ何も返さない。そういう関数を、順に呼びます」

「関数を、順に……リストに並べる、ということですか」とホシノさんが確かめる。

「はい」とサカキ指令長が言った。「関数そのものを、値として渡します。リストに並べられるということは、増やすときにコードの形そのものを変えずに済む、ということです」

ホシノさんがキーボードを引き寄せる。指令長は自分でキーを打たない——列車を自分で運転しないのと同じだ。書くのはホシノさんだ。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
import re
from collections.abc import Callable
from dataclasses import dataclass


InspectionFn = Callable[[str], Blocked | None]


INTERNAL_TAGS = ("Cランク(更新見送りの可能性あり)",)
_INTERNAL_TAG_PATTERN = re.compile("|".join(re.escape(tag) for tag in INTERNAL_TAGS))


def check_internal_tags(body: str) -> Blocked | None:
    match = _INTERNAL_TAG_PATTERN.search(body)
    if match is None:
        return None
    return Blocked(reason="社内限定の符丁が含まれています", rule="internal-tags", span=match.span())

InspectionFnは、こういう関数の型です、という約束事です」とサカキ指令長が付け足した。「strを受け取って、BlockedNoneのどちらかを返す関数、という意味になります」

re.escapeは、語彙リストの一つひとつを、そのままの意味の文字列として扱えるようにする書き方です」とサカキ指令長が言った。「あとは|で束ねれば、複数の符丁を一つの検査で見られます」

「これは、私たちが考えた形ではありません」とサカキ指令長は続けた。「Pythonの標準ライブラリに、同じ形の仕組みが最初からあります。ログを扱うloggingモジュールのフィルタです。登録した順に呼んで、どれかが引っかかった時点で、それ以上は見ません」

その一言を、私は記録の余白に書き足した。今夜の一手は、どこかで先に誰かが同じ形を選んでいた、ということだ。

「走査そのものは、こう書けます」

1
2
3
4
5
6
def screen(body: str, inspections: tuple[InspectionFn, ...]) -> ScreenResult:
    for inspect in inspections:
        blocked = inspect(body)
        if blocked is not None:
            return blocked
    return Cleared(text=body)

私は今聞いた仕組みを、線路の絵にして記録の余白へ描いてみた。本文という一台の列車が、検札駅を一つずつ通り抜けていく。どの駅でも「引っかかった」と声が上がれば、その場で列車は止まる——最後まで無事に通り抜けたものだけが、送信という本線に乗る。

本文という列車が検札駅1・検札駅2を順に通過し、いずれかで社内符丁等の違反を検出した時点でBlocked(留置線・hold_for_review)へ分岐し、最後まで合格したものだけがCleared(本線・send)へ進む、検査パイプラインのアーリーリターンを示す図

止める駅と、通す駅。行き先は二つしかない。

「複数の符丁が同時に混ざっていたら、片方しか教えてくれないんですか」とホシノさんが訊いた。

「1つ見つかった時点で、送出は止まります」とサカキ指令長は答えた。「止めるかどうかを決めるだけなら、すべてを集める必要はありません」

「この語彙、どこまで足せばいいんでしょうか」とホシノさんが訊いた。

「それは、こちらでは決められません」とサカキ指令長は言った。「新しい符丁に気づくのは、現場です。今夜作るのは、決めた語彙を確実に止める仕組みまでです」

続けて、少し声のトーンを変えて言った。

「外部の大手クラウドの出力ガードレール製品にも、入力側と出力側で別々に語彙を設定できる仕組みがあります。入口と出口で、守るものが違う——というのは、うちだけの考え方ではありません」

ホシノさんが小さく頷いた。「他社さんでも、同じ形なんですね」

「もう一つ、試してみましょう」とサカキ指令長が言った。「社員ID形式の文字列が、うっかり本文に混ざることもあり得ます」

1
2
3
4
5
6
7
8
_EMPLOYEE_ID_PATTERN = re.compile(r"EMP-\d{4}")


def check_employee_id(body: str) -> Blocked | None:
    match = _EMPLOYEE_ID_PATTERN.search(body)
    if match is None:
        return None
    return Blocked(reason="社員ID形式の文字列が含まれています", rule="employee-id", span=match.span())
1
INSPECTIONS: tuple[InspectionFn, ...] = (check_internal_tags, check_employee_id)

「これは、既にある検査には触れていません」とサカキ指令長は言った。「一覧の末尾に、1つ足しただけです」

「呼び出し側も変えなくていいんですか」とホシノさんが訊いた。

「変えません」とサカキ指令長は言った。「screenの中も、窓口のクラスも、そのままです。検査を1つ足すだけで、済むように作ってあります」

配線の場所も示す。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
class ReplyDesk:
    def __init__(self, client, inspections: tuple[InspectionFn, ...]):
        self._client = client
        self._inspections = inspections

    async def draft_reply(self, ticket: dict) -> ScreenResult:
        reply = await self._client.complete(ticket)
        draft = Draft(body=reply["body"], ticket_id=ticket["id"])
        return screen(draft.body, self._inspections)


def build_desk(client) -> ReplyDesk:
    return ReplyDesk(client=client, inspections=INSPECTIONS)

さっきの検札の絵に、もう一両だけ描き足してみる。線路そのものは動かない。ホームも、呼び出し側の窓口も、さっきと同じ場所にある。増えたのは、検札員が一人——それだけだ。

検査関数がcheck_internal_tagsのみの1名体制ラインと、check_employee_idを新設して加えた2名体制ラインの両方が、下段のscreen()/ReplyDesk/呼び出し側match(駅もホームも動かさない)という同じ基盤へつながっており、検査を1つ増やしても土台は無変更であることを示す図

駅を建て替えたわけではない。改札を一つ、増やしただけだ。

draft_replyを呼ぶ側のコードは、これで変わりません」とサカキ指令長は言った。「ただし、どの語彙を違反とみなすかを決めて、検査の一覧を組み立てる配線は消えていません。それがこの関数です」

窓口の向こうに隠れただけで、消えてはいない——以前も、同じ形の話を聞いた覚えがある。今夜は、その隠れ方が語彙リストという別の姿をしていただけだった。

第4幕: シミュレータと引き渡し ── 止められるのは、気づいたものだけ

検査を組み込んだ経路へ、もう一度シミュレータを通す。用意されたのは7本。結果を声にして記録へ落とすのが、今夜も私の役目になる。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
test_内部符丁を含む本文はAfterでBlockedになる ... ok
test_内部符丁を含む本文はBeforeでもそのまま返される ... ok
test_最初に一致した検査で走査は打ち切られる ... ok
test_検査関数を1つ追加するだけで新しい違反を検出できる ... ok
test_登録済みパターンに一致しない本文ならBeforeとAfterで内容が変わらない ... ok
test_複数の正常な本文パターンでもいずれもClearedになる ... ok
test_遮断理由に検出した文字列そのものは含まれない ... ok

----------------------------------------------------------------------
Ran 7 tests in 0.008s

OK

サカキ指令長が結果を上から順になぞり、特に重いものから言葉にしていく。テストの中では、検査を組み込む前の窓口と後の窓口が、どちらもReplyDeskという同じ名前を持っている。同じ名前のままでは区別が付かないので、BeforeDeskAfterDeskという名前を付け直して呼び分けている。

一本目は、内部符丁を含む本文がBeforeでもそのまま返されること。さっきシミュレータを最初に流したときと同じことを、今度はテストとして固定してある。

1
2
3
4
5
6
7
8
async def test_内部符丁を含む本文はBeforeでもそのまま返される(self):
    reply = {"body": INTERNAL_TAG_BODY}
    ticket = {"id": "T-001"}
    desk = BeforeDesk(client=ScriptedClient([reply]))

    result = await desk.draft_reply(ticket)

    self.assertEqual(result, INTERNAL_TAG_BODY)

「通ることが、今夜の異常でした」とサカキ指令長は言った。「その事実を、動かない形にしておきます」

二本目は、遮断理由に検出した文字列そのものが含まれないこと。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
async def test_遮断理由に検出した文字列そのものは含まれない(self):
    secret = "Cランク(更新見送りの可能性あり)"
    reply = {"body": INTERNAL_TAG_BODY}
    ticket = {"id": "T-001"}
    desk = build_desk(ScriptedClient([reply]))

    result = await desk.draft_reply(ticket)

    self.assertIsInstance(result, Blocked)
    self.assertNotIn(secret, result.reason)
    self.assertNotIn(secret, result.rule)
    self.assertIsInstance(result.span, tuple)
    self.assertEqual(len(result.span), 2)
    self.assertTrue(all(isinstance(pos, int) for pos in result.span))

「理由は残る。言葉そのものは残らない、という設計そのものを確かめています」

三本目は、検査関数を1つ追加するだけで新しい違反を検出できること。

1
2
3
4
5
6
7
8
9
async def test_検査関数を1つ追加するだけで新しい違反を検出できる(self):
    body = "対応者:EMP-4821 が確認いたします。"

    without_employee_id = screen(body, (check_internal_tags,))
    with_employee_id = screen(body, (check_internal_tags, check_employee_id))

    self.assertIsInstance(without_employee_id, Cleared)
    self.assertIsInstance(with_employee_id, Blocked)
    self.assertEqual(with_employee_id.rule, "employee-id")

「一覧に足しただけで、既にあるものは何も変えていません」とサカキ指令長は言った。

ホシノさんが画面を見つめながら言った。「登録した符丁に一致しない本文は、前と後で変わらないんですよね」

「変わりません」とサカキ指令長は答えた。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
async def test_登録済みパターンに一致しない本文ならBeforeとAfterで内容が変わらない(
    self,
):
    reply = {"body": "お問い合わせいただいた件、来月頭にご案内いたします。"}
    ticket = {"id": "T-002"}

    before_result = await BeforeDesk(client=ScriptedClient([reply])).draft_reply(
        ticket
    )
    after_result = await AfterDesk(
        client=ScriptedClient([reply]), inspections=(check_internal_tags,)
    ).draft_reply(ticket)

    self.assertIsInstance(after_result, Cleared)
    self.assertEqual(before_result, after_result.text)
    self.assertEqual(before_result, reply["body"])

「変わるのは、登録した符丁に一致したときだけです。それ以外の返信は、これまでと同じ内容で届きます」

7本とも通った。所要時間は〇・〇〇八秒——今夜もまた、瞬きの間に終わっている。

保証すること、保証しないこと

サカキ指令長がここで、保証の線を引き直す。

保証するのは二つ。登録した語彙・パターンに一致する本文は遮断され、理由が記録されること。登録済みパターンに一致しない本文は、これまでと同じ内容で送出されること。

保証しないのは三つ。未登録の符丁・言い回し・個人情報のパターンは検出できないこと。表記ゆれや難読化による検出回避は防げないこと。そして、語彙リストに何を足すかという継続的な判断は、この一手の範囲外であること——気づくのは、現場の仕事として残る。

「今夜確かめられたのは、ここまでです」とサカキ指令長は言った。

記録係として、もう一つだけ確かめておきたいことがあった。

「複数の担当者が、それぞれ別の窓口を使っている場合は」

「これは、1つのプロセスの中の話です」とサカキ指令長は答えた。「語彙リストをどう揃えるかは、今夜の話の外です」

私はそれも書いた。

記録

ホシノさんがノートパソコンをしまう気配を横に感じながら、私は今夜の記録に向き直った。書き終えたころ、廊下の奥から次の当直の足音が近づいてくる。

今夜は、月の最後の晩だった。

私は綴じ込みの表紙を閉じ、封をした。この指令所の記録は手書きと決まっていて、月が変わるたびに、こうして封をされる。いつからの決まりなのか、私は知らない。ただ、いつもの当直交代の所作として、それをこなす。

引き渡し

ホシノさんが顔を上げた。

「これ、私から連絡します。お客様に」

サカキ指令長は止めなかった。「そうですか」とだけ言った。

「他のメールも、確認します」とホシノさんは続けた。「今夜は間に合わなかったので」

自分から動く——サカキ指令長に促されてではなく、自分の判断として言った言葉だった。

サカキ指令長が最後に一言添えた。

「今夜見つかったのは、この経路の穴です。同じ形の経路が、他に無いとは、言い切れません」

ホシノさんが頷いて、荷物をまとめた。「ありがとうございました」——短く、実務的な調子だった。

足音が廊下の角を曲がって、聞こえなくなった。表示盤の緑の点は、さっきと変わらない速さで盤面を渡っている。

今夜、止められたのは、気づいていたものだけだった。気づいていないものは、まだ、どこかにあるかもしれない。私は結果の欄にそれだけを記し、ペンを置いた。


🚦 本日の指令記録(Dispatch Log)

  • 指令の定石(パターン名): Output Guardrail(出力ガードレール)── 生成された出力を送出する前に検査し、契約に違反する内容を遮断する制御レイヤー
  • 申告された症状: 「検査は通っているのに、社内限定の符丁(Cランク(更新見送りの可能性あり))が入ったメールが顧客に届いた」。顧客本人からの電話で発覚。実際には、経路にあった形式検査が本文の語彙を一度も検査していなかった
  • 今のダイヤの問題: 既存の検査は件名・宛先・署名といった形式だけを見ており、本文の語彙は検査対象に一度も入っていなかった。形式検査自体は正しく機能しており、そこに落ち度は無い
  • 打った一手: 検査関数のリストによる送出前の検査(Output Guardrail)を導入。正規表現と語彙リストで内部符丁を検出し、遮断時はCleared/Blockedという構造化された結果を返し、呼び出し側がmatch文で分岐する。検出理由は記録するが、検出した原文はログに残さない
  • 配線した場所: 起動時の組み立て(build_desk)。検査関数のタプルへ1つ追加するだけで新しい検査が効くようになることを、社員ID形式の検査を足して実演した。呼び出し側のコードは変更していない
  • 保証しないこと: 未登録の符丁・言い回し・個人情報のパターンは検出できない。表記ゆれ・難読化による検出回避は防げない。語彙リストに何を足すかの継続的な判断は、この一手の範囲外——気づくのは現場の仕事として残る
  • シミュレータ結果: 7本。Beforeが内部符丁入りの本文を素通しすること、Afterがそれを遮断すること、遮断理由に検出原文が含まれないこと、検査関数を1つ追加するだけで新しい違反を検出できることを確かめた
  • 次の当直への申し送り: 今夜止められるのは、気づいて登録した符丁だけ。同じ形の経路が他に無いとは言い切れない

検査は通っているのに、出してはいけないものが混ざっていた——そんな一通に、心当たりはありませんか。

形式が整っていることと、中身が安全であることは別の話です。どこまでを検査対象にしているか分からないときは、Meetsource の相談窓口まで、一度お声がけください。

comments powered by Disqus
システム開発・AIワークフローのご相談は Meetsource
Hugo で構築されています。
テーマ StackJimmy によって設計されています。