Featured image of post コードディスパッチャー【Model Cascade】「それ自体は、想定の範囲内です」〜全部を特急に乗せる必要はない〜

コードディスパッチャー【Model Cascade】「それ自体は、想定の範囲内です」〜全部を特急に乗せる必要はない〜

難易度に関わらず全件を最上位モデルへ通し費用が処理量に比例して膨らむ問題を、確信度による前捌きとエスカレーション(Model Cascade)で解決し、標準ライブラリだけで実装し確かめます。

第1幕: 持ち込み ── まだ、何も起きていないんです

その夜、指令所には特に変わったところがなかった。空調の低い音が、いつもと同じ調子で流れていた。サカキは盤の前で、いつもと変わらない姿勢だった。私はいつもの位置で記録帳を開き、ペン先を整える。棚に並んだ綴じ込みの背表紙が、蛍光灯の下でわずかに光っていた。今夜は何も予兆がない夜だと、私はそう思っていた。

ドアが静かに開いた。押し開ける手に、迷いも急ぎもない。

「こんばんは。サカキさん、少しお時間よろしいですか」

コダマさんが入ってくる。声は落ち着いていた。疲れている様子も、焦っている様子もない。手には数枚の紙を持っていて、それを丁寧に胸元で持ち直している。歩き方にも、急いでいる気配はなかった。むしろ、事前に段取りを決めてきたような、静かな足取りだった。

「座ってください」とサカキ。

コダマさんが腰を下ろし、持ってきた紙を机の上に丁寧に広げた。表計算ソフトで作った、細かい数字の並んだ試算表だった。角がきちんと揃えられていて、何度も見直した跡がある。私は「来訪」の欄を書き始める。予約どおりの時刻。予約でも、飛び込みでもない、ただの静かな来訪だった。これまでの来訪には、たいてい何かしらの気配があった。引け目、焦り、疲労、期限。今夜は、そのどれとも違う。静けさそのものが、今夜の質だった。

「先週の会議で、来月から取引先が増えて、処理する文書の量が三倍になることが決まりました」

コダマさんの声は平静だった。「うちは、届いた契約書や問い合わせ文書を、内容ごとに自動で仕分けるシステムを使っています。それが、今の三倍の量を捌くことになります」

「費用も、単純計算では三倍になります」。コダマさんが試算表の数字を指でなぞる。「それ自体は、想定の範囲内です」

私は独白した。困っているようには見えない。むしろ、既に答えを出したうえで、その答えを確かめに来ているように見えた。

「それで……何かご不安が」とサカキ。

コダマさんが一拍、間を置いた。声の張りは変わらない。ただ、その間だけが、今夜の唯一の違和感だった。

「三倍という数字を、上の者にそのまま報告していいのか、確信が持てなくて。今の月額の、単純に三倍——差額だけで、数十万円になります。処理時間も三倍になったら、今の受付から数時間以内という締切も、守れるのか」

私の独白:三倍という数字は、既に出ている。困っているのは、その数字の重さではなく、その数字が本当に正しいのかどうか、だった。

「まず、その試算がどこから来ているか、一緒に見ましょう」とサカキ。

私の独白:コダマさんは、何かが壊れて来たのではなかった。壊れる前に、数字の重さを確かめに来ていた。これまで指令所に持ち込まれてきたのは、いつも既に起きたことだった。今夜、初めて、まだ起きていないことが持ち込まれた。

第2幕: 照合 ── 迷うものと、迷わないもの

コダマさんが鞄からノートパソコンを取り出し、画面をサカキの方へ向けた。キーボードに触れる手つきに、迷いはなかった。半年前から毎日触ってきた道具、という手つきだった。

「今のシステムは、届いた文書を、全部同じモデルに通しています」

サカキがコードを覗き込む。

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


class Classifier(Protocol):
    async def classify(self, document: str) -> tuple[str, float]: ...


@dataclass(frozen=True)
class TopTierOnly:
    heavy: Classifier

    async def classify(self, document: str) -> tuple[str, float]:
        return await self.heavy.classify(document)

「難易度で分けている様子はありませんね」とサカキ。

コダマさんが頷く。「はい。実際、これまでは、それで困ったことはありませんでした」

「全件を同じモデルに通す、という判断自体は、量が少ないうちは合理的です」。サカキが続ける。「精度を最優先にできますから」

コダマさんが少し表情を緩めた。「はい。半年前に立ち上げたときは、量もそれほど多くなかったので、迷わず一番良いモデルに任せていました」。当時は、月に数百件程度だったという。精度を確かめながら育てていく段階では、費用のことより、間違えないことの方がずっと重要だった。「その頃の判断が、今も残っている、ということですね」とサカキ。「量が少ないうちは、それで何の問題もありませんでした」

「実際の文書を、何件か見せてもらえますか」とサカキ。コダマさんが直近の分類対象からいくつか持ってくる。私が1件ずつ、書き写していく。

1
2
3
4
文書A「見積書のフォーマットに関する問い合わせ」        → 一読して即断できる(型どおり)
文書B「休暇申請の承認フローについて」                  → 一読して即断できる(型どおり)
文書C「業務委託契約 第14条の解除条件の解釈について」   → 読んでも判断に迷う(例外条項が絡む)
文書D「新規取引先との秘密保持契約の適用範囲」          → 読んでも判断に迷う(前例と微妙に違う)

私の独白:A・Bは、私が読んでも一目で分かる。見積書の書式についてと聞けば、答えはすぐに出る。けれど、C・Dは何度読み返しても迷う。解除条件の解釈も、秘密保持契約の適用範囲も、条文の細部まで読み込まないと判断できない。同じ「文書」という括りの中に、まるで違う重さのものが混ざっていた。これが、今夜唯一、私の手が止まった瞬間だった。

サカキが試算を並べ直す。「今の月間処理件数がN件、上位モデルの単価がP円だとすると、今の費用はN×P円です。来月は三倍でも、単価が変わらなければ3N×P円になります」

ここまでは、コダマさんの試算と一致する。

「この中で、A・Bのような文書は、どれくらいの割合ですか」とサカキ。

コダマさんが少し考える。「感覚的には、八割くらいは、そう複雑ではないと思います」

「では、その八割にまで、常に一番重い判定をかけている、ということです」

コダマさんが試算表の数字を見つめ直す。声には出さないが、何かに気づいたような間があった。

「シミュレータを一度流します」とサカキ。流す前に、確認する。「難易度に関わらず、全部が同じ重さで処理されている——それを、今夜の異常ということにして、いいですか」

「はい」とコダマさん。

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

サカキが手元の台本を実行に移す。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
async def test_Beforeは易しい文書も難しい文書も同じ上位モデルへ送る(self):
    heavy = ScriptedClassifier(
        [
            ("型どおり", 0.98),
            ("型どおり", 0.95),
            ("例外条項", 0.60),
            ("例外条項", 0.55),
        ]
    )
    responder = TopTierOnly(heavy=heavy)
    documents = [
        "見積書のフォーマットに関する問い合わせ",
        "休暇申請の承認フローについて",
        "業務委託契約 第14条の解除条件の解釈について",
        "新規取引先との秘密保持契約の適用範囲",
    ]

    for document in documents:
        await responder.classify(document)

    self.assertEqual(heavy.call_count, 4)

易しい文書も、難しい文書も、heavyが4回とも呼ばれる。表示盤の隅で、通過を示す印が一つ、また一つと灯っていく。四つとも、同じ重さの処理として並んだ。

私の独白:壊れてはいなかった。ただ、区別していなかっただけだった。

第3幕: 指令の一手 ── 各駅停車と特急の乗り継ぎ

コダマさんが試算表から顔を上げる。「防ぐ手立ては、あるものなんでしょうか」。

「まず安いモデルに文書を見せて、自信を持って答えられそうならそれで確定させる。自信が持てないものだけを、上のモデルへ回す」。サカキが続ける。「Model Cascade——段階を追って処理する仕組みです」

サカキが比喩で言い換える。「全部を特急に乗せる必要はありません」

「締切の方は、どうなりますか」とコダマさん。

「迷わない文書は、前捌きだけで終わります。乗り継ぎが要らない分、平均すれば早くなるはずです」。サカキが少し間を置く。「ただ、今夜はっきり測ったのは費用の側だけです。処理時間がどれだけ縮むかは、また別に確かめる必要があります」

コダマさんが頷いた。「まずは、費用の話として持ち帰ります」

コダマさんが尋ねる。「前にどこかで、“振り分け"の仕組みについて聞いたことがあるんですが、それと同じですか」

「似ていますが、違います」。サカキが少し考えてから答える。「一度の判定で行き先を決めて終わる仕組みもあれば、まず安いところで試して、自信が持てなければ、もう一段上へ送る——という繰り返しの仕組みもあります。今回は後者です」

私の独白:判定を1回で終える構造か、確信が持てるまで段を上げていく構造か。違うのは、そこだった。

サカキがコードを示す。

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


class Classifier(Protocol):
    async def classify(self, document: str) -> tuple[str, float]: ...


@dataclass(frozen=True)
class ModelCascade:
    light: Classifier
    heavy: Classifier
    threshold: float

    async def classify(self, document: str) -> tuple[str, float]:
        label, confidence = await self.light.classify(document)
        if confidence >= self.threshold:
            return label, confidence
        return await self.heavy.classify(document)

「前捌きも、本処理も、同じ形の口を持っています」。サカキがコードのClassifierを指す。「渡す文書も、渡す指示も、まったく同じです。変わるのは、答えるモデルの力量だけ」

「わざわざ同じにしているのは、何か意味があるんですか」とコダマさん。

「はい。同じものを見せて、力量だけを比べる——そうでなければ、どちらが賢いから正しいのか、単純に比べられません」

私はClassifierのところに目を留めた。class Classifier(Protocol)——継承しなくても、同じ形の口さえ持っていれば、同じものとして扱える。実際、シミュレータで使う偽の分類器は、Classifierを継承していない。それでもclassifyという同じ形のメソッドさえ持っていれば、ModelCascadeはそれを前捌きにも本処理にも使える。前捌きのモデルも、本処理のモデルも、この口さえ実装していれば、ModelCascadeから見れば同じ種類のものだった。型ヒントが実行時に強制されるわけではないことは、以前にも聞いた。それでも、同じ形の口を約束事として決めておくことに意味がある——そういうことなのだろうと、私は自分なりに納得した。

サカキがコードの分岐を指す。「confidencethreshold以上なら、そこで確定します。未満なら、次の段へ回します」

私の独白:サカキの言葉を、頭の中で線路の図に描き直してみる。文書は、まず前捌き駅に停まる。信号が青なら、そこで旅は終わる。合図が変われば、同じ文書を乗せたまま、本線の列車が本処理駅へ向かう。前捌きの申告が、その合図の色を決めていた。

ModelCascadeの判定フロー図。文書は前捌き駅(light.classify)へ渡り、確信度が閾値以上なら確定へ、未満なら本処理駅(heavy.classify)へ乗り継ぐ。本処理駅へは前捌きの結果ではなく同じ文書がそのまま独立して渡ることを示す運行指令所風インフォグラフィック

「その閾値の見極めは、どう決めるんですか」とコダマさん。

サカキが黒板代わりの紙に、2行を書き出す。

1
2
前捌きだけに任せたときの損失  = 誤って確定させた件数 × 誤判定1件あたりの実害
上位へ回すことで増える費用    = エスカレーションした件数 × 上位単価

「前捌きだけに任せたときの損失と、上位へ回すことで増える費用。今夜は、この2つを比べます」

「上位へ回した分は、前捌きの費用が無駄になるわけじゃないんですか」とコダマさん。

「前捌きの費用は、確定するにせよ回すにせよ、どのみち払っています」。サカキが続ける。「回したときに純粋に増えるのは、上位の費用の分だけです」

「文書ごとの難しさは、見ないんですか」とコダマさん。

「本当は、文書1件ごとの難しさで損失は変わります。でも今夜は、まず全体の平均でこの2つを比べるところから始めます」

私の独白:粗いと分かったうえで、まずここから始める。サカキはそう言っていた。

コダマさんが問う。「その『自信がある』というのは、何を根拠にしているんですか」

「前捌きのモデル自身が申告してきた数字です」。サカキが続ける。「その申告が、いつも正確とは限りません。申告そのものを正しく直す方法もありますが、今夜はそこまでは踏み込みません。まずは、決めた閾値と照らし合わせるところから始めます」

コダマさんが小さく頷いた。「まずは粗くても、始めるということですね」

「今夜の仕組みは、一つのバッチ処理の中だけの話です」とサカキ。「文書が来るたびに、この一つのModelCascadeが順番に判定しています」

「複数のバッチが同時に動いていたら、どうなりますか」とコダマさん。

「そこは、また別の設計が要ります。今夜は、そこまでは扱いません」

サカキが、組み立て側のコードを指し示す。

1
2
def build_cascade(light: Classifier, heavy: Classifier, *, threshold: float) -> ModelCascade:
    return ModelCascade(light=light, heavy=heavy, threshold=threshold)

「文書を分類する側の呼び出しコードは、変わりません」。サカキが続ける。「閾値をいくつにするかは、組み立てるときの一箇所だけを差し替えれば足ります」

「その数字は、私が決めるんですか」とコダマさん。

「はい。許容できる誤判定の割合と、上位に回したときの追加費用を見比べて、決めてください」。サカキが向きだけを示す。「損失の方が重ければ閾値を下げて確定を増やし、追加費用の方が気になるなら閾値を上げて回す数を絞る——向きは、それだけです」

私はその一言を、欄の隅にそのまま書き留めた。数字を出すところまではサカキがやる。数字を決めるのは、コダマさんの領分だった。

第4幕: シミュレータと引き渡し ── 迷わなかった分だけ、乗り換える

同じシミュレータを、今度は8本流す。結果を私が順に声に出す。特に2本を噛み砕く。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
async def test_確信度が閾値以上なら前捌きだけで確定し上位モデルは呼ばれない(self):
    light = ScriptedClassifier([("型どおり", 0.95)])
    heavy = ScriptedClassifier([])
    cascade = build_cascade(light, heavy, threshold=0.8)

    await cascade.classify("見積書のフォーマットに関する問い合わせ")

    self.assertEqual(light.call_count, 1)
    self.assertEqual(heavy.call_count, 0)

async def test_確信度が閾値未満なら上位モデルへエスカレーションする(self):
    light = ScriptedClassifier([("例外条項", 0.60)])
    heavy = ScriptedClassifier([("解除条件・要確認", 0.90)])
    cascade = build_cascade(light, heavy, threshold=0.8)

    await cascade.classify("業務委託契約 第14条の解除条件の解釈について")

    self.assertEqual(light.call_count, 1)
    self.assertEqual(heavy.call_count, 1)

確信度が閾値以上の文書は、前捌きだけで確定し、上位モデルが呼ばれない。閾値未満の文書は、これまでどおり上位モデルへ回る。もう1本、迷った文書の結果がBeforeと一致することを確かめる。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
async def test_エスカレーションされた文書はBeforeと同じ判定結果になる(self):
    document = "業務委託契約 第14条の解除条件の解釈について"

    before_heavy = ScriptedClassifier([("解除条件・要確認", 0.90)])
    before = TopTierOnly(heavy=before_heavy)
    before_result = await before.classify(document)

    light = ScriptedClassifier([("例外条項", 0.60)])
    after_heavy = ScriptedClassifier([("解除条件・要確認", 0.90)])
    cascade = build_cascade(light, after_heavy, threshold=0.8)
    after_result = await cascade.classify(document)

    # 低確信度側は必ず heavy が判定するため、Before/After の結果は構造的に一致する。
    self.assertEqual(after_result, before_result)

「迷った文書は、最終的に同じheavyが答えます」とサカキ。「だから、そこの結果は変わりません」

もう1本は、名前こそ似ているが、中身は逆側だった。「これは、証明ではありません。台本を揃えただけの確認です」。サカキがそう前置きしてから示す。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
async def test_前捌きで確定した文書はBeforeと同じ判定結果になる(self):
    # 台本上、light の判定結果を heavy と同じ値にそろえているだけであり、
    # 実運用でこの一致を保証するものではない(同値範囲は非対称)。
    document = "見積書のフォーマットに関する問い合わせ"

    before_heavy = ScriptedClassifier([("型どおり", 0.98)])
    before = TopTierOnly(heavy=before_heavy)
    before_result = await before.classify(document)

    light = ScriptedClassifier([("型どおり", 0.98)])
    heavy = ScriptedClassifier([])
    cascade = build_cascade(light, heavy, threshold=0.8)
    after_result = await cascade.classify(document)

    self.assertEqual(after_result, before_result)
    self.assertEqual(heavy.call_count, 0)

「このテストは、前捌きの答えを、あらかじめ上位モデルと同じ値にそろえて用意しています」。サカキがコードのコメントを指す。「台本の中だけの一致です。実際に運用したとき、安い方が同じ答えを出すとは限りません」

コダマさんの声が少し引き締まる。「つまり、費用は減らせても、精度は無条件では保証されない、ということですね」

「そのとおりです」

私の独白:迷った文書の答えは、変わらないと言い切れる。迷わなかった文書の答えは、変わらないとは言い切れない。同じ「一致」という言葉の中に、確かさの違うものが二つ並んでいた。

もう2本、確かめる。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
async def test_前捌きと本処理は同じ文書を受け取る(self):
    light = ScriptedClassifier([("例外条項", 0.60)])
    heavy = ScriptedClassifier([("解除条件・要確認", 0.90)])
    cascade = build_cascade(light, heavy, threshold=0.8)
    document = "業務委託契約 第14条の解除条件の解釈について"

    await cascade.classify(document)

    self.assertEqual(light.received_documents, [document])
    self.assertEqual(heavy.received_documents, [document])

async def test_8割が前捌きで確定する分布では上位モデルの呼び出しは全体の2割に留まる(self):
    light_script = [("型どおり", 0.95)] * 8 + [("例外条項", 0.60)] * 2
    light = ScriptedClassifier(light_script)
    heavy = ScriptedClassifier([("解除条件・要確認", 0.90)] * 2)
    cascade = build_cascade(light, heavy, threshold=0.8)
    documents = [f"文書{i}" for i in range(10)]

    for document in documents:
        await cascade.classify(document)

    self.assertEqual(heavy.call_count, 2)

前捌きと本処理が、同じ文書をそのまま受け取っていること。八割が前捌きで確定する分布では、上位モデルの呼び出しが全体の二割に留まること。コダマさんが試算表の数字を見比べる。「私が見積もった八割という数字が、そのまま費用の削減にもつながる、ということですね」

コダマさんの言葉に、私は先ほど声に出して読み上げた数字を思い出す。十件のうち、乗り継ぎ合図が出たのはたった二件。残りの八件は、前捌き駅から動かないまま、そこで旅を終えていた。

文書10件のバッチが前捌き駅を通過し、確信度が閾値以上の8件はそのまま確定、閾値未満の2件だけが本処理駅へ乗り継ぐ比率を示す運行指令所風インフォグラフィック

サカキが二つに分けて示す。

保証すること。迷った文書は、これまでどおり上位モデルが答え、結果が変わらないこと。同じ文書・同じ指示を、両方のモデルへ渡していること。

保証しないこと。迷わなかった文書について、前捌きの判定が常に上位モデルと同じであること。確信度の申告そのものの正確性。文書1件ごとの精密な損失計算——今夜は、全体平均の二項比較に留めた。

「今夜の仕組みが効くのは、この一つのバッチ処理の中だけです」とサカキ。「複数のバッチが同時に動くようになったら、閾値の統一や集計は、また別の設計が要ります。今夜見たのは、そこまでです」

私はそれも、そのまま書いた。

コダマさんが荷物をまとめ始める気配を横目に、私は記録の最後の欄へ向かっていた。ペン先が紙をこする音だけが、静かな指令所に小さく響く。今夜の「来訪」の欄を書き終えたところで、目が隣の欄に留まった。様式の中に、うちの指令所では一度も使ったことのない欄がある。まだ起きていないことのための欄なのだろうか——そこまで考えて、私は首を振った。この欄、何のためにあるのだろう。理由は分からない。サカキに尋ねることもしなかった。私はただ、空欄のままそこを飛ばして、次の行を書いた。

コダマさんが、持ち込んだ試算表をもう一度手に取った。「三倍という数字、書き直します」。ペンを取り、元の数字の横に新しい数字を書き足す。「全部が三倍になるのではなく、迷う分だけが、上位に回る。その割合で、費用を出し直します」

サカキが短く応じる。「その資料の方が、上の方には伝わると思います」

コダマさんが試算表を鞄へしまい、軽く会釈した。「……ありがとうございました。これで、報告できます」。落ち着いた調子のまま、静かな納得へと変わっていた。

足音が遠ざかる。表示盤の点は、来たときと同じ調子で光っていた。何も壊れていない夜が、そのまま終わろうとしていた。

私の独白で、この夜を閉じる。今夜渡せたのは、まだ起きていないことへの、数字の書き直し方だけだった。壊れてから直すことにも、壊れる前に書き直すことにも、同じ指令所の仕事がある。壊れる前に、書き直せることもあるのだと、知った。


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

  • 指令の定石(パターン名): Model Cascade(多段モデル)── 安価な前捌きモデルで先に判定し、確信度が閾値未満のものだけを上位モデルへ回す仕組み
  • 申告された症状: 「来月から取引先拡大で処理量が三倍になることが正式決定した。自分で試算した数字に確信が持てず相談に来た」。当事者はまだ困ってはおらず、確定した将来の変化への予防的な相談だった
  • 今の運行計画の問題: 文書の難易度に関わらず、全件を同じ上位モデルへ通していた。量が少ないうちは合理的だったが、処理量が増えるほど費用と処理時間がそのまま比例して膨らむ
  • 打った一手: 前捌きモデルと本処理モデルに同じClassifierという口(Protocol)を実装させ、確信度が閾値以上なら前捌きだけで確定、閾値未満なら上位モデルへ回すModelCascadeを導入
  • 配線した場所: 起動時の組み立て(build_cascade)。閾値をいくつにするかは、組み立てるときの一箇所だけを差し替えれば足りる。呼び出す側のコードは変更していない
  • 保証しないこと: 前捌きが確信を持って答えた判定が、常に上位モデルと同じであることは保証しない。確信度の申告そのものの正確性(較正)も、文書1件ごとの精密な損失計算も、今夜は扱わない
  • シミュレータ結果: 8本。迷った文書はBeforeと同じ結果になること、迷わなかった文書は前捌きだけで確定すること、同じ文書・同じ指示を両方のモデルへ渡していることを確かめた
  • 次の当直への申し送り: 今夜決めたのは、二段階の振り分けまで。文書1件ごとの精密な損失計算と、確信度そのものの較正は、まだ手つかず

処理量の増加に比例して費用が膨らむシステムをお持ちなら、前捌きと確信度によるエスカレーションという設計を、これまでにも承っています。

相談窓口はこちら

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