なぜ遅い呼び出しをジョブにするのか
ジョブキューは、遅い仕事を別の場所で進め、リクエストにはすぐ返事をする仕組みです。モデル呼び出しは数秒かかることがあり、待たせ続ける Web リクエストは、途中で切れたり、ほかの処理を塞いだりします。
ユーザーが四十ページの要約を頼みます。サーバーは数ミリ秒で「受け付けました、job-17」と返します。要約ができるまで、ページは状態を尋ね続けます。
このレッスンのゴール
メモ
私は、遅い呼び出しをリクエストから外し、その進み具合を知らせられます。
キーワード
| ことば | いみ | れい |
|---|---|---|
| ジョブ | 状態を持つ、時間のかかる仕事の単位 | |
| キュー | ワーカーを待つジョブの列 | |
| ワーカー | ジョブを取り出して実行するコード | |
| 冪等キー | 繰り返しても害がないようにする、クライアントの札 | order-1 |
| バックオフ | 失敗するたびに待つ時間を延ばすこと |
学生が入力するセルは、このプレビューでは空白です。
ジョブの四つの状態
ジョブは、待機、実行中、そして完了か失敗のどれかです。再試行する失敗のあとは、待機に戻ります。ジョブは、試行の回数と結果を持ち続けます。
クライアントは、すぐにジョブ ID をもらいます。その ID で状態を尋ね、状態が完了になってから結果を読みます。
例: job-2 だけが一度失敗する
| 段階 | 状態 | 試行 | 時計(job-2 だけ) |
|---|---|---|---|
| 受け付け | 待機 | 0 | 0 秒 |
| 一回目の試行が始まる | 実行中 | 1 | 2 秒 |
| 上流でタイムアウト | 待機 | 1 | 7 秒 |
| 二回目の試行が始まる | 実行中 | 2 | 9 秒 |
| 結果を保存 | 完了 | 2 | 9 秒 |
学生が入力するセルは、このプレビューでは空白です。
冪等キーで二度実行を防ぐ
ネットワークはリクエストを繰り返します。クライアントが送信を二度押すことも、プロキシが再送することもあります。守りがないと、コピーごとにジョブができ、仕事も請求も倍になります。
クライアントは冪等キーを送ります。キューは、キーからジョブ ID への表を持ちます。見たことのあるキーには、今あるジョブを返し、新しいジョブは作りません。
再試行には上限と間隔を
試行の最大回数を決めます。それを超えたジョブは失敗とし、次は人か規則が決めます。失敗のたびに待つ時間を延ばすと、困っているサービスに余裕ができます。
完了したジョブは、巡回が二度訪れても、二度と実行しません。完了したジョブは結果であり、依頼ではないからです。
よくある間違い
注意
再試行はいつも安全だ、と考える人は多いです。安全なのは、仕事が冪等か、キューが覚えているときだけです。支払い済みで終わった呼び出しを再試行すると、また払うことになります。実行の前に、毎回状態を確かめます。
モデル用の補助を書く（1）
docker exec -w /home/node/project agent sh -c "printf '%s\n' 'const base = process.env.AI_URL || \"http://ai:4000\";' 'async function ask(messages, sample, max) {' ' try {' ' const res = await fetch(base + \"/v1/chat/completions\", {' ' method: \"POST\",' ' headers: { \"content-type\": \"application/json\", authorization: \"Bearer lab\" },' ' body: JSON.stringify({ model: \"lab\", messages: messages, max_tokens: max || 200 }),' ' signal: AbortSignal.timeout(30000),' ' });' ' if (res.ok === false) throw new Error(\"status \" + res.status);' ' return (await res.json()).choices[0].message.content;' > model.js"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
モデル用の補助を書く（2）
docker exec -w /home/node/project agent sh -c "printf '%s\n' ' } catch (error) {' ' console.log(\"記録済みサンプル: モデルが答えませんでした (\" + error.message + \")。\");' ' const list = [].concat(sample);' ' const turn = messages.filter(function (m) { return m.role === \"assistant\"; }).length;' ' return list[Math.min(turn, list.length - 1)];' ' }' '}' 'module.exports = { ask };' >> model.js"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
キューを書く（1）
docker exec -w /home/node/project agent sh -c "printf '%s\n' 'const { ask } = require(\"./model\");' 'const jobs = {}; // ジョブ ID からジョブへ' 'const byKey = {}; // 冪等キーからジョブ ID へ' 'let clock = 0; // 作り物の時計(秒)' 'let next = 1;' 'function submit(key, payload) {' ' if (key in byKey) return byKey[key];' ' const id = \"job-\" + next++;' ' jobs[id] = { id: id, payload: payload, state: \"待機\", tries: 0, result: null };' ' byKey[key] = id;' ' return id;' '}' > queue.js"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
キューを書く（2）
docker exec -w /home/node/project agent sh -c "printf '%s\n' 'const failFirst = { \"job-2\": 1 }; // job-2 は一度失敗し、次は成功する' 'function work(job) {' ' if (job.tries <= (failFirst[job.id] || 0)) throw new Error(\"上流でタイムアウト\");' ' return job.payload.toUpperCase();' '}' 'function runOne(job) {' ' if (job.state === \"完了\" || job.state === \"失敗\") return;' ' while (job.tries < 3) {' ' job.tries++;' ' job.state = \"実行中\";' ' clock += 2;' ' try { job.result = work(job); job.state = \"完了\"; return; }' ' catch (error) { job.state = \"待機\"; clock += job.tries * 5; }' ' }' ' job.state = \"失敗\";' '}' >> queue.js"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
キューを書く（3）
docker exec -w /home/node/project agent sh -c "printf '%s\n' 'submit(\"order-1\", \"invoice a\");' 'submit(\"order-2\", \"invoice b\");' 'submit(\"order-1\", \"invoice a\"); // クライアントの再送: 同じキーなので同じジョブ' 'Object.values(jobs).forEach(runOne);' 'Object.values(jobs).forEach(runOne); // 二周目は何も実行しない' 'for (const id in jobs) console.log(id, jobs[id].state, \"試行\", jobs[id].tries, jobs[id].result);' 'console.log(\"ジョブ数:\", Object.keys(jobs).length, \"時計:\", clock + \" 秒\");' >> queue.js"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
キューを書く（4）
docker exec -w /home/node/project agent sh -c "printf '%s\n' 'async function main() {' ' const sample = \"二件の請求書の準備ができました。\";' ' const text = \"二件のジョブが終わりました。お客様向けの短い状況の一文を書いてください。\";' ' console.log(\"本物の呼び出し:\", await ask([{ role: \"user\", content: text }], sample, 60));' '}' 'main();' >> queue.js"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
キューを動かす
docker exec -w /home/node/project agent node queue.jsここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
本物ではなく代用品です
ここでのキューはメモリ上のオブジェクトで、時計はただの数です。プロセスが止まると、ジョブはすべて消えます。本物のシステムは、そのための専用サービスにジョブを置き、ワーカーは別のプロセスで動きます。
考え方は同じです。状態、試行の回数、キーの表、再試行の上限です。
出力の一例
| 行 | 何を示すか |
|---|---|
| job-1 完了 試行 1 INVOICE A | 一回で成功したジョブ |
| job-2 完了 試行 2 INVOICE B | 一度失敗してから成功したジョブ |
| ジョブ数: 2 時計: 11 秒 | 三回の受け付けで、ジョブは二つ |
| 本物の呼び出し | 短い状況の一文。または、ラベル付きの記録済みサンプル |
学生が入力するセルは、このプレビューでは空白です。