3キュー設計
request / reply / cancel

箱は、1つでは足りなかった

前回、リクエストを1つの箱（キュー）に並べ、 prefetch_count=1 で1件ずつ直列に処理する仕組みを作りました。 殺到したリクエストを並べてピークをならす——このパターンは、非力なVPSでよく効きます。

ところが AIコンシェルジュを実際に組むと、1つの箱では回らない場面が出てきました。 やり取りの中に、性質のまるで違う3つの「流れ」が混ざっていたからです。 これらを1つの箱に押し込むと、特に「途中で止めたい」がまともに動きません。

この記事は、その3つの流れをなぜ request / reply / cancel の3つのキューに分けたのか、 という設計判断の記録です。

性質が、まるで違う

コンシェルジュのやり取りには、方向も順序も緊急度も違う3つの流れがあります。

request（依頼）

ユーザーの質問。ユーザー → サーバーの向き。並んで順番待ちして構わない、素直な FIFO です。前回の「箱」がまさにこれ。

reply（応答）

LLMが生成したトークンを返す流れ。サーバー → ユーザーの逆向き。しかも一度に全部ではなく、生成しながら少しずつ流れます。

cancel（中止）

「もう要らない、止めて」という割り込み。待たせたら意味がないのが決定的な違いです。順番待ちの列に並んだ時点で、キャンセルはキャンセルの役目を果たせません。

この表の cancel の行だけ、明らかに毛色が違います。ここが3キューに分けた理由の核心です。

なぜ cancel を同じ箱に入れられないか

いちばん素朴なのは「キャンセル依頼も、質問と同じ request キューに流す」案です。実装は一見ラク。 ですがこれはFIFOの罠に落ちます。

キューは先入れ先出し。キャンセル依頼を投げても、それは列の最後尾に並びます。 前に10件の質問が待っていたら、その10件を処理し終えるまでキャンセルは Consumer に届きません。 届いた頃には——止めたかった処理は、とっくに終わっている。CPUもメモリも、丸ごと無駄に使い切ったあとです。

言葉で書くと当たり前に見えますが、動かして比べると差は歴然です。下のデモで、同じ操作を2つの設計で試してみてください。

cancel は、どの箱に入れるべきか

質問を送ると、Consumer が長い回答を生成し始めます（トークンが1つずつ流れます）。 生成の途中で「キャンセル」を押したとき、2つの設計で何が起きるかを見比べます。

モードAでは、キャンセルが request の最後尾に並び、前の待ち依頼を全部処理し終えるまで届きません。 その頃には止めたかった生成は完了済み——徒労です。 モードBでは、独立キューから即座に拾われ、走行中の生成がその場で止まります。

「止める」を、どう伝えるか

cancel を独立キューにしても、話はもう一段あります。キューはメッセージを届けるだけ。 走行中のLLM推論を実際に止めるのは、別の仕組みが要ります。

共有セット cancelled_sessions

採った方法はシンプルです。キャンセルされたセッションIDを1つの集合に入れておき、 推論ループがトークンを生成するたびに、その集合を確認して自分から降りる。

# cancel キューを監視する側：届いたら集合に足すだけ
async def on_cancel(session_id):
    cancelled_sessions.add(session_id)

# 推論ループ側：1トークンごとに自分で確認して降りる
for token in llm.stream(prompt):
    if session_id in cancelled_sessions:
        break                 # ここで即座に停止
    await send_token(token)

なぜ集合（set）か。 理由は2つ。複数のConsumerプロセス間で「キャンセルされたか」を共有でき、 かつ「入っているか」の判定が高速だからです。推論ループは毎トークンこれを引くので、ここが遅いと本末転倒になります。

放置と残骸に、どう備えるか

キューを本番で回すと、「誰も取りに来ないメッセージ」「再起動で残った古いメッセージ」への備えが要ります。

メッセージTTL — 放置は自動消滅

投げたきり誰も応答を待っていない依頼を、いつまでも抱えるのは無駄です。メッセージに寿命（TTL）を持たせ、 一定時間で自動的に消えるようにします。aio_pika では、この寿命は文字列ではなく timedelta で渡します （ここは実装で一度ハマった点です）。

起動時パージ — 残骸を一掃してから受付

サーバーを再起動したとき、前回の残りメッセージが箱に残っていると、いきなり古い依頼を処理し始めてしまいます。 そこで起動時にキューを空にしてから受付を開始します。パージは main() から 1回だけ呼ぶのが肝で、接続取得のたびに呼ぶと毎回中身を消してしまいます。

reply のタイムアウトは別扱い

通常の応答待ちと、「全記事の再スクレイプ」のような重い処理とでは、待つべき時間が桁違いです。 同じタイムアウトで縛ると、前者は無駄に長く、後者は途中で切れます。そこで通常用と 再スクレイプ用（長め）を別々の定数に分けました。