はじめに

アプリやエージェントから LLM の API を叩いていると、だいたい次の2つで困ります。

同じ質問が何度も飛んで、そのたびに課金される
プロバイダが落ちたとき、リクエストがタイムアウトまで溜まり続けてアプリ全体が詰まる

これを解決するために、アプリと LLM (Anthropic、OpenAI、llama.cpp、vLLM) の間に置く Rust 製のオーケストレーター tokio-prompt-orchestrator を作りました。サーバー (orchestrator コマンド) としても、Rust のライブラリとしても使えます。

この記事では、使い方よりも「中で何をしているか」を中心に書きます。特に、意味が同じ質問を使い回す「セマンティック重複排除」で埋め込みベクトルだけに頼ると危ない話は、LLM のキャッシュを自作している人の参考になると思います。

リポジトリ: https://gitlab.com/mattbusel/tokio-prompt-orchestrator
日本語 README: https://gitlab.com/mattbusel/tokio-prompt-orchestrator/-/blob/main/README.ja.md

既存の OpenAI クライアントのまま使う

いちばん手軽なのは、OpenAI 互換のプロキシとして使う方法です。orchestrator を起動して、クライアントのベース URL を向けるだけで、重複排除、サーキットブレーカー、タイムアウト、利用額の上限、デッドレターキューが入ります。コードの変更はありません。

export OPENAI_BASE_URL=http://127.0.0.1:8080/v1

orchestrator --provider echo (プロンプトをそのまま返すモードで、API キー不要) を起動した状態で、公式の Python クライアントを何も変えずに実行します。

from openai import OpenAI

client = OpenAI()  # reads OPENAI_BASE_URL and OPENAI_API_KEY

reply = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Summarize this ticket"}],
)
print(reply.choices[0].message.content, reply.usage.total_tokens)

for chunk in client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Now stream it"}],
    stream=True,
):
    if chunk.choices:
        print(chunk.choices[0].delta.content or "", end="")
print()

Summarize this ticket 10
Now stream it

同じ質問を curl からもう一度送ると、重複排除のウィンドウから返ってきてモデルは呼ばれません。レスポンスヘッダに x-orchestrator-dedup: cached が付きます。

curl -s -i localhost:8080/v1/chat/completions -H 'Content-Type: application/json' \
  -d '{"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "Summarize this ticket"}]}'

x-orchestrator-dedup: cached
{"choices":[{"finish_reason":"stop","index":0,"logprobs":null,"message":{"content":"Summarize this ticket","refusal":null,"role":"assistant"}}],"created":1790804186,"id":"chatcmpl-4d5284a811bd4c0f90bee01018103e1a","model":"echo","object":"chat.completion","system_fingerprint":null,"usage":{"completion_tokens":5,"prompt_tokens":5,"total_tokens":10}}

エラーも OpenAI の形式で返ります。ブレーカーが開いている間は 503、利用額の上限に達したら 429 です。OpenAI のモデルなら usage のトークン数 (と利用額の上限) はそのモデル自身のトークナイザで数えるので、請求額と一致します。それ以外のモデルは「約4文字で1トークン」の推定です。

Anthropic の SDK も同じで、ベース URL を向ければ POST /v1/messages に同じ仕組みが入ります。

import anthropic

client = anthropic.Anthropic(base_url="http://127.0.0.1:8080", api_key="local")
msg = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=200,
    messages=[{"role": "user", "content": "Summarize this ticket"}],
)
print(msg.content[0].text, msg.stop_reason)

with client.messages.stream(model="claude-sonnet-4-6", max_tokens=200,
                            messages=[{"role": "user", "content": "Now stream it"}]) as s:
    print("".join(s.text_stream))

Summarize this ticket end_turn
Now stream it

テキストの会話、システムプロンプト、ストリーミングに対応しています。画像とツールのブロックは invalid_request_error で明示的に拒否します。

中の仕組み: 5段のパイプラインと有界チャネル

リクエストは input_tx.send() か POST /api/v1/infer から入り、Retrieve、Assemble、Inference、Post-process、Stream の5段を通ります。各段は独立した Tokio タスクで、段と段の間は容量 512、512、512、1024、512、256 の有界チャネルでつないでいます。

ポイントは「チャネルが満杯でもメモリを増やさない」ことです。満杯のときはリクエストを理由付きでデッドレターキュー (1000件) に落とし、HTTP API は 429 と Retry-After を返します。キューが無限に伸びてプロセスごと落ちる、という LLM 連携でよくある壊れ方を構造的に起こさないようにしています。

推論段 (3段目) では、すべてのリクエストが次の順に通ります。

締め切り (deadline) のチェック
サーキットブレーカー: 5回失敗すると開き、60秒間は呼び出しを拒否し、その後1件だけ試しに通す
120秒のタイムアウト
あなたの ModelWorker

重複排除 (Deduplicator)、リトライ (RetryPolicy)、レート制限 (RateLimiter) は、ワーカーを包むオプションのラッパーです。落とされたリクエストには backpressure、deadline_expired、circuit_breaker_open、inference_timeout、inference_failure のような理由が付きます。

図の内容はすべて src/stages.rs から読み取ったものです。

実際に障害を起こしてみる

cargo run --example llm_pipeline (キー不要) を実行すると、4ユーザー x 3問の12リクエストを流したあと、プロバイダ障害をシミュレートします。実際の出力を一部省略して載せます。

1) 12 requests: 4 users x 3 questions
   req-01 alice  The capital of France is Paris.
   req-02 alice  Backpressure means a slow consumer makes fast producers wait instead of letting queues grow without bound.
   ...
   req-12 dave   Bounded channels fill / the sender waits its turn now / memory stays calm
   -> 12 answers in 0.9s, 3 model calls (9 saved by dedup)

2) Provider outage: 8 new requests while every call fails
   DLQ req-13  inference_failure:inference failed: 503 Service Unavailable (simulated outage)
   ...
   DLQ req-17  inference_failure:inference failed: 503 Service Unavailable (simulated outage)
   DLQ req-18  circuit open, failed fast (provider not called)
   DLQ req-19  circuit open, failed fast (provider not called)
   DLQ req-20  circuit open, failed fast (provider not called)
   -> breaker is Open; it lets one probe through after 60s to test recovery

12リクエストに対してモデル呼び出しは3回 (9回は重複排除で節約)。障害中は5回失敗したところでブレーカーが開き、残りの3件はプロバイダを呼ばずに即座に失敗しています。8件とも理由付きでデッドレターキューに入るので、あとから再送できます。

PROVIDER=anthropic ANTHROPIC_API_KEY=... か PROVIDER=openai OPENAI_API_KEY=... を付ければ、同じ3回の呼び出しを本物のモデルに対して行います。

セマンティック重複排除: 埋め込みだけでは危ない

完全一致の重複排除は、まったく同じプロンプトしか拾えません。そこで埋め込みモデルを使い、言い換えた質問もキャッシュから返せるようにしました (x-orchestrator-dedup: semantic、類似度は x-orchestrator-similarity ヘッダ)。orchestrator --semantic-dedup はローカルモデル (fastembed の BGE-small、API キー不要、初回に 128 MB をキャッシュディレクトリへダウンロード) を使います。

ところが、埋め込みだけで判定するのは危険でした。BGE-small で測ると、「Convert 10 miles to kilometers」と「Convert 10 kilometers to miles」の類似度は 0.99 で、試したどの本物の言い換えよりも高かったのです。これをキャッシュヒットにすると、逆向きの換算の答えを返してしまいます。

そこで、類似度に加えて「数字が同じであること」「共通する単語の順序が同じであること」を一致の条件にしました。テストのペア (tests/semantic_dedup_tests.rs) での結果です。デフォルトの閾値は 0.93 です。

ペア	類似度	答えを再利用したか
"What is the capital of France?" / "Which city is the capital of France?"	0.959	する
"How many ounces are in a pound?" / "How many oz in one lb?"	0.931	する
"How do I reverse a list in Python?" / "What's the way to reverse a Python list?"	0.985	しない (語順のため。1回分の呼び出しが余計にかかる)
"Convert 10 miles to kilometers" / "Convert 10 kilometers to miles"	0.992	しない
"Is 17 a prime number?" / "Is 21 a prime number?"	0.845	しない

3行目のように、本当は同じ意味なのに再利用しないケースもあります。設計として「迷ったらもう1回モデルを呼ぶ。他人の答えを返すほうには倒さない」にしています。閾値を下げる前に、自分のトラフィックでヒットの中身を確認してください。

自分のドキュメントから答える

orchestrator --docs ./my-docs でフォルダ内の Markdown とテキストをインデックスし (tantivy の BM25、メモリ上)、質問の前に関連する箇所をファイル名付きで差し込みます。--provider echo だとモデルが受け取る内容がそのまま見えます。

  Answering from 2 passages in ./demo-docs
> How long do refunds take?
Use the following excerpts to answer. If they do not contain the answer, say so. [1] (policies/refunds.md) # Refunds Refunds are issued to the original card within 14 days of us receiving the return. Gift cards cannot be refunded. Question: How long do refunds take?

検索が失敗したり2秒を超えたりした場合は、コンテキストなしでプロンプトを送ります。検索のせいでリクエストが落ちることはありません。

なお組み込みの BM25 は英語のステミング前提です。日本語のドキュメントで使う場合は、Retriever トレイトを実装して、普段お使いの検索 (ベクトル DB、Elasticsearch、Postgres など) につなぐこともできます。

Rust から使う

cargo run --example quickstart のコードです。

use std::{collections::HashMap, sync::Arc};
use tokio_prompt_orchestrator::{spawn_pipeline, EchoWorker, ModelWorker, PromptRequest, SessionId};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Swap EchoWorker for AnthropicWorker, OpenAiWorker, LlamaCppWorker, VllmWorker,
    // or a worker over the client you already use (next section).
    let worker: Arc<dyn ModelWorker> = Arc::new(EchoWorker::new());
    let handles = spawn_pipeline(worker);
    let mut output = handles.take_output_rx().await.ok_or("output already taken")?;

    handles.input_tx.send(PromptRequest {
        session: SessionId::new("demo"),
        request_id: "req-1".into(),
        input: "Hello, pipeline!".into(),
        meta: HashMap::new(),
        deadline: None,
    }).await?;

    let answer = output.recv().await.ok_or("pipeline closed")?;
    println!("{}", answer.text);
    for dropped in handles.dlq.drain() { println!("dropped {}: {}", dropped.request_id, dropped.reason); }
    Ok(())
}

Cargo.toml には tokio-prompt-orchestrator = "2" と tokio = { version = "1", features = ["rt-multi-thread", "macros"] } が必要です。

すでに async-openai、genai、rig、tower を使っているなら、クライアントはそのままでワーカーに変換できます (それぞれ cargo feature)。たとえばローカルの Ollama に向けた async-openai のクライアントなら、こうなります。

use std::sync::Arc;
use async_openai::{config::OpenAIConfig, Client};
use tokio_prompt_orchestrator::{integrations::AsyncOpenAiWorker, spawn_pipeline};

// The client you already have, here pointed at a local Ollama.
let client = Client::with_config(
    OpenAIConfig::new().with_api_base("http://localhost:11434/v1").with_api_key("ollama"),
);
let handles = spawn_pipeline(Arc::new(AsyncOpenAiWorker::new(client, "llama3.2")));

4つの連携はいずれもモックの OpenAI サーバーに対してエンドツーエンドでテストしています (応答、ストリーミング、拒否されたキーはリトライしないこと、429 ではバックオフすること)。

依存クレートについて

feature なしのライブラリはパイプライン、ワーカー、耐障害性の部品だけで、依存ツリーは 167 クレートです。HTTP サーバー (web-api)、OpenTelemetry (otel)、Redis キャッシュなどはすべてオプトインです。

また、細かいところで間違えやすい部分は自前で書かず、広く使われているクレートに任せています。リトライのタイミングは backon、重複排除のキャッシュは moka (サイズ上限があるので、バラバラのプロンプトが大量に来てもメモリが無制限に増えない)、OpenAI のトークン数は tiktoken-rs、/metrics は prometheus です。

リトライについて

orchestrator --retries 2 で、429、503、切断のような一瞬の失敗を、ランダム化した指数バックオフで再試行します。プロバイダの Retry-After は守り、不正な API キーはリトライしません。また、1リクエストの複数回の試行はサーキットブレーカー上では1回として数えます。デフォルトはオフです。

インストール

Linux (x86_64) は依存なしの1行です。

mkdir -p ~/.local/bin && curl -fsSL https://gitlab.com/mattbusel/tokio-prompt-orchestrator/-/releases/permalink/latest/downloads/orchestrator-linux-x86_64.tar.gz | tar xz --strip-components=1 -C ~/.local/bin --wildcards '*/orchestrator'

Windows は orchestrator-windows-x86_64.exe をダウンロードして実行します (署名なしなので SmartScreen が出たら「詳細情報」から「実行」)。macOS やソースからは cargo install --locked tokio-prompt-orchestrator --features web-api,tantivy です。

おわりに

まだ手を入れたいところは多いです。上の表のテストペアはすべて英語なので、日本語のプロンプトでセマンティック重複排除がどう振る舞うかは、正直まだわかっていません。日本語で試した方がいたら、ヒットした/しなかったペアをコメントか Issue で教えてもらえると助かります。

リポジトリ: https://gitlab.com/mattbusel/tokio-prompt-orchestrator
crates.io: https://crates.io/crates/tokio-prompt-orchestrator
docs.rs: https://docs.rs/tokio-prompt-orchestrator
動きを順に再生するサイト: https://tokio-prompt-orchestrator.vercel.app/

MIT ライセンスです。