diff --git a/README.md b/README.md index 9640d0f9..5840dab7 100644 --- a/README.md +++ b/README.md @@ -3,3 +3,33 @@ insomnia(i6a)は不休のエージェントループを回すためのエージェントプラットフォーム。 ワークフローを統括し、四六時中電力を消費し、イテレーションします。 + +## Crates + +| クレート | 概要 | +|---|---| +| `insomnia` | トップレベルアプリケーション（未実装） | +| `llm-worker` | 自律的なLLMシステムを構築するためのライブラリ | +| `llm-worker-macros` | `llm-worker`用の手続きマクロ (`#[tool_registry]`, `#[tool]`) | + +## ドキュメント + +`llm-worker` の設計ドキュメントは [`crates/llm-worker/docs/`](crates/llm-worker/docs/) に配置されています。 + +### 設計仕様 (`spec/`) +- [basis](crates/llm-worker/docs/spec/basis.md) — 基盤用語とアーキテクチャ概要 +- [timeline](crates/llm-worker/docs/spec/timeline.md) — Timeline抽象レイヤー設計 +- [cache_lock](crates/llm-worker/docs/spec/cache_lock.md) — KVキャッシュ最適化（Type-stateパターン） +- [hooks](crates/llm-worker/docs/spec/hooks.md) — フックライフサイクルと介入システム +- [worker](crates/llm-worker/docs/spec/worker.md) — Workerオーケストレーション設計 +- [cancellation](crates/llm-worker/docs/spec/cancellation.md) — 非同期キャンセル機構 +- [tools](crates/llm-worker/docs/spec/tools.md) — ツールシステム設計 + +### 調査資料 (`research/`) +- [openresponses_mapping](crates/llm-worker/docs/research/openresponses_mapping.md) — Open Responsesマッピング +- [llm-streaming](crates/llm-worker/docs/research/2026-01-02-llm-streaming.md) — LLMストリーミング調査 +- [provider-event-specs](crates/llm-worker/docs/research/2026-01-02-provider-event-specs.md) — プロバイダイベント仕様 + +### 計画 (`plan/`) +- [rig_adoption](crates/llm-worker/docs/plan/rig_adoption.md) — rig参照設計の採用計画 +- [worker_api](crates/llm-worker/docs/plan/worker_api.md) — Worker API/DSL設計計画 diff --git a/crates/llm-worker/docs/plan/rig_adoption.md b/crates/llm-worker/docs/plan/rig_adoption.md new file mode 100644 index 00000000..b39204a9 --- /dev/null +++ b/crates/llm-worker/docs/plan/rig_adoption.md @@ -0,0 +1,207 @@ +# rig 参照設計の導入実装方針 + +## 目的 + +`llm_worker_rs` の既存方針（Timeline中心、Worker主導）を維持しながら、`rig` 由来で有効だった設計を段階導入する。 + +対象は以下の5項目。 + +1. Tool 実行を `ToolServer` 的に分離する +2. Hook をストリーミング粒度まで拡張する +3. Typed output（構造化出力）の第一級 API 化 +4. Provider 固有最適化を provider 層で吸収する +5. GenAI telemetry を標準実装にする + +--- + +## 1. Tool 実行を `ToolServer` 的に分離する ✅ + +### 目的 + +Worker から「ツール登録・呼び出し・定義解決・動的選択」の責務を分離し、Worker は turn orchestration と hook 制御に集中させる。 + +### 実装状況: ✅ 完了 + +- `tool_server.rs` に `ToolServer` / `ToolServerHandle` を実装済み。 +- `Worker` は `ToolServerHandle` を保持し、直接 `Tool` マップを持たない設計に移行済み。 +- `Worker::register_tool()` は内部で `ToolServerHandle::register_tool()` を呼ぶ形に統一済み。 +- ツール実行は `ToolServerHandle::call_tool(name, args)` 経由に統一済み。 +- `tool_definitions_sorted()` で決定的順序のツール定義送信を実装済み。 +- 単体テスト（重複登録・未登録呼び出し・ソート順）を `tool_server.rs` 内に実装済み。 + +### 計画との差分 + +- `ToolExecutor` trait 抽象は未導入。現状 `ToolServerHandle` が具象型として直接使われている。remote tool / MCP 対応時に trait 化する余地あり。 + +--- + +## 2. Hook をストリーミング粒度まで拡張する ✅ + +### 目的 + +Text/Tool delta を受けた時点でフック介入できるようにし、監査・UI 連携・早期停止を改善する。 + +### 実装状況: ✅ 完了 + +- `hook.rs` に以下の streaming 系イベント種別を追加済み: + - `OnTextDelta` (入力: `TextDeltaContext`) + - `OnToolCallDelta` (入力: `ToolCallDeltaContext`) + - `OnStreamChunk` (入力: `StreamChunkContext`) + - `OnStreamComplete` (入力: `StreamCompleteContext`) +- 戻り値は `StreamHookResult { Continue | Abort(String) | Pause }` で統一済み。 +- `Worker` に個別登録 API を実装済み: + - `add_on_text_delta_hook()` + - `add_on_tool_call_delta_hook()` + - `add_on_stream_chunk_hook()` + - `add_on_stream_complete_hook()` +- `HookRegistry` に全 streaming hook のストレージを追加済み。 +- `worker.rs` の stream dispatch ループ内で `BlockDelta` 受信時に各 hook を順序保証付きで呼び出し済み。 +- `Abort` / `Pause` の制御線が Worker の run ループに接続済み。 + +### 計画との差分 + +- 高レベル統合 API（subscriber/DSL）は `worker_api_plan` 側で別途進行中。 +- 実行順序仕様の文書化は未完了（コード上は登録順実行を実装済み）。 + +--- + +## 3. Typed output（構造化出力）の第一級 API 化 ⬚ + +### 目的 + +「テキストを返して呼び出し側で JSON parse」から脱却し、型付きレスポンスを API レベルで保証する。 + +### 実装状況: ⬚ 未着手 + +- `run_typed()` API は未実装。 +- `output_schema` の request builder 統合は未実装。 +- エラー種別（`StructuredOutputUnsupported` / `Deserialize` / `Empty`）は未追加。 + +### 次のステップ + +1. `output_schema` を Worker request builder に追加 +2. typed 実行 API と decode ロジックを実装 +3. provider capability 判定を `llm_client` 層に追加 +4. エラー型とユーザー向けメッセージを整理 + +--- + +## 4. Provider 固有最適化を provider 層で吸収する 🔄 + +### 目的 + +キャッシュ、beta header、細粒度 stream 差分などの provider 特有処理を Worker から排除する。 + +### 実装状況: 🔄 部分的に進行 + +- `llm_client/providers/*` に Anthropic / OpenAI / Gemini / Ollama の provider 実装が存在する。 +- `llm_client/scheme/*` に各 provider の request/events 変換が分離済み。 +- Worker は provider 非依存の `Event` を Timeline 経由で処理する設計に移行済み。 +- ただし `ProviderOptions` 型や capability trait は未導入。 + +### 次のステップ + +1. provider capability trait を追加 +2. 既存 provider 実装を capability 駆動へ移行 +3. Worker 側に残る provider 分岐があれば削除 + +--- + +## 5. GenAI telemetry を標準実装にする ⬚ + +### 目的 + +運用時に必要な追跡情報（turn、tool call、usage、latency、abort reason）を一貫して記録可能にする。 + +### 実装状況: ⬚ 未着手 + +- `tracing` クレートは依存に含まれており、`worker.rs` 内で `debug!` / `info!` / `trace!` / `warn!` マクロが使用されている。 +- ただし GenAI semantic conventions に沿った構造化 span は未実装。 +- `TelemetryConfig` / telemetry モジュールは未追加。 + +### 次のステップ + +1. telemetry モジュール追加（field key 定義） +2. Worker/timeline/tool/hook に span 計装 +3. usage と stop reason を終了イベントで集約 +4. ログとメトリクス出力の最小 adapter を実装 + +--- + +## 導入順序（更新版） + +1. ~~ToolServer 分離~~ ✅ +2. ~~Streaming Hook 拡張~~ ✅ +3. Telemetry 標準化 ⬚ +4. Typed output API ⬚ +5. Provider capability 移行の仕上げ 🔄 + +--- + +## チケット分解（実装順・進捗付き） + +### Epic A: ToolServer 分離 ✅ + +| ID | タイトル | 状態 | 備考 | +| --- | --- | --- | --- | +| A1 | ToolServer 最小骨格の追加 | ✅ | `tool_server.rs` に `ToolServer` / `ToolServerHandle` 実装済み | +| A2 | Worker の tool map を ToolServerHandle に置換 | ✅ | `Worker` は `ToolServerHandle` を保持。`register_tool` / `call_tool` / `tool_definitions_sorted` 全て経由 | +| A3 | Hook context を ToolServer 参照へ接続 | ✅ | `pre_tool_call` / `post_tool_call` で `ToolCallContext` に `meta` / `tool` を `ToolServerHandle::get_tool()` から取得 | +| A4 | KV キャッシュ保護ルールの実装 | ✅ | `CacheLocked` 型状態パターンで実装。`lock()` 後は prefix の変更が型レベルで防止される | +| A5 | ToolServer 単体/統合テスト追加 | ✅ | `tool_server.rs` 内に重複登録・未登録呼び出し・ソート順のテスト実装済み | + +### Epic B: Streaming Hook 拡張 ✅ + +| ID | タイトル | 状態 | 備考 | +| --- | --- | --- | --- | +| B1 | Hook event 種別追加（delta/stream） | ✅ | `OnTextDelta` / `OnToolCallDelta` / `OnStreamChunk` / `OnStreamComplete` 定義済み | +| B2 | Timeline dispatch に hook 呼び出し点追加 | ✅ | `BlockDelta` 受信時に hook 実行。stream dispatch ループ内に組み込み済み | +| B3 | Abort/Pause 制御線の接続 | ✅ | `StreamHookResult` の `Abort` / `Pause` が Worker の run ループに接続済み | +| B4 | 実行順序仕様と回帰テスト | 🔄 | コード上は登録順実行を実装。仕様文書化・専用回帰テストは未完了 | + +### Epic C: Telemetry 標準化 ⬚ + +| ID | タイトル | 状態 | 備考 | +| --- | --- | --- | --- | +| C1 | telemetry モジュールと semantic key 定義 | ⬚ | 未着手 | +| C2 | worker/turn/llm/tool/hook span 計装 | ⬚ | 未着手。`tracing` の基本利用はあるが構造化 span は未実装 | +| C3 | マスキング設定と no-op 運用 | ⬚ | 未着手 | +| C4 | telemetry テスト/サンプル整備 | ⬚ | 未着手 | + +### Epic D: Typed Output API ⬚ + +| ID | タイトル | 状態 | 備考 | +| --- | --- | --- | --- | +| D1 | Request builder に output_schema 追加 | ⬚ | 未着手 | +| D2 | `run_typed` / `run_typed_with_history` 実装 | ⬚ | 未着手 | +| D3 | 非対応 provider のフォールバックとエラー分離 | ⬚ | 未着手 | +| D4 | typed 出力の回帰テスト | ⬚ | 未着手 | + +### Epic E: Provider Capability 移行 🔄 + +| ID | タイトル | 状態 | 備考 | +| --- | --- | --- | --- | +| E1 | provider capability trait 導入 | ⬚ | 未着手 | +| E2 | Anthropic/OpenAI/Gemini の capability 移行 | 🔄 | scheme 層での request/events 分離は完了。capability trait は未導入 | +| E3 | Worker 側 provider 分岐の削除 | 🔄 | Worker は Event ベースで provider 非依存に近いが、完全な分離は未検証 | +| E4 | snapshot テストと責務ドキュメント更新 | ⬚ | 未着手 | + +--- + +## チケット運用ルール + +- 各 ticket は `feature/-...` ブランチで実装する。 +- 依存 ticket が `main` に入るまで、後続 ticket は着手しない。 +- 各 ticket で最低 1 つ回帰テストを追加する（ドキュメント専用 ticket を除く）。 +- `CacheLocked` の prefix 不変性を崩す変更は、A4 と同時に防止テストを追加する。 +- tool 定義のリクエスト化は決定的順序にする（内部保持が `HashMap` でも送信順は固定）。 + +## 非目標 + +- 既存 Worker API の全面破壊的変更 +- 1回の PR で5項目を同時に実装 +- Open Responses 完全準拠の即時達成 + +## 備考 + +`llm_worker_rs` は Timeline 駆動設計が中核であり、この方針は維持する。`rig` の設計は部品として参照し、全体アーキテクチャは置換しない。 diff --git a/crates/llm-worker/docs/plan/worker_api.md b/crates/llm-worker/docs/plan/worker_api.md new file mode 100644 index 00000000..b6d36872 --- /dev/null +++ b/crates/llm-worker/docs/plan/worker_api.md @@ -0,0 +1,147 @@ +# Worker API/DSL 実装計画 + +## 目的 + +- [Open Responses](https://www.openresponses.org)(以後"OR")に準拠した正規化を前提に、 + Item/Part の2段スコープを扱える Worker API を設計する。 +- APIの煩雑化を防ぐため、worker.on_xxx として公開するのを避けつつ、 + Text/Thinking/Tool など型の違いを静的に扱える DSL を提供する。 + +## 方針 + +- 内部は Timeline が Event を正規化し、Item/Part/Meta + を単一ストリームとして扱う。 +- API では Item/Part 型ごとに ctx を持てるようにし、DSL + で記述の冗長さを削減する。 +- まず macro_rules! 版を作り、必要なら proc-macro に拡張する。 +- Item/Part の型パラメータはクレートが公開する Kind 型を使う。 + +## 仕様の前提 + +- Item は OR の item (message, function_call, reasoning など) に対応する。 +- Part は OR の content part (output_text, reasoning_text など) に対応する。 +- Item は必ず start/stop を持つ。Part は Item 内で複数発生し得る。 +- Item/Part の型指定は `Item` / `Part` のように書く。 + +--- + +## 設計ステップ + +### 1. 内部イベントモデルの整理 ✅ + +- Event を Item/Part/Meta の3層に整理する。 + +#### 実装状況: ✅ 完了 + +`timeline/event.rs` に3層のイベントモデルを実装済み: +- **Meta イベント**: `Ping`, `Usage`, `Status`, `Error` +- **Block イベント**: `BlockStart`, `BlockDelta`, `BlockStop`, `BlockAbort` +- Block 種別として `Text`, `Thinking`, `ToolUse`, `ToolResult` を定義 +- `DeltaContent` で `Text(String)`, `Thinking(String)`, `InputJson(String)` を区別 + +#### 計画との差分 + +- 計画の「ItemEvent / PartEvent は型パラメータで区別する」方針ではなく、`BlockType` enum + `DeltaContent` enum で区別する設計を採用。Item/Part の2段ではなく Block/Delta の2段として実装。 +- OR の Item 型は `llm_client/types.rs` の `Item` enum で直接モデル化済み（`Message`, `FunctionCall`, `FunctionCallOutput`, `Reasoning`）。 + +### 2. スコープの二段化 ✅ + +- Item ctx: Item 型ごとに1つ +- Part ctx: Part 型ごとに1つ + +#### 実装状況: ✅ 完了（Block/Handler スコープとして実装） + +`handler.rs` の `Handler` trait で Block 単位のスコープを実装済み: +- `type Scope: Default` で Handler ごとのスコープ型を定義 +- `on_event(&mut self, scope: &mut Self::Scope, event: &K::Event)` でスコープ付きイベント処理 +- `timeline.rs` の `ErasedHandler` で `start_scope()` / `end_scope()` のライフサイクル管理 + +#### 計画との差分 + +- 計画の「Item ctx + Part ctx の二段」ではなく、Block 単位の単一スコープとして実装。 +- Block 開始で `Scope::default()` 生成、Block 終了で破棄する方式。Item/Part の入れ子構造は持たない。 +- これは十分実用的であり、追加の階層化は必要に応じて将来検討。 + +### 3. Handler trait の再定義 ✅ + +- Item/Part を型で指定できる trait を導入する。 + +#### 実装状況: ✅ 完了（Kind/Handler として実装） + +`handler.rs` に以下を実装済み: +- `Kind` trait: イベント型を関連型 `type Event` で指定するマーカー trait +- `Handler` trait: Kind ごとのイベント処理 +- Block Kind 定義: + - `TextBlockKind` (Event = `TextBlockEvent { Start | Delta | Stop }`) + - `ThinkingBlockKind` (Event = `ThinkingBlockEvent { Start | Delta | Stop }`) + - `ToolUseBlockKind` (Event = `ToolUseBlockEvent { Start | InputJsonDelta | Stop }`) +- Meta Kind 定義: + - `UsageKind`, `PingKind`, `StatusKind`, `ErrorKind` + +#### 計画との差分 + +- 計画の `ItemHandler` / `PartHandler` 構造ではなく、`Handler` の単一 trait で統一。 +- PartHandler に ItemCtx を必須で渡す設計は採用せず、Block 単位のフラットな構造。 + +### 4. Timeline との結合 ✅ + +- Timeline は BlockStart で Scope を生成、BlockStop で Scope を破棄。 + +#### 実装状況: ✅ 完了 + +`timeline/timeline.rs` に以下を実装済み: +- `Timeline` が `ErasedHandler` のリストを保持 +- `dispatch(&Event)` でイベントを各 Handler にディスパッチ +- `BlockStart` で `start_scope()`, `BlockStop` / `BlockAbort` で `end_scope()` +- 実用的な collector として `TextBlockCollector`, `ToolCallCollector` を実装済み +- `Worker` が Timeline を所有し、stream ループ内で `timeline.dispatch()` を呼び出し + +`subscriber.rs` の `WorkerSubscriber` trait で高レベルのイベント購読を提供: +- `on_text_block()`, `on_tool_use_block()` (スコープ付き) +- `on_usage()`, `on_status()`, `on_error()` (メタイベント) +- `on_text_complete()`, `on_tool_call_complete()` (完了イベント) +- `on_turn_start()`, `on_turn_end()` (ターン制御) + +### 5. DSL (macro_rules!) の導入 ⬚ + +- 宣言的 DSL を提供する。 + +#### 実装状況: ⬚ 未着手 + +- `macro_rules!` による `handler!` DSL は未実装。 +- 代わりに `llm-worker-macros` クレートで `#[tool_registry]` / `#[tool]` proc-macro を実装済み（ツール定義の自動生成用）。 +- Handler 定義の DSL は未提供。現状は trait を直接実装する方式。 + +#### 次のステップ + +- Handler 定義の冗長さが問題になった時点で `handler!` DSL を導入する。 +- 現状の `Handler` trait 直接実装 + `WorkerSubscriber` trait の組み合わせで多くのユースケースをカバーできているため、優先度は低い。 + +### 6. 拡張ポイント 🔄 + +- 追加 Part (output_image など) を DSL に追加しやすい形にする。 + +#### 実装状況: 🔄 部分的 + +- `BlockType` enum に新しい種別を追加し、対応する `Kind` と `Event` を定義すれば拡張可能な設計。 +- `ThinkingBlockKind` が後から追加された実績あり。 +- proc-macro への移行は未実施（現時点では不要）。 + +--- + +## 実装順序（更新版） + +1. ~~Event/Item/Part の型定義の整理~~ ✅ +2. ~~Item/Part ctx を持つ Timeline 実装~~ ✅ (Block/Scope として) +3. ~~Handler trait の定義・既存コードの移行~~ ✅ +4. macro_rules! DSL の実装 ⬚ +5. ~~既存ユースケースの移植~~ ✅ (WorkerSubscriber 経由) + +--- + +## TODO（更新版） + +- ~~Item と Part の型対応表を整理する~~ ✅ `Item` enum + `BlockType` + `Kind` で対応完了 +- ~~OR と既存 llm_client の差分を再確認する~~ ✅ `Item` 型が OR ネイティブに移行済み（Message / FunctionCall / FunctionCallOutput / Reasoning） +- ~~Tool args の delta を OR 拡張として扱うか検討する~~ ✅ `ToolUseBlockEvent::InputJsonDelta` として実装 +- macro_rules! で表現可能な DSL の最小文法を確定する ⬚ （優先度低: WorkerSubscriber で大半のユースケースをカバー済み） diff --git a/crates/llm-worker/docs/spec/basis.md b/crates/llm-worker/docs/spec/basis.md new file mode 100644 index 00000000..e1d7b9d8 --- /dev/null +++ b/crates/llm-worker/docs/spec/basis.md @@ -0,0 +1,143 @@ +# llm-worker + +LLMを用いた自律的なワーカーを構築するためのライブラリ。 + +ツールの定義・呼び出し、コンテキスト管理、ストリーミングイベント処理、フック制御などを責務にもつ。 + +ワークスペース `insomnia` に属し、以下の3クレートで構成される: + +| クレート | 概要 | +|---|---| +| `insomnia` | アプリケーション本体 | +| `llm-worker` | LLMワーカーライブラリ (本クレート) | +| `llm-worker-macros` | `#[tool_registry]` / `#[tool]` 手続きマクロ | + +## 用語定義 + +- **Item**: + 会話の基本単位。Open Responses仕様に準じた列挙型で、`Message`・`FunctionCall`・`FunctionCallOutput`・`Reasoning`の4バリアントを持つ。従来のメッセージベースAPIと異なり、ツール呼び出しや思考をファーストクラスのItemとして扱う。 +- **ContentPart**: + Message Item内のコンテンツ要素。`InputText`(ユーザー入力)・`OutputText`(アシスタント出力)・`Refusal`(拒否)を区別する。 +- **ブロック (Block)**: + モデルが「開始→デルタ→終了」で囲む一塊の出力。OpenAIの`response.output_item`やAnthropicの`content_block`、Geminiの`candidates[].content.parts`などが該当する。`BlockType`として`Text`・`Thinking`・`ToolUse`・`ToolResult`が定義されている。 +- **メタイベント (Meta Event)**: + `Ping`・`Usage`・`Status`・`Error`など、ブロックに属さない補助イベント群。ステータス更新やハートビートとして処理される。 +- **Event**: + LLMからのストリーミングレスポンスを表現する統一型。メタイベントとブロックイベント(`BlockStart`・`BlockDelta`・`BlockStop`・`BlockAbort`)から成る。`llm_client`層と`event`モジュールの双方に定義が存在する。 +- **Worker**: + LLMとのインタラクションを管理する中心コンポーネント。ツール実行ループ、フック呼び出し、ストリーミングイベントのディスパッチを統括する。 + +以降の仕様では「ブロック」を上記の生成・ツール単位の総称として扱う。 + +## アーキテクチャ + +モジュール構成概念図: + +```plaintext +llm-worker +├── worker # Workerコア (実行ループ、状態管理) +├── state # Type-stateパターン (Mutable / CacheLocked) +├── tool # Tool trait, ToolMeta, ToolDefinition +├── tool_server # ToolServer (ツールのレジストリと実行) +├── hook # Hook trait, HookRegistry (10種のフックポイント) +├── handler # Kind/Handler trait (イベントディスパッチ) +├── subscriber # WorkerSubscriber trait (UI向けイベント購読) +├── event # Worker層の公開Event型 +├── message # Item/ContentPart/Roleのre-export +├── timeline # イベントストリームの状態管理とHandlerディスパッチ +│ ├── event # Timeline内部イベント型 +│ ├── timeline # Timelineコア +│ ├── text_block_collector # テキスト収集Handler +│ └── tool_call_collector # ツール呼び出し収集Handler +└── llm_client # LLMクライアント層 + ├── client # LlmClient trait + ├── event # llm_client層のEvent型 + ├── error # ClientError + ├── types # Item, ContentPart, Role, Request, RequestConfig, ToolDefinition + ├── scheme # APIスキーマ変換 + │ ├── openai # OpenAI Chat Completions API + │ ├── anthropic # Anthropic Messages API + │ └── gemini # Google Gemini API + └── providers # プロバイダ固有クライアント + ├── openai # OpenAI + ├── anthropic # Anthropic + ├── gemini # Google Gemini + └── ollama # Ollama (ローカルLLM) +``` + +OpenAI互換のプロバイダでスキーマを使い回せるよう、`scheme`と`providers`モジュールは分離されている。 + +### scheme層 + +単純な変換を責務とするスキーマを定義する。 + +- リクエスト変換: `Request`(SystemPrompt + Items + Tools + RequestConfig) → プロバイダ固有のリクエストJSON +- レスポンス変換: SSEイベント → 型付き`Event`構造体のストリーム + +各APIスキーマ(OpenAI / Anthropic / Gemini)ごとに実装を持ち、APIスキーマに準じたパース・バリデーションを行う。 + +### llm_client (providers) 層 + +`LlmClient` traitを実装する各プロバイダクライアントがリクエストを送信し、差異が吸収され統一された`Event`ストリームを出力する。 + +ストリーミング中のバッファリング(ToolArguments の累積等)もこの層で行う。 + +`ConfigWarning`により、プロバイダがサポートしない設定オプションを警告として通知する仕組みを持つ。 + +### timeline層 + +`llm_client`からのイベントストリームを受信し、登録された`Handler`にディスパッチする。 + +`Kind` traitがイベント型を定義し、`Handler` traitが`Scope`(ブロックのライフサイクルに対応する状態)を持つイベント処理を行う。組み込みHandlerとして`TextBlockCollector`と`ToolCallCollector`が提供される。 + +### worker層 + +`Worker`はType-stateパターンにより`Mutable`状態と`CacheLocked`状態を持つ。 + +- **Mutable**: システムプロンプトの設定変更、メッセージ履歴の編集、ツール・フックの登録が可能 +- **CacheLocked**: `Worker::lock()`で遷移。LLM APIのKVキャッシュヒット率を最大化するため、既存履歴の変更を制限する + +`WorkerSubscriber` traitにより、テキスト生成やツール呼び出しのストリーミングイベントをUI等に配信できる。 + +## Tools + +各種プロバイダのLLMのAPI仕様として存在するTool CallやFunction Callingを`Tool` traitとして統一的に扱う。 + +- **`ToolMeta`**: ツール名・説明・入力スキーマ(JSON Schema)を保持する不変のメタ情報。Worker登録後は変更されない。 +- **`ToolDefinition`**: `Fn() -> (ToolMeta, Arc)`型のファクトリ。Worker登録時に一度呼び出され、メタ情報とインスタンスがセッションスコープでキャッシュされる。 +- **`Tool` trait**: `async fn call(&self, arguments: Value) -> Result`を持つ非同期trait。セッション中に状態を保持できる。 +- **`ToolServer`**: 登録済みツールのレジストリと実行を管理するインメモリサーバー。 + +`llm-worker-macros`クレートが提供する`#[tool_registry]` / `#[tool]`手続きマクロにより、`impl`ブロック上のメソッドから`Tool` trait実装を自動生成する。 + +## Hooks + +Claude Codeに存在するようなHooksと同様のターン制御・介入機構を搭載する。 + +以下の10種のフックポイントが定義されている: + +| フックポイント | タイミング | 戻り値による制御 | +|---|---|---| +| `OnPromptSubmit` | プロンプト送信時 | Continue / Cancel | +| `PreLlmRequest` | LLMリクエスト直前 | Continue / Cancel | +| `PreToolCall` | ツール実行直前 | Continue / Skip / Abort / Pause | +| `PostToolCall` | ツール実行直後 | Continue / Abort | +| `OnTurnEnd` | ターン終了時 | Finish / ContinueWithMessages / Paused | +| `OnAbort` | 中断時 | (通知のみ) | +| `OnTextDelta` | テキストデルタ受信時 | (ストリーミング) | +| `OnToolCallDelta` | ツール引数デルタ受信時 | (ストリーミング) | +| `OnStreamChunk` | ストリームチャンク受信時 | (ストリーミング) | +| `OnStreamComplete` | ストリーム完了時 | (ストリーミング) | + +フックは`Hook` traitとして表現され、`HookRegistry`で管理される。 + +## ストリーミングイベント処理 + +イベントストリームを扱う設計目標として、Text / InputJSON / Thinkingのデルタと、[細粒度ツールストリーミング](https://platform.claude.com/docs/en/agents-and-tools/tool-use/fine-grained-tool-streaming)を含む、デルタの低遅延・リアルタイム処理がある。 + +ブロックのライフサイクル: `BlockStart` → `BlockDelta`(複数回) → `BlockStop` / `BlockAbort` + +`DeltaContent`の種別: +- `Text(String)` - テキスト生成のデルタ +- `Thinking(String)` - 思考(Extended Thinking等)のデルタ +- `InputJson(String)` - ツール引数JSONの部分文字列デルタ diff --git a/crates/llm-worker/docs/spec/cache_lock.md b/crates/llm-worker/docs/spec/cache_lock.md new file mode 100644 index 00000000..e7b3a0c5 --- /dev/null +++ b/crates/llm-worker/docs/spec/cache_lock.md @@ -0,0 +1,189 @@ +# KVキャッシュを中心とした設計 + +LLMのKVキャッシュのヒット率を重要なメトリクスであるとし、APIレベルでキャッシュ操作を中心とした設計を行う。 + +## 前提 + +リクエスト間キャッシュ(Context Caching)は、複数のリクエストで同じ入力トークン列が繰り返された際、プロバイダ側が計算済みの状態を再利用することでレイテンシと入力コストを下げる仕組みである。 +キャッシュは主に**先頭一致 (Common Prefix)** によってHitするため、前提となるシステムプロンプトや、会話ログの過去部分（前方）を変化させると、以降のキャッシュは無効となる。 + +## 要件 + +1. **前方不変性の保証 (Prefix Immutability)** + * 後方に会話が追加されても、前方のデータ（システムプロンプトや確定済みのメッセージ履歴）が変化しないことをAPIレベルで保証する。 + * これにより、意図しないキャッシュミス（Cache Miss）を防ぐ。 + +2. **データ上の再現性** + * コンテキストのデータ構造が同一であれば、生成されるリクエスト構造も同一であることを保証する。 + * シリアライズ結果のバイト単位の完全一致までは求めないが、論理的なリクエスト構造は保たれる必要がある。 + +## アプローチ: Type-state Pattern + +RustのType-stateパターンを利用し、Workerの状態によって利用可能な操作をコンパイル時に制限する。 + +### 1. 状態定義（`state.rs`） + +`WorkerState`トレイトはsealedパターンで実装され、外部からの実装を防ぐ: + +```rust +pub trait WorkerState: private::Sealed + Send + Sync + 'static {} + +mod private { + pub trait Sealed {} +} +``` + +* **`Mutable` (初期状態)** + * 自由な編集が可能な状態。 + * システムプロンプトの設定・変更が可能。 + * メッセージ履歴の初期構築（ロード、編集、クリア）が可能。 + * ツール・フックの登録が可能。 + * リクエスト設定（`max_tokens`, `temperature`, `top_p`, `top_k`, `stop_sequences`）の変更が可能。 +* **`CacheLocked` (キャッシュ保護状態)** + * キャッシュの有効活用を目的とした、前方不変状態。 + * **システムプロンプトの変更不可**。 + * **既存メッセージ履歴の変更不可**（`run()`による末尾追記のみ許可）。 + * 実行（`run`）はこの状態で行うことを推奨する。 + +両状態とも`Debug`, `Clone`, `Copy`, `Default`を実装する。 + +### 2. Worker構造体 + +`Worker`は状態パラメータ`S: WorkerState`を持ち、デフォルトは`Mutable`: + +```rust +pub struct Worker { + client: C, + timeline: Timeline, + text_block_collector: TextBlockCollector, + tool_call_collector: ToolCallCollector, + tool_server: ToolServerHandle, + hooks: HookRegistry, + system_prompt: Option, + history: Vec, + locked_prefix_len: usize, // ロック時の履歴長 + turn_count: usize, + turn_notifiers: Vec>, + request_config: RequestConfig, + last_run_interrupted: bool, + cancel_tx: mpsc::Sender<()>, + cancel_rx: mpsc::Receiver<()>, + _state: PhantomData, +} +``` + +`Worker`自身がコンテキスト（履歴）のオーナーとなり、状態によってアクセサを制限する。 + +### 3. 状態遷移とAPI + +#### 共通API（全状態で利用可能） + +```rust +impl Worker { + pub async fn run(&mut self, user_input: impl Into) -> Result; + pub fn history(&self) -> &[Item]; // 参照のみ + pub fn system_prompt(&self) -> Option<&str>; + pub fn cancel_token(&self) -> CancelToken; // キャンセルトークン取得 + // ... その他参照系メソッド +} +``` + +#### Mutable限定API + +```rust +impl Worker { + pub fn new(client: C) -> Self; + + // システムプロンプト + pub fn system_prompt(self, prompt: impl Into) -> Self; // ビルダー + pub fn set_system_prompt(&mut self, prompt: impl Into); + + // 履歴操作 + pub fn history_mut(&mut self) -> &mut Vec; + pub fn set_history(&mut self, items: Vec); + pub fn with_item(self, item: Item) -> Self; + pub fn push_item(&mut self, item: Item); + pub fn with_items(self, items: impl IntoIterator) -> Self; + pub fn extend_history(&mut self, items: impl IntoIterator); + pub fn clear_history(&mut self); + + // ツール登録 + pub fn register_tool(&mut self, factory: WorkerToolDefinition) -> Result<(), ToolRegistryError>; + pub fn register_tools(&mut self, factories: impl IntoIterator) -> Result<(), ToolRegistryError>; + + // リクエスト設定 + pub fn max_tokens(self, max_tokens: u32) -> Self; + pub fn temperature(self, temperature: f32) -> Self; + pub fn top_p(self, top_p: f32) -> Self; + pub fn top_k(self, top_k: u32) -> Self; + pub fn stop_sequence(self, sequence: impl Into) -> Self; + pub fn with_config(self, config: RequestConfig) -> Self; + pub fn validate(self) -> Result; + + // 状態遷移 + pub fn lock(self) -> Worker; +} +``` + +#### CacheLocked限定API + +```rust +impl Worker { + pub fn locked_prefix_len(&self) -> usize; + pub fn unlock(self) -> Worker; +} +``` + +### 4. 使用例 + +```rust +// 1. Mutable状態で初期化 +let mut worker = Worker::new(client) + .system_prompt("You are a helpful assistant.") + .max_tokens(4096); + +// 2. コンテキストの構築（Mutableなので自由に変更可） +worker.push_item(Item::user_message("Hello")); +worker.register_tool(my_tool)?; + +// 3. ロックしてCacheLocked状態へ遷移 +// ここまでの履歴長がlocked_prefix_lenとして記録される +let mut locked_worker = worker.lock(); + +// 4. 利用（CacheLocked状態） +// 実行は可能。新しいメッセージは履歴の末尾に追記される。 +// 前方の履歴やシステムプロンプトは変更できないため、キャッシュヒットが保証される。 +locked_worker.run("user input").await?; + +// NG操作（コンパイルエラー） +// locked_worker.set_system_prompt("New prompt"); +// locked_worker.history_mut().clear(); +// locked_worker.register_tool(another_tool); + +// 5. 必要に応じてアンロック（キャッシュ保護は解除される） +let mut mutable_worker = locked_worker.unlock(); +mutable_worker.set_system_prompt("New prompt"); +``` + +### 5. lock/unlockの実装 + +`lock()`と`unlock()`はWorkerの全フィールドを移動して新しい状態のWorkerを構築する。値の所有権移動のため、元のWorkerは使用不可になる。 + +- `lock()`: `locked_prefix_len`に現在の`history.len()`を記録 +- `unlock()`: `locked_prefix_len`を0にリセット + +### 6. UsageEventによるキャッシュ監視 + +`UsageEvent`には`cache_read_input_tokens`と`cache_creation_input_tokens`が含まれており、キャッシュヒット率のモニタリングが可能: + +```rust +pub struct UsageEvent { + pub input_tokens: Option, + pub output_tokens: Option, + pub total_tokens: Option, + pub cache_read_input_tokens: Option, + pub cache_creation_input_tokens: Option, +} +``` + +TimelineのUsageKind Handlerを登録することで、キャッシュの効果を実行時に確認できる。 diff --git a/crates/llm-worker/docs/spec/cancellation.md b/crates/llm-worker/docs/spec/cancellation.md new file mode 100644 index 00000000..1c54ffe3 --- /dev/null +++ b/crates/llm-worker/docs/spec/cancellation.md @@ -0,0 +1,251 @@ +# 非同期キャンセル仕様 + +Workerの非同期キャンセル機構についての仕様ドキュメント。 + +## 概要 + +`tokio::sync::mpsc`チャネル（バッファサイズ1）を用いて、別タスクからWorkerの実行を安全にキャンセルできる。Worker内部では`tokio::select!`により、ストリーム受信・ツール実行の各フェーズでキャンセルシグナルを検知する。 + +## 基本的な使い方 + +### cancel() メソッドによるキャンセル + +```rust +let worker = Arc::new(Mutex::new(Worker::new(client))); + +// 実行タスク +let w = worker.clone(); +let handle = tokio::spawn(async move { + w.lock().await.run("prompt").await +}); + +// キャンセル（try_sendによる非同期安全な送信） +worker.lock().await.cancel(); +``` + +### cancel_sender() によるキャンセル + +ロックを取得せずにキャンセルする場合、事前に`Sender`を取得しておく。 + +```rust +let worker = Arc::new(Mutex::new(Worker::new(client))); + +// ロック中にSenderを取得 +let cancel_tx = { + let w = worker.lock().await; + w.cancel_sender() +}; + +// 実行タスク +let worker_clone = worker.clone(); +let task = tokio::spawn(async move { + let mut w = worker_clone.lock().await; + w.run("Tell me a long story").await +}); + +// 別タスクからキャンセル（ロック不要） +tokio::spawn(async move { + tokio::time::sleep(Duration::from_secs(2)).await; + let _ = cancel_tx.send(()).await; +}); + +task.await?; +``` + +## API + +| メソッド / フィールド | 説明 | +| --------------------- | --------------------------------------------- | +| `cancel()` | `try_send`でキャンセルをトリガー | +| `cancel_sender()` | `mpsc::Sender<()>`のcloneを返す | +| `is_cancelled()` | キャンセルキューにシグナルがあるか確認 | +| `last_run_interrupted()` | 前回のrunが中断されたかどうか | + +## キャンセル検知ポイント + +Worker内部には複数のキャンセル検知ポイントが存在する。 + +### 1. ターンループ先頭 + +```rust +loop { + if self.try_cancelled() { + self.timeline.abort_current_block(); + return Err(WorkerError::Cancelled); + } + // ... +} +``` + +各ターンの開始時に`try_recv()`でキャンセルキューを確認する。 + +### 2. ストリーム取得時 + +```rust +let mut stream = tokio::select! { + stream_result = self.client.stream(request) => stream_result?, + cancel = self.cancel_rx.recv() => { + self.timeline.abort_current_block(); + return Err(WorkerError::Cancelled); + } +}; +``` + +LLMクライアントへのリクエスト送信中にキャンセル可能。 + +### 3. ストリーム受信中 + +```rust +loop { + tokio::select! { + event_result = stream.next() => { + // イベント処理 + } + cancel = self.cancel_rx.recv() => { + self.timeline.abort_current_block(); + return Err(WorkerError::Cancelled); + } + } +} +``` + +ストリーミング中のイベント受信ループで、各イベント間にキャンセルが割り込める。 + +### 4. ツール並列実行中 + +```rust +let mut results = tokio::select! { + results = join_all(futures) => results, + cancel = self.cancel_rx.recv() => { + self.timeline.abort_current_block(); + return Err(WorkerError::Cancelled); + } +}; +``` + +`join_all`によるツール並列実行中にもキャンセル可能。 + +## キャンセル時の処理フロー + +``` +キャンセルシグナル検知 + ↓ +timeline.abort_current_block() // 進行中ブロックの終端処理 + ↓ +last_run_interrupted = true // 中断フラグをセット + ↓ +Err(WorkerError::Cancelled) を返す + ↓ +finalize_interruption() // 中断の最終処理 + ↓ +run_on_abort_hooks("Cancelled") // on_abort フック呼び出し + ↓ +Err(WorkerError::Cancelled) を返す（呼び出し元へ） +``` + +## キャンセルキューの管理 + +### drain_cancel_queue + +`run_turn_loop()`の開始時に、キューに溜まった古いキャンセルシグナルを排出する。これにより、前回のキャンセルが次回の`run()`に影響することを防ぐ。 + +```rust +fn drain_cancel_queue(&mut self) { + loop { + match self.cancel_rx.try_recv() { + Ok(()) => continue, + Err(TryRecvError::Empty) | Err(TryRecvError::Disconnected) => break, + } + } +} +``` + +### try_cancelled + +ノンブロッキングでキャンセル状態を確認する。チャネルがdisconnectedの場合もキャンセル扱いとなる。 + +```rust +fn try_cancelled(&mut self) -> bool { + match self.cancel_rx.try_recv() { + Ok(()) => true, + Err(TryRecvError::Empty) => false, + Err(TryRecvError::Disconnected) => true, + } +} +``` + +## 中断状態の管理 + +### last_run_interrupted フラグ + +Workerは`last_run_interrupted`フラグで前回の実行が中断されたかどうかを追跡する。 + +- `run()` / `resume()` の開始時に`false`にリセット +- エラー発生時に`true`にセット +- `Pause`による中断時にも`true`にセット +- 正常終了（`WorkerResult::Finished`）時に`false`にセット + +### finalize_interruption + +すべての`run()`/`resume()`の結果は`finalize_interruption()`を経由して返される。結果が`Err`の場合、中断理由を抽出して`on_abort`フックを呼び出す。 + +```rust +async fn finalize_interruption(&mut self, result: Result) -> Result { + match result { + Ok(value) => Ok(value), + Err(err) => { + self.last_run_interrupted = true; + let reason = match &err { + WorkerError::Aborted(reason) => reason.clone(), + WorkerError::Cancelled => "Cancelled".to_string(), + _ => err.to_string(), + }; + self.run_on_abort_hooks(&reason).await?; + Err(err) + } + } +} +``` + +## on_abort フック + +`on_abort`フックはキャンセルだけでなく、あらゆる中断時に発火する。 + +**入力**: `&mut String` - 中断理由 + +**発火条件**: + +- `WorkerError::Cancelled` -- reason: `"Cancelled"` +- `WorkerError::Aborted(reason)` -- reason: フックが指定した理由 +- `WorkerError::Client(e)` -- reason: エラーの表示文字列 +- `WorkerError::Tool(e)` -- reason: エラーの表示文字列 +- `WorkerError::Hook(e)` -- reason: エラーの表示文字列 + +```rust +struct CleanupHook; + +#[async_trait] +impl Hook for CleanupHook { + async fn call(&self, reason: &mut String) -> Result<(), HookError> { + tracing::info!("Worker aborted: {}", reason); + Ok(()) + } +} +``` + +## resume() との関係 + +`resume()`はPause状態からの再開に使用される。内部では`run_turn_loop()`を呼び出し、保留中のツール呼び出し（historyに`FunctionCall`があるが対応する`FunctionCallOutput`がないもの）を検出して実行を再開する。 + +`resume()`中もキャンセルは同様に機能し、`finalize_interruption()`経由で`on_abort`フックが発火する。 + +## WorkerError の種別 + +| エラー種別 | 発生条件 | +| --------------------- | --------------------------------------------- | +| `Cancelled` | mpscチャネル経由のキャンセルシグナル受信 | +| `Aborted(String)` | フックによるAbort/Cancel、またはstream hookのPause | +| `Client(ClientError)` | LLMクライアントのエラー | +| `Tool(ToolError)` | ツール実行エラー | +| `Hook(HookError)` | フック実行中のエラー | +| `ConfigWarnings(Vec)` | サポートされていない設定オプション | diff --git a/crates/llm-worker/docs/spec/hooks.md b/crates/llm-worker/docs/spec/hooks.md new file mode 100644 index 00000000..48f20ec4 --- /dev/null +++ b/crates/llm-worker/docs/spec/hooks.md @@ -0,0 +1,556 @@ +# Hooks 仕様 + +## 概要 + +HookはWorker層でのターン制御に介入するためのメカニズムである。 + +メッセージ送信・ツール実行・ストリーム処理・ターン終了等の各ポイントで処理を差し込むことができる。 + +## コンセプト + +- **制御の介入**: ターンの進行、メッセージの内容、ツールの実行に対して介入 +- **Contextへのアクセス**: Item履歴を読み書き可能 +- **非破壊的チェーン**: 複数のHookを登録順に実行、後続Hookへの影響を制御 + +## Hook一覧 + +| Hook | タイミング | 主な用途 | 戻り値 | +| -------------------- | -------------------------- | -------------------------- | ----------------------- | +| `on_prompt_submit` | `run()` 呼び出し時 | ユーザーItemの前処理 | `OnPromptSubmitResult` | +| `pre_llm_request` | 各ターンのLLM送信前 | コンテキスト改変/検証 | `PreLlmRequestResult` | +| `pre_tool_call` | ツール実行前 | 実行許可/引数改変 | `PreToolCallResult` | +| `post_tool_call` | ツール実行後 | 結果加工/マスキング | `PostToolCallResult` | +| `on_stream_chunk` | ストリーム各イベント受信後 | 監査/低遅延介入 | `StreamHookResult` | +| `on_text_delta` | Text delta受信時 | 出力監視/中断 | `StreamHookResult` | +| `on_tool_call_delta` | Tool JSON delta受信時 | 引数監視/中断 | `StreamHookResult` | +| `on_stream_complete` | 1回のstream完了時 | サマリ/完了検証 | `StreamHookResult` | +| `on_turn_end` | ツールなしでターン終了直前 | 検証/リトライ指示 | `OnTurnEndResult` | +| `on_abort` | 中断時 | クリーンアップ/通知 | `()` | + +## Hook Trait + +```rust +#[async_trait] +pub trait Hook: Send + Sync { + async fn call(&self, input: &mut E::Input) -> Result; +} +``` + +## HookEventKind / 制御フロー型 + +Hookイベントごとに入力型(`Input`)と出力型(`Output`)を分離し、意味のない制御フローを排除する。 + +```rust +pub trait HookEventKind: Send + Sync + 'static { + type Input; + type Output; +} +``` + +### イベント種別一覧 + +```rust +pub struct OnPromptSubmit; // Input: Item, Output: OnPromptSubmitResult +pub struct PreLlmRequest; // Input: Vec, Output: PreLlmRequestResult +pub struct PreToolCall; // Input: ToolCallContext, Output: PreToolCallResult +pub struct PostToolCall; // Input: PostToolCallContext, Output: PostToolCallResult +pub struct OnTurnEnd; // Input: Vec, Output: OnTurnEndResult +pub struct OnAbort; // Input: String, Output: () +pub struct OnTextDelta; // Input: TextDeltaContext, Output: StreamHookResult +pub struct OnToolCallDelta; // Input: ToolCallDeltaContext, Output: StreamHookResult +pub struct OnStreamChunk; // Input: StreamChunkContext, Output: StreamHookResult +pub struct OnStreamComplete; // Input: StreamCompleteContext, Output: StreamHookResult +``` + +### 制御フロー Result 型 + +```rust +pub enum OnPromptSubmitResult { + Continue, + Cancel(String), +} + +pub enum PreLlmRequestResult { + Continue, + Cancel(String), +} + +pub enum PreToolCallResult { + Continue, + Skip, + Abort(String), + Pause, +} + +pub enum PostToolCallResult { + Continue, + Abort(String), +} + +pub enum OnTurnEndResult { + Finish, + ContinueWithMessages(Vec), + Paused, +} + +pub enum StreamHookResult { + Continue, + Abort(String), + Pause, +} +``` + +### コンテキスト型 + +#### ToolCallContext (PreToolCall用) + +```rust +pub struct ToolCallContext { + pub call: ToolCall, // ツール呼び出し情報（改変可能） + pub meta: ToolMeta, // 不変メタデータ + pub tool: Arc, // 状態アクセス用 +} +``` + +#### PostToolCallContext (PostToolCall用) + +```rust +pub struct PostToolCallContext { + pub call: ToolCall, // ツール呼び出し情報 + pub result: ToolResult, // 実行結果（改変可能） + pub meta: ToolMeta, // 不変メタデータ + pub tool: Arc, // 状態アクセス用 +} +``` + +#### TextDeltaContext (OnTextDelta用) + +```rust +pub struct TextDeltaContext { + pub index: usize, // ブロックインデックス + pub delta: String, // テキストデルタ内容 +} +``` + +#### ToolCallDeltaContext (OnToolCallDelta用) + +```rust +pub struct ToolCallDeltaContext { + pub index: usize, // ブロックインデックス + pub delta_json_fragment: String, // 部分JSONフラグメント +} +``` + +#### StreamChunkContext (OnStreamChunk用) + +```rust +pub struct StreamChunkContext { + pub event: crate::event::Event, // Worker層の公開イベント +} +``` + +#### StreamCompleteContext (OnStreamComplete用) + +```rust +pub struct StreamCompleteContext { + pub turn: usize, // 現在のターン番号 + pub event_count: usize, // このリクエストでのストリームイベント数 +} +``` + +## 呼び出しタイミング + +``` +Worker::run(user_input) +│ +├── on_prompt_submit ─────────────────────────┐ +│ ユーザーItemの前処理・検証 │ +│ （最初の1回のみ） │ +│ │ +└── loop { + │ + ├── pre_llm_request ──────────────────│ + │ history の clone に対して改変・検証 │ + │ （毎ターン実行） │ + │ │ + ├── LLMリクエスト送信 & ストリーム処理 │ + │ │ │ + │ ├── on_stream_chunk (各イベント) │ + │ ├── on_text_delta (テキストデルタ) │ + │ └── on_tool_call_delta (JSONデルタ) │ + │ │ + ├── on_stream_complete │ + │ │ + ├── ツール呼び出しがある場合: │ + │ │ │ + │ ├── pre_tool_call (各ツールごと・逐次) │ + │ │ 実行可否の判定、引数の改変 │ + │ │ │ + │ ├── ツール並列実行 (join_all) │ + │ │ │ + │ └── post_tool_call (各結果ごと・逐次) │ + │ 結果の確認、加工、ログ出力 │ + │ │ + ├── ツール結果をhistoryに追加 │ + │ → ループ先頭へ │ + │ │ + └── ツールなしの場合: │ + │ │ + └── on_turn_end ───────────────┘ + 最終応答のチェック（Lint/Fmt等） + エラーがあればContinueWithMessagesでリトライ +} + +※ 中断時は on_abort が呼ばれる（finalize_interruption経由） +``` + +## 各Hookの詳細 + +### on_prompt_submit + +**呼び出しタイミング**: `run()` でユーザーメッセージを受け取った直後（最初の1回のみ） + +**入力**: `&mut Item` - ユーザーItem（改変可能） + +**用途**: + +- ユーザー入力のバリデーション +- 入力のサニタイズ・フィルタリング +- ログ出力 +- `OnPromptSubmitResult::Cancel` による実行キャンセル（`WorkerError::Aborted`になる） + +**例**: 入力のバリデーション + +```rust +struct InputValidator; + +#[async_trait] +impl Hook for InputValidator { + async fn call( + &self, + item: &mut Item, + ) -> Result { + if let Item::Message { content, .. } = item { + if content.trim().is_empty() { + return Ok(OnPromptSubmitResult::Cancel("Empty input".to_string())); + } + } + Ok(OnPromptSubmitResult::Continue) + } +} +``` + +### pre_llm_request + +**呼び出しタイミング**: 各ターンのLLMリクエスト送信前（ループの毎回） + +**入力**: `&mut Vec` - historyのclone（改変可能、元のhistoryは変更されない） + +**用途**: + +- コンテキストへのシステムメッセージ注入 +- Itemのバリデーション +- 機密情報のフィルタリング +- リクエスト内容のログ出力 +- `PreLlmRequestResult::Cancel` による送信キャンセル（`WorkerError::Aborted`になる） + +### pre_tool_call + +**呼び出しタイミング**: 各ツール実行前（並列実行フェーズの前に逐次実行） + +**入力**: `&mut ToolCallContext`（`ToolCall` + `ToolMeta` + `Arc`） + +**用途**: + +- 危険なツールのブロック（`Skip`） +- 引数のサニタイズ（`context.call.input` の直接改変） +- 確認プロンプトの表示（UIとの連携） +- 実行ログの記録 +- `PreToolCallResult::Pause` による一時停止（`WorkerResult::Paused`を返す） +- `PreToolCallResult::Abort` による処理全体の中断 + +**備考**: 未登録ツール（ToolServerに存在しないツール名）の場合、Hookは適用されずそのまま実行される（実行時にエラーとなる）。 + +**例**: 特定ツールをブロック + +```rust +struct ToolBlocker { + blocked_tools: HashSet, +} + +#[async_trait] +impl Hook for ToolBlocker { + async fn call( + &self, + ctx: &mut ToolCallContext, + ) -> Result { + if self.blocked_tools.contains(&ctx.call.name) { + Ok(PreToolCallResult::Skip) + } else { + Ok(PreToolCallResult::Continue) + } + } +} +``` + +### post_tool_call + +**呼び出しタイミング**: 各ツール実行後（並列実行フェーズの後に逐次実行） + +**入力**: `&mut PostToolCallContext`（`ToolCall` + `ToolResult` + `ToolMeta` + `Arc`） + +**用途**: + +- 結果の加工・フォーマット（`context.result` の直接改変） +- 機密情報のマスキング +- 結果のキャッシュ +- 実行結果のログ出力 +- `PostToolCallResult::Abort` による処理全体の中断 + +**例**: 結果にプレフィックスを追加 + +```rust +struct ResultFormatter; + +#[async_trait] +impl Hook for ResultFormatter { + async fn call( + &self, + ctx: &mut PostToolCallContext, + ) -> Result { + if !ctx.result.is_error { + ctx.result.content = format!("[OK] {}", ctx.result.content); + } + Ok(PostToolCallResult::Continue) + } +} +``` + +### on_stream_chunk + +**呼び出しタイミング**: Timeline dispatchの後、各ストリームイベント受信時 + +**入力**: `&mut StreamChunkContext`（Worker層の`Event`を含む） + +**用途**: + +- 監査・ログ記録 +- 低遅延での介入 + +### on_text_delta + +**呼び出しタイミング**: `BlockDelta`イベントのうち`DeltaContent::Text`受信時 + +**入力**: `&mut TextDeltaContext`（`index` + `delta`） + +**用途**: + +- テキスト出力の監視 +- 特定パターン検出による中断 + +### on_tool_call_delta + +**呼び出しタイミング**: `BlockDelta`イベントのうち`DeltaContent::InputJson`受信時 + +**入力**: `&mut ToolCallDeltaContext`（`index` + `delta_json_fragment`） + +**用途**: + +- ツール引数JSONの部分監視 +- 危険な引数パターン検出による中断 + +### on_stream_complete + +**呼び出しタイミング**: 1回のストリームが完了した後（内側ループ終了後） + +**入力**: `&mut StreamCompleteContext`（`turn` + `event_count`） + +**用途**: + +- ストリーム完了の検証 +- イベント数の記録 + +### on_turn_end + +**呼び出しタイミング**: ツール呼び出しなしでターンが終了する直前 + +**入力**: `&mut Vec` - historyのclone（改変可能） + +**用途**: + +- 生成されたコードのLint/Fmt +- 出力形式のバリデーション +- 自己修正のためのリトライ指示（`ContinueWithMessages`でItemを追加して再ループ） +- 最終結果のログ出力 +- `OnTurnEndResult::Paused` による一時停止 + +**例**: JSON形式のバリデーション + +```rust +struct JsonValidator; + +#[async_trait] +impl Hook for JsonValidator { + async fn call( + &self, + items: &mut Vec, + ) -> Result { + // 最後のアシスタントメッセージを検査 + let last_text = items.iter().rev().find_map(|item| { + if let Item::Message { role, content, .. } = item { + if role == "assistant" { Some(content.as_str()) } else { None } + } else { None } + }); + + if let Some(text) = last_text { + if serde_json::from_str::(text).is_err() { + return Ok(OnTurnEndResult::ContinueWithMessages(vec![ + Item::user_message("Invalid JSON. Please fix and try again.") + ])); + } + } + Ok(OnTurnEndResult::Finish) + } +} +``` + +### on_abort + +**呼び出しタイミング**: Worker実行が中断された時（`finalize_interruption`経由） + +**入力**: `&mut String` - 中断理由 + +**用途**: + +- クリーンアップ処理 +- 中断理由のログ出力 +- 外部システムへの通知 + +**発火条件**: `run()`や`resume()`の結果が`Err`の場合に必ず発火する。具体的には以下のケース: + +- `WorkerError::Cancelled` -- reason: `"Cancelled"` +- `WorkerError::Aborted(reason)` -- reason: フックやキャンセルが指定した理由 +- その他のエラー（Client, Tool, Hook等） -- reason: エラーの表示文字列 + +## ストリームイベントの処理順序 + +`BlockDelta`到着時の順序: + +1. Timelineへdispatch（collector/subscriber更新） +2. `on_stream_chunk` +3. `on_text_delta` または `on_tool_call_delta`（デルタ種別による） + +※ `DeltaContent::Thinking` に対するHookは現在存在しない。 + +## 複数Hookの実行順序 + +Hookは**イベントごとに登録順**に実行される。 + +```rust +worker.add_pre_tool_call_hook(HookA); // 1番目に実行 +worker.add_pre_tool_call_hook(HookB); // 2番目に実行 +worker.add_pre_tool_call_hook(HookC); // 3番目に実行 +``` + +### 制御フローの伝播 + +- `Continue` / `Finish`: 後続Hookも実行 +- `Skip`: 現在の処理をスキップし、後続Hookは実行しない +- `Abort`: 即座に`WorkerError::Aborted`を返し、処理全体を中断 +- `Pause`: Workerを一時停止（`WorkerResult::Paused`を返す。`resume()`で再開可能） +- `Cancel`: `WorkerError::Aborted`を返し、処理を中断 + +``` +Hook A: Continue → Hook B: Skip → (Hook Cは実行されない) + ↓ + 処理をスキップ + +Hook A: Continue → Hook B: Abort("reason") + ↓ + WorkerError::Aborted + +Hook A: Continue → Hook B: Pause + ↓ + WorkerResult::Paused +``` + +注: stream系Hookの`Pause`は現状 `WorkerError::Aborted("Paused by stream hook")` として扱われる。 + +## Hook登録API + +Workerは各Hookイベントに対応する登録メソッドを提供する: + +```rust +worker.add_on_prompt_submit_hook(hook); +worker.add_pre_llm_request_hook(hook); +worker.add_pre_tool_call_hook(hook); +worker.add_post_tool_call_hook(hook); +worker.add_on_turn_end_hook(hook); +worker.add_on_abort_hook(hook); +worker.add_on_text_delta_hook(hook); +worker.add_on_tool_call_delta_hook(hook); +worker.add_on_stream_chunk_hook(hook); +worker.add_on_stream_complete_hook(hook); +``` + +これらのメソッドはWorkerの状態（`Mutable`/`CacheLocked`）に関係なく利用可能。 + +## HookRegistry + +全Hookは`HookRegistry`構造体で内部管理される。各フィールドは`Vec>>`であり、Workerが初期化時に空のレジストリを作成する。 + +## HookError + +```rust +pub enum HookError { + Aborted(String), // 処理の中断 + Internal(String), // 内部エラー +} +``` + +`HookError`は`WorkerError::Hook`に変換され、`finalize_interruption`で`on_abort`フックが発火する。 + +## 設計上のポイント + +### 1. イベントごとの実装 + +必要なイベントのみ `Hook` を実装する。1つの構造体で複数イベントの`Hook`を実装可能。 + +### 2. 可変参照による改変 + +`&mut`で引数を受け取るため、直接改変が可能。 + +```rust +async fn call(&self, ctx: &mut ToolCallContext) -> ... { + ctx.call.input["sanitized"] = json!(true); + Ok(PreToolCallResult::Continue) +} +``` + +### 3. 並列実行との統合 + +- `pre_tool_call`: 並列実行**前**に逐次実行（許可判定のため） +- ツール実行: `join_all`で**並列**実行（`tokio::select!`によりキャンセル可能） +- `post_tool_call`: 並列実行**後**に逐次実行（結果加工のため） + +### 4. Send + Sync 要件 + +`Hook`は`Send + Sync`を要求するため、スレッドセーフな実装が必要。 +状態を持つ場合は`Arc>`や`AtomicUsize`などを使用する。 + +### 5. pre_llm_request / on_turn_end のコンテキスト + +`pre_llm_request`と`on_turn_end`はhistoryの**clone**に対して操作する。Hookによる改変はリクエスト構築時またはリトライ用メッセージ追加時のみ反映され、Worker内部のhistoryは直接変更されない。 + +## 典型的なユースケース + +| ユースケース | 使用Hook | 処理内容 | +| ------------------ | ---------------------- | -------------------------- | +| ツール許可制御 | `pre_tool_call` | 危険なツールをSkip | +| 実行ログ | `pre/post_tool_call` | 呼び出しと結果を記録 | +| 出力バリデーション | `on_turn_end` | 形式チェック、リトライ指示 | +| コンテキスト注入 | `pre_llm_request` | システムメッセージ追加 | +| 結果のサニタイズ | `post_tool_call` | 機密情報のマスキング | +| レート制限 | `pre_tool_call` | 呼び出し頻度の制御 | +| ストリーム監視 | `on_text_delta` | 出力内容のリアルタイム監視 | +| 中断時通知 | `on_abort` | 外部システムへの通知 | diff --git a/crates/llm-worker/docs/spec/timeline.md b/crates/llm-worker/docs/spec/timeline.md new file mode 100644 index 00000000..d9764551 --- /dev/null +++ b/crates/llm-worker/docs/spec/timeline.md @@ -0,0 +1,497 @@ +# Timeline層設計 + +## 目的 + +- OpenAI / Anthropic / Gemini のストリーミングイベントを単一の抽象レイヤーに正規化し、LLMクライアントの状態遷移を制御する。 +- イベント単位処理（Meta系など）とブロック単位処理（テキスト/Thinking/ToolCall）を同一のパイプラインで扱えるようにする。 + +## 要件 + +イベントストリームに対して直接的にループ処理をしようとすると発生する煩雑な状態管理を避ける。 + +イベントをloop+matchで処理をするような手段を取ると、テキストのデルタの更新先や、完了タイミングなどの状態管理が必要になる。 +また、コンテンツブロックに対する単純なイテレータではping/usageなどの単発イベントを同期的に処理することができない。 + +- Meta系イベントの即時処理（ブロック内部イベントと順序が前後しないようにする） +- ブロック開始/差分/終了でスコープを保持する +- 型安全なハンドラー + - blockでキャッチするdeltaについて、Text/InputJson/Thinking等、ブロックに即したイベントの型が必要。 +- エラーの適切な制御 + +## Memo + +- ブロックを処理するHandlerで保持するコンテキストについて、LLMで用いられるコンテキストと混同を避けるために「スコープ」と呼称する。 +- ブロックは常に一つである前提。複数のブロックが同時に存在することは無いため。`Timeline`は`current_block: Option`で現在のブロックを追跡する。 + +## モジュール構成 + +``` +crates/llm-worker/src/ +├── handler.rs # Kind / Handler トレイト定義、Meta系・Block系Kind定義 +├── timeline/ +│ ├── mod.rs # 公開API re-export +│ ├── event.rs # Timeline層のイベント型（llm_client::eventからの変換含む） +│ ├── timeline.rs # Timeline本体、ErasedHandler、BlockHandlerWrapper群 +│ ├── text_block_collector.rs # TextBlockCollector（組み込みHandler） +│ └── tool_call_collector.rs # ToolCallCollector（組み込みHandler） +└── event.rs # Worker層の公開イベント型（timeline::eventからの変換含む） +``` + +イベント型は3層に存在する: + +1. `llm_client::event` - プロバイダ正規化済みイベント +2. `timeline::event` - Timeline内部イベント（`llm_client::event`からの`From`変換） +3. `crate::event`（Worker層公開） - 外部公開イベント（`timeline::event`からの`From`変換） + +## イベントモデル + +前提：`llm_client`層は各プロバイダのストリーミングレスポンスを正規化し、**フラットなEvent列挙**として出力する。Timeline層はそれを受け取り同一構造の`timeline::event::Event`に変換する。 + +```rust +// timeline::event::Event +pub enum Event { + // Meta events + Ping(PingEvent), + Usage(UsageEvent), + Status(StatusEvent), + Error(ErrorEvent), + + // Block lifecycle events + BlockStart(BlockStart), + BlockDelta(BlockDelta), + BlockStop(BlockStop), + BlockAbort(BlockAbort), +} +``` + +### メタイベント型 + +```rust +pub struct PingEvent { + pub timestamp: Option, +} + +pub struct UsageEvent { + pub input_tokens: Option, + pub output_tokens: Option, + pub total_tokens: Option, + pub cache_read_input_tokens: Option, + pub cache_creation_input_tokens: Option, +} + +pub struct StatusEvent { + pub status: ResponseStatus, +} + +pub enum ResponseStatus { + Started, + Completed, + Cancelled, + Failed, +} + +pub struct ErrorEvent { + pub code: Option, + pub message: String, +} +``` + +### ブロックイベント型 + +```rust +pub enum BlockType { + Text, + Thinking, + ToolUse, + ToolResult, +} + +pub struct BlockStart { + pub index: usize, + pub block_type: BlockType, + pub metadata: BlockMetadata, +} + +pub enum BlockMetadata { + Text, + Thinking, + ToolUse { id: String, name: String }, + ToolResult { tool_use_id: String }, +} + +pub struct BlockDelta { + pub index: usize, + pub delta: DeltaContent, +} + +pub enum DeltaContent { + Text(String), + Thinking(String), + InputJson(String), +} + +pub struct BlockStop { + pub index: usize, + pub block_type: BlockType, + pub stop_reason: Option, +} + +pub enum StopReason { + EndTurn, + MaxTokens, + StopSequence, + ToolUse, +} + +pub struct BlockAbort { + pub index: usize, + pub block_type: BlockType, + pub reason: String, +} +``` + +## TimelineとKind/Handler + +Timelineは`Event`ストリームを受け取り、登録された`Handler`にディスパッチする。 + +### Kind（`handler.rs`） + +`Kind`はイベント型のみを定義する。スコープはHandler側で定義するため、同じKindに対して異なるスコープを持つHandlerを登録できる。 + +```rust +pub trait Kind { + type Event; +} +``` + +### Handler（`handler.rs`） + +`Handler`はKindに対する処理を定義し、自身のスコープ型も決定する。 + +```rust +pub trait Handler { + type Scope: Default; + + fn on_event(&mut self, scope: &mut Self::Scope, event: &K::Event); +} +``` + +- `Kind`によって受け取るイベント型が決定される +- `Handler::Scope`によってHandler固有のスコープ型が決定される +- Meta系とBlock系を統一的に扱える + +### Meta系Kind + +スコープ不要の単発イベント。登録時にスコープが即座に開始される: + +```rust +pub struct UsageKind; +impl Kind for UsageKind { + type Event = UsageEvent; +} + +pub struct PingKind; +impl Kind for PingKind { + type Event = PingEvent; +} + +pub struct StatusKind; +impl Kind for StatusKind { + type Event = StatusEvent; +} + +pub struct ErrorKind; +impl Kind for ErrorKind { + type Event = ErrorEvent; +} +``` + +### Block系Kind + +ライフサイクル（Start/Delta/Stop）を持つ。スコープはHandler側で定義。Block系は3種定義されている: + +#### TextBlockKind + +```rust +pub struct TextBlockKind; +impl Kind for TextBlockKind { + type Event = TextBlockEvent; +} + +pub enum TextBlockEvent { + Start(TextBlockStart), + Delta(String), + Stop(TextBlockStop), +} + +pub struct TextBlockStart { + pub index: usize, +} + +pub struct TextBlockStop { + pub index: usize, + pub stop_reason: Option, +} +``` + +#### ThinkingBlockKind + +```rust +pub struct ThinkingBlockKind; +impl Kind for ThinkingBlockKind { + type Event = ThinkingBlockEvent; +} + +pub enum ThinkingBlockEvent { + Start(ThinkingBlockStart), + Delta(String), + Stop(ThinkingBlockStop), +} + +pub struct ThinkingBlockStart { + pub index: usize, +} + +pub struct ThinkingBlockStop { + pub index: usize, +} +``` + +#### ToolUseBlockKind + +```rust +pub struct ToolUseBlockKind; +impl Kind for ToolUseBlockKind { + type Event = ToolUseBlockEvent; +} + +pub enum ToolUseBlockEvent { + Start(ToolUseBlockStart), + InputJsonDelta(String), + Stop(ToolUseBlockStop), +} + +pub struct ToolUseBlockStart { + pub index: usize, + pub id: String, + pub name: String, +} + +pub struct ToolUseBlockStop { + pub index: usize, + pub id: String, + pub name: String, +} +``` + +### 使用例 + +```rust +// 使用例1: デルタを即時出力（スコープ不要） +struct PrintHandler; +impl Handler for PrintHandler { + type Scope = (); + + fn on_event(&mut self, _scope: &mut (), event: &TextBlockEvent) { + if let TextBlockEvent::Delta(s) = event { + print!("{}", s); + } + } +} + +// 使用例2: テキストを蓄積して収集（Stringをスコープとして利用） +struct TextCollector { results: Vec } +impl Handler for TextCollector { + type Scope = String; + + fn on_event(&mut self, buffer: &mut String, event: &TextBlockEvent) { + match event { + TextBlockEvent::Start(_) => {} + TextBlockEvent::Delta(s) => buffer.push_str(s), + TextBlockEvent::Stop(_) => { + self.results.push(std::mem::take(buffer)); + } + } + } +} +``` + +## Timelineの責務 + +1. `Event`ストリームを受信 +2. Meta系イベントを即座に該当Handlerへディスパッチ +3. Block系イベント（BlockStart/Delta/Stop/Abort）をBlockKindごとのライフサイクルイベントに変換 +4. 各Handlerごとのスコープの生成・管理（BlockStart時に生成、BlockStop/Abort時に破棄） +5. 登録されたHandlerへの登録順ディスパッチ +6. 暗黙的なスコープ開始（BlockStartなしにDeltaが到達した場合の対応） + +## Handlerの型消去（`timeline.rs`） + +### Meta系: `ErasedHandler` + +各Handlerは独自のScope型を持つため、Timelineで保持するには型消去が必要。`Send + Sync`境界付き: + +```rust +pub trait ErasedHandler: Send + Sync { + fn dispatch(&mut self, event: &K::Event); + fn start_scope(&mut self); + fn end_scope(&mut self); +} + +pub struct HandlerWrapper +where + H: Handler, + K: Kind, +{ + handler: H, + scope: Option, + _kind: PhantomData K>, // Send+Sync安全なPhantomData +} +``` + +Meta系Handlerは登録時に`start_scope()`が呼ばれ、スコープが常にアクティブな状態で保持される。 + +### Block系: `ErasedBlockHandler` + +Block系は`BlockStart`/`BlockDelta`/`BlockStop`/`BlockAbort`の4メソッドに分離されたtrait: + +```rust +trait ErasedBlockHandler: Send + Sync { + fn dispatch_start(&mut self, start: &BlockStart); + fn dispatch_delta(&mut self, delta: &BlockDelta); + fn dispatch_stop(&mut self, stop: &BlockStop); + fn dispatch_abort(&mut self, abort: &BlockAbort); + fn start_scope(&mut self); + fn end_scope(&mut self); + fn has_scope(&self) -> bool; +} +``` + +各BlockKindに対し専用のラッパー構造体が実装されている: + +- `TextBlockHandlerWrapper` - `DeltaContent::Text`のみをフィルタしてディスパッチ +- `ThinkingBlockHandlerWrapper` - `DeltaContent::Thinking`のみをフィルタしてディスパッチ +- `ToolUseBlockHandlerWrapper` - `DeltaContent::InputJson`をフィルタし、`BlockMetadata::ToolUse`からid/nameを追跡。Stopイベントにもid/nameを含めてディスパッチ + +## Timeline構造体 + +```rust +pub struct Timeline { + // Meta系ハンドラー + usage_handlers: Vec>>, + ping_handlers: Vec>>, + status_handlers: Vec>>, + error_handlers: Vec>>, + + // Block系ハンドラー（BlockTypeごとにグループ化） + text_block_handlers: Vec>, + thinking_block_handlers: Vec>, + tool_use_block_handlers: Vec>, + + // 現在アクティブなブロック + current_block: Option, +} +``` + +### ハンドラ登録メソッド + +| メソッド | Kind | 備考 | +|---|---|---| +| `on_usage()` | `UsageKind` | Meta系（登録時にスコープ開始） | +| `on_ping()` | `PingKind` | Meta系 | +| `on_status()` | `StatusKind` | Meta系 | +| `on_error()` | `ErrorKind` | Meta系 | +| `on_text_block()` | `TextBlockKind` | Block系 | +| `on_thinking_block()` | `ThinkingBlockKind` | Block系 | +| `on_tool_use_block()` | `ToolUseBlockKind` | Block系 | + +### ディスパッチ処理 + +```rust +impl Timeline { + pub fn dispatch(&mut self, event: &Event) { + match event { + // Meta系: 即時ディスパッチ（登録順） + Event::Usage(u) => self.dispatch_usage(u), + Event::Ping(p) => self.dispatch_ping(p), + Event::Status(s) => self.dispatch_status(s), + Event::Error(e) => self.dispatch_error(e), + + // Block系: スコープ管理しながらディスパッチ + Event::BlockStart(s) => self.handle_block_start(s), + Event::BlockDelta(d) => self.handle_block_delta(d), + Event::BlockStop(s) => self.handle_block_stop(s), + Event::BlockAbort(a) => self.handle_block_abort(a), + } + } +} +``` + +Block系のディスパッチフロー: + +- **BlockStart**: `current_block`を設定 → 該当BlockTypeのハンドラに対し`start_scope()` → `dispatch_start()` +- **BlockDelta**: `current_block`未設定時は暗黙的に設定。スコープ未開始のハンドラには暗黙的に`start_scope()`を呼ぶ → `dispatch_delta()` +- **BlockStop**: `dispatch_stop()` → `end_scope()` → `current_block`をクリア +- **BlockAbort**: `dispatch_abort()` → `end_scope()` → `current_block`をクリア + +`BlockType::ToolResult`は`text_block_handlers`にルーティングされる（テキストとして扱う）。 + +### ブロック中断 + +```rust +pub fn abort_current_block(&mut self) +``` + +キャンセルやエラー時に呼び出し、進行中のブロックに対して`BlockAbort`イベントを発火してスコープをクリーンアップする。 + +## 組み込みHandler + +### TextBlockCollector + +テキストブロックを収集する組み込みHandler。`Arc>>`を内部に持ち、`Clone`可能。 + +```rust +pub struct TextBlockCollector { + collected: Arc>>, +} + +impl Handler for TextBlockCollector { + type Scope = TextCollectorState; // buffer: String を保持 + // Start: バッファクリア + // Delta: バッファに追記 + // Stop: バッファの内容をcollectedに確定 +} +``` + +主なメソッド: `take_collected()`, `collected()`, `has_content()`, `clear()` + +### ToolCallCollector + +ToolUseブロックを収集する組み込みHandler。完了したツール呼び出しを`ToolCall`として収集する。 + +```rust +pub struct ToolCallCollector { + collected: Arc>>, +} + +impl Handler for ToolCallCollector { + type Scope = CollectorState; // current_id, current_name, input_json_buffer を保持 + // Start: id/nameを記録、バッファクリア + // InputJsonDelta: JSONバッファに追記 + // Stop: バッファをパースしToolCallを確定 +} +``` + +主なメソッド: `take_collected()`, `collected()`, `has_pending_calls()`, `clear()` + +Workerは内部で`TextBlockCollector`と`ToolCallCollector`を自動的にTimelineに登録して使用する。 + +## 期待効果 + +- **統一インターフェース**: Meta系もBlock系も`Handler`で統一 +- **型安全**: `Kind`によってイベント型がコンパイル時に決定、`Handler::Scope`によってHandler固有のスコープ型が決定 +- **スコープの柔軟性**: 同一Kindに対して異なるScopeを持つ複数Handlerを登録可能 +- **責務分離**: `llm_client`層はフラットなEvent出力、Timeline層がブロック構造化、Worker層が公開イベントへの変換 +- **スコープ管理の自動化**: Handlerは自前でスコープ保持を意識せずに済む +- **暗黙的スコープ開始**: BlockStartを送らないプロバイダ（OpenAI等）にも対応 +- **Send + Sync**: 全ハンドラが`Send + Sync`境界を持ち、非同期コンテキストで安全に使用可能 diff --git a/crates/llm-worker/docs/spec/tools.md b/crates/llm-worker/docs/spec/tools.md new file mode 100644 index 00000000..4bcb63b7 --- /dev/null +++ b/crates/llm-worker/docs/spec/tools.md @@ -0,0 +1,390 @@ +# Tool 設計 + +## 概要 + +`llm-worker`のツールシステムは、LLMが外部リソースにアクセスしたり計算を実行するための仕組みを提供する。 +メタ情報の不変性とセッションスコープの状態管理を両立させる設計となっている。 + +ツールの管理は`ToolServer` / `ToolServerHandle`に分離されており、 +Workerから独立したツール管理・実行が可能。 + +## 主要な型 + +``` +type ToolDefinition + Fn() -> (ToolMeta, Arc) + +Worker.register_tool() → ToolServerHandle.register_tool() で呼び出し + + | + v + +- struct ToolMeta (name, desc, schema) + 不変・登録時固定 +- trait Tool (execute) + 登録時生成・セッション中再利用 +- struct ToolServer / ToolServerHandle + ツールの管理・実行を担当 +``` + +### ToolMeta + +ツールのメタ情報を保持する不変構造体。登録時に固定され、Worker内で変更されない。 + +```rust +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ToolMeta { + pub name: String, + pub description: String, + pub input_schema: Value, +} + +impl ToolMeta { + pub fn new(name: impl Into) -> Self; + pub fn description(mut self, desc: impl Into) -> Self; + pub fn input_schema(mut self, schema: Value) -> Self; +} +``` + +**目的:** + +- LLM へのツール定義として送信 +- Hook からの参照（読み取り専用） +- 登録後の不変性を保証 + +### Tool trait + +ツールの実行ロジックのみを定義するトレイト。 + +```rust +#[async_trait] +pub trait Tool: Send + Sync { + async fn execute(&self, input_json: &str) -> Result; +} +``` + +**設計方針:** + +- メタ情報（name, description, schema）は含まない +- 状態を持つことが可能（セッション中のカウンターなど） +- `Send + Sync` で並列実行に対応 + +**インスタンスのライフサイクル:** + +1. `register_tool()` 呼び出し時にファクトリが実行され、インスタンスが生成される +2. LLM がツールを呼び出すと、既存インスタンスの `execute()` が実行される +3. 同じセッション中は同一インスタンスが再利用される + +※ 「最初に呼ばれたとき」の遅延初期化ではなく、**登録時の即時初期化**である。 + +### ToolDefinition + +メタ情報とツールインスタンスを生成するファクトリ。 + +```rust +pub type ToolDefinition = Arc (ToolMeta, Arc) + Send + Sync>; +``` + +**なぜファクトリか:** + +- Worker への登録時に一度だけ呼び出される +- メタ情報とインスタンスを同時に生成し、整合性を保証 +- クロージャでコンテキスト（`self.clone()`）をキャプチャ可能 + +### ToolError + +```rust +#[derive(Debug, Error)] +pub enum ToolError { + #[error("Invalid argument: {0}")] + InvalidArgument(String), + #[error("Execution failed: {0}")] + ExecutionFailed(String), + #[error("Internal error: {0}")] + Internal(String), +} +``` + +## ToolServer / ToolServerHandle + +ツール管理がWorkerから分離された専用コンポーネント。 +`ToolServer`がツールマップを所有し、`ToolServerHandle`がそのクローン可能な参照を提供する。 + +### ToolServer + +```rust +#[derive(Clone, Default)] +pub struct ToolServer { + tools: Arc>, // HashMap)> +} + +impl ToolServer { + pub fn new() -> Self; + pub fn handle(&self) -> ToolServerHandle; +} +``` + +### ToolServerHandle + +```rust +#[derive(Clone, Default)] +pub struct ToolServerHandle { + tools: Arc>, +} + +impl ToolServerHandle { + // 登録（pub(crate) - Worker経由でのみ呼び出し可能） + pub(crate) fn register_tool(&self, factory: WorkerToolDefinition) -> Result<(), ToolServerError>; + pub(crate) fn register_tools(&self, factories: impl IntoIterator) -> Result<(), ToolServerError>; + + // 検索（Hookからのアクセス用） + pub fn get_tool(&self, name: &str) -> Option<(ToolMeta, Arc)>; + + // 実行 + pub async fn call_tool(&self, name: &str, input_json: &str) -> Result; + + // LLMリクエスト用定義生成（名前順ソート済み） + pub fn tool_definitions_sorted(&self) -> Vec; +} +``` + +### ToolServerError + +```rust +#[derive(Debug, Error, PartialEq, Eq)] +pub enum ToolServerError { + #[error("Tool with name '{0}' already registered")] + DuplicateName(String), + #[error("Tool '{0}' not found")] + ToolNotFound(String), + #[error("Tool execution failed: {0}")] + ToolExecution(String), +} +``` + +### 設計上のポイント + +- `ToolServer`は`Clone`可能で、`Arc>`により複数箇所で共有可能 +- `register_tool` / `register_tools` は `pub(crate)` でWorker経由のみ +- `get_tool`, `call_tool` は `pub` でHookやアプリケーションから直接利用可能 +- `tool_definitions_sorted()` は名前でソートされた決定的な定義リストを返す +- Worker側では `tool_server_handle()` で `ToolServerHandle` を取得可能 + +## Worker でのツール管理 + +Worker は `ToolServerHandle` を内部に持ち、ツール登録のAPIを提供する。 +登録は `Mutable` 状態でのみ可能。 + +```rust +// Worker API (Mutable状態のみ) +pub fn register_tool(&mut self, factory: WorkerToolDefinition) -> Result<(), ToolRegistryError>; +pub fn register_tools(&mut self, factories: impl IntoIterator) -> Result<(), ToolRegistryError>; + +// ハンドルの取得（全状態） +pub fn tool_server_handle(&self) -> ToolServerHandle; +``` + +登録時の処理: + +1. ファクトリを呼び出し `(meta, instance)` を取得 +2. 同名ツールが既に登録されていればエラー（`ToolRegistryError::DuplicateName`） +3. HashMap に `(meta, instance)` を保存 + +## マクロによる自動生成 (`llm-worker-macros`) + +`#[tool_registry]` マクロと `#[tool]` マクロにより、メソッドから自動的に +`Tool` トレイト実装と `ToolDefinition` ファクトリを生成する。 + +### マクロ一覧 + +| マクロ | 種類 | 用途 | +| --- | --- | --- | +| `#[tool_registry]` | proc_macro_attribute | `impl`ブロックに適用。`#[tool]`メソッドを検出し、コード生成 | +| `#[tool]` | proc_macro_attribute | メソッドに適用。マーカー属性（`tool_registry`が処理） | +| `#[description]` | proc_macro_attribute | 引数に適用。schemarsの`description`に変換 | + +### 使用例 + +```rust +#[tool_registry] +impl MyApp { + /// 検索を実行する + /// 指定されたクエリでデータベースを検索します。 + #[tool] + async fn search( + &self, + #[description = "検索クエリ文字列"] query: String, + ) -> String { + format!("Results for: {}", query) + } +} + +// 生成されるもの: +// - SearchArgs 構造体 +// - ToolSearch 構造体 (Tool trait実装) +// - MyApp::search_definition() メソッド +``` + +### 生成されるコード詳細 + +マクロは以下を自動生成します。 + +**1. 引数構造体（`{PascalCase}Args`）** + +```rust +#[derive(serde::Deserialize, schemars::JsonSchema)] +struct SearchArgs { + #[schemars(description = "検索クエリ文字列")] + pub query: String, +} +``` + +- `#[description = "..."]` 属性は `#[schemars(description = "...")]` に変換される +- 引数がない場合は空の構造体が生成される + +**2. ラッパー構造体（`Tool{PascalCase}`）** + +```rust +#[derive(Clone)] +pub struct ToolSearch { + ctx: MyApp, +} + +#[async_trait] +impl Tool for ToolSearch { + async fn execute(&self, input_json: &str) -> Result { + let args: SearchArgs = serde_json::from_str(input_json) + .map_err(|e| ToolError::InvalidArgument(e.to_string()))?; + let result = self.ctx.search(args.query).await; + Ok(format!("{:?}", result)) + } +} +``` + +- 戻り値が `Result` の場合、`Err` は `ToolError::ExecutionFailed` に変換される +- 戻り値が `Result` でない場合、`format!("{:?}", result)` で文字列化される +- async/非asyncの両方をサポート + +**3. ファクトリメソッド（`{method_name}_definition`）** + +```rust +impl MyApp { + pub fn search_definition(&self) -> ToolDefinition { + let ctx = self.clone(); + Arc::new(move || { + let schema = schemars::schema_for!(SearchArgs); + let meta = ToolMeta::new("search") + .description("検索を実行する\n指定されたクエリでデータベースを検索します。") + .input_schema(serde_json::to_value(schema).unwrap_or(json!({}))); + let tool: Arc = Arc::new(ToolSearch { ctx: ctx.clone() }); + (meta, tool) + }) + } +} +``` + +- ツール名はメソッド名（snake_case）がそのまま使われる +- ツールの説明はメソッドのdocコメント全体が使われる（docコメントがない場合は `"Tool: {name}"` がフォールバック） +- `input_schema` は `schemars::schema_for!` で生成される + +### 命名規則 + +| 元のメソッド名 | 引数構造体 | ラッパー構造体 | ファクトリメソッド | +| --- | --- | --- | --- | +| `search` | `SearchArgs` | `ToolSearch` | `search_definition()` | +| `get_user` | `GetUserArgs` | `ToolGetUser` | `get_user_definition()` | + +PascalCaseへの変換は `to_pascal_case()` 関数で行われる（snake_caseの各セグメントを先頭大文字に変換）。 + +## Hook との連携 + +Hook は `ToolCallContext` / `PostToolCallContext` +を通じてメタ情報とインスタンスにアクセスできる。 + +```rust +pub struct ToolCallContext { + pub call: ToolCall, // 呼び出し情報（改変可能） + pub meta: ToolMeta, // メタ情報（読み取り専用） + pub tool: Arc, // インスタンス（状態アクセス用） +} + +pub struct PostToolCallContext { + pub call: ToolCall, // 呼び出し情報 + pub result: ToolResult, // 実行結果（改変可能） + pub meta: ToolMeta, // メタ情報（読み取り専用） + pub tool: Arc, // インスタンス（状態アクセス用） +} +``` + +**用途:** + +- `meta` で名前やスキーマを確認 +- `tool` でツールの内部状態を読み取り（ダウンキャスト必要） +- `call` の引数を改変してツールに渡す（PreToolCall） +- `result` の内容を改変（PostToolCall） + +## 使用例 + +### 手動実装 + +```rust +struct Counter { count: AtomicUsize } + +#[async_trait] +impl Tool for Counter { + async fn execute(&self, _: &str) -> Result { + let n = self.count.fetch_add(1, Ordering::SeqCst); + Ok(format!("count: {}", n)) + } +} + +let def: ToolDefinition = Arc::new(|| { + let meta = ToolMeta::new("counter") + .description("カウンターを増加") + .input_schema(json!({"type": "object"})); + (meta, Arc::new(Counter { count: AtomicUsize::new(0) })) +}); + +worker.register_tool(def)?; +``` + +### マクロ使用（推奨） + +```rust +#[derive(Clone)] +struct App; + +#[tool_registry] +impl App { + /// 挨拶する + #[tool] + async fn greet(&self, #[description = "名前"] name: String) -> String { + format!("Hello, {}!", name) + } +} + +let app = App; +worker.register_tool(app.greet_definition())?; +``` + +### 複数ツールの一括登録 + +```rust +worker.register_tools([ + app.search_definition(), + app.get_user_definition(), + app.greet_definition(), +])?; +``` + +## 設計上の決定 + +| 問題 | 決定 | 理由 | +| --- | --- | --- | +| メタ情報の変更可能性 | ToolMeta を分離・不変化 | 登録後の整合性を保証 | +| 状態管理 | 登録時にインスタンス生成 | セッション中の状態保持、同一インスタンス再利用 | +| Factory vs Instance | Factory + 登録時即時呼び出し | コンテキストキャプチャと登録時検証 | +| Hook からのアクセス | Context に meta と tool を含む | 柔軟な介入を可能に | +| ツール管理の分離 | ToolServer / ToolServerHandle | Worker外からのツール検索・実行を可能に | +| 登録の可視性 | `register_tool` は `pub(crate)` | Worker経由でのみ登録を許可し、整合性を維持 | +| 定義リストの順序 | 名前順ソート | 決定的なリクエスト生成を保証 | diff --git a/crates/llm-worker/docs/spec/worker.md b/crates/llm-worker/docs/spec/worker.md new file mode 100644 index 00000000..bab30ae1 --- /dev/null +++ b/crates/llm-worker/docs/spec/worker.md @@ -0,0 +1,460 @@ +# Worker 設計 + +## 概要 + +`Worker`はアプリケーションの「ターン」を制御する高レベルコンポーネントです。 +`LlmClient`と`Timeline`を内包し、ユーザー定義の`Tool`と`Hook`を用いて自律的なインタラクションを行います。 + +Type-stateパターンにより、`Mutable`（編集可能）と`CacheLocked`（キャッシュ保護）の2つの状態を持ちます。 + +## アーキテクチャ + +```mermaid +graph TD + User[Application / User] -->|1. Run| Worker + Worker -->|2. Event Loop| Timeline + Timeline -->|3. Dispatch| Handler[Handlers (TextBlockCollector, ToolCallCollector)] + + subgraph "Worker Layer" + Worker + Hook[HookRegistry] + ToolServer[ToolServerHandle] + end + + subgraph "Core Layer" + Timeline + LlmClient + end + + Worker -.->|Intervene| Hook + ToolServer -.->|Execute| Tool[User Defined Tools] + Worker -.->|Subscribe| Subscriber[WorkerSubscriber] +``` + +## Worker 構造体 + +```rust +pub struct Worker { + client: C, + timeline: Timeline, + text_block_collector: TextBlockCollector, + tool_call_collector: ToolCallCollector, + tool_server: ToolServerHandle, + hooks: HookRegistry, + system_prompt: Option, + history: Vec, + locked_prefix_len: usize, + turn_count: usize, + turn_notifiers: Vec>, + request_config: RequestConfig, + last_run_interrupted: bool, + cancel_tx: mpsc::Sender<()>, + cancel_rx: mpsc::Receiver<()>, + _state: PhantomData, +} +``` + +## 状態遷移 (Type-state) + +| 状態 | 説明 | 利用可能な操作 | +| --- | --- | --- | +| `Mutable` | 初期状態。自由に編集可能 | system_prompt設定、history編集、tool/hook登録、lock() | +| `CacheLocked` | キャッシュ保護状態 | historyへの追記のみ、unlock() | + +```rust +// Mutable → CacheLocked +let locked = worker.lock(); + +// CacheLocked → Mutable +let mutable = locked.unlock(); +``` + +`CacheLocked`状態ではLLM APIのKVキャッシュヒットを確保するため、 +システムプロンプトと既存の履歴（ロック時点のプレフィックス）は不変となります。 + +## ライフサイクル (ターン制御) + +Workerは以下のループ（ターン）を実行します。 + +1. **Start Turn**: `Worker::run(user_input)` 呼び出し +2. **Hook: OnPromptSubmit**: + - ユーザーメッセージ(`Item`)の改変、バリデーション、キャンセルが可能。 + - `Cancel` を返すと `WorkerError::Aborted` で終了。 +3. **Resume Check**: + - 未応答のToolCall（Pauseから復帰した場合等）があれば、先にツール実行を行う。 +4. **Turn Loop**: + 1. **Hook: PreLlmRequest**: + - LLMリクエスト送信前にコンテキスト（`Vec`）を改変可能。 + - `Cancel` を返すとターン中断。 + 2. **Request & Stream**: + - LLMへリクエスト送信。イベントストリーム開始。 + - `Timeline`によるイベント処理（`TextBlockCollector`, `ToolCallCollector`）。 + - ストリーム中に **Hook: OnStreamChunk**, **OnTextDelta**, **OnToolCallDelta** を実行。 + - ストリーム完了時に **Hook: OnStreamComplete** を実行。 + 3. **Tool Handling (Parallel)**: + - レスポンス内に含まれる全てのTool Callを収集。 + - 各Toolに対して **Hook: PreToolCall** を実行（実行可否、引数改変、Pause）。 + - 許可されたToolを**並列実行 (`join_all`)**（`ToolServerHandle`経由）。 + - 各Tool実行後に **Hook: PostToolCall** を実行（結果の確認、加工）。 + 4. **Next Request Decision**: + - Tool実行結果がある場合 -> 結果を`Item::FunctionCallOutput`としてhistoryに追加し、**Step 4.1へ戻る**（自動ループ）。 + - Tool実行がない場合 -> Step 5へ。 +5. **Hook: OnTurnEnd**: + - 最終的な応答に対するチェック（Lint/Fmtなど）。 + - `ContinueWithMessages(items)`: 追加メッセージをhistoryに追加して**Step 4.1へ戻る**ことで自己修正を促せる。 + - `Paused`: `WorkerResult::Paused` を返し、後で `resume()` で再開可能。 + - `Finish`: ターン正常終了。 + +## キャンセル機構 + +`cancel_tx` / `cancel_rx` チャネルによる非同期キャンセルをサポートします。 + +```rust +// キャンセルトリガーの取得 +let sender = worker.cancel_sender(); + +// 別タスクからキャンセル +sender.try_send(()).ok(); + +// または直接 +worker.cancel(); +``` + +キャンセルはストリーム開始前、ストリーム受信中、ツール実行中のいずれのタイミングでも +`tokio::select!` により検出され、`WorkerError::Cancelled` を返します。 + +## 実行結果 + +```rust +pub enum WorkerResult { + /// 完了（ユーザー入力待ち） + Finished, + /// 一時停止（resume()で再開可能） + Paused, +} +``` + +`resume()` メソッドにより、Paused状態からユーザーメッセージを追加せずにターン処理を再開できます。 + +## Hook 設計 + +### Hook Trait + +```rust +#[async_trait] +pub trait Hook: Send + Sync { + async fn call(&self, input: &mut E::Input) -> Result; +} + +pub trait HookEventKind: Send + Sync + 'static { + type Input; + type Output; +} +``` + +### Hook イベント一覧 + +| Hook | Input | Output | タイミング | +| --- | --- | --- | --- | +| `OnPromptSubmit` | `Item` | `OnPromptSubmitResult` | `run()` 直後、ユーザーメッセージ受信時 | +| `PreLlmRequest` | `Vec` | `PreLlmRequestResult` | 各ターンのLLMリクエスト送信前 | +| `PreToolCall` | `ToolCallContext` | `PreToolCallResult` | 各ツール実行前 | +| `PostToolCall` | `PostToolCallContext` | `PostToolCallResult` | 各ツール実行後 | +| `OnTurnEnd` | `Vec` | `OnTurnEndResult` | ツール呼び出しなしでターン終了時 | +| `OnAbort` | `String` | `()` | エラーまたはキャンセルで中断時 | +| `OnTextDelta` | `TextDeltaContext` | `StreamHookResult` | テキストデルタ受信時 | +| `OnToolCallDelta` | `ToolCallDeltaContext` | `StreamHookResult` | ツール呼び出しJSON断片受信時 | +| `OnStreamChunk` | `StreamChunkContext` | `StreamHookResult` | ストリームイベント受信時 | +| `OnStreamComplete` | `StreamCompleteContext` | `StreamHookResult` | ストリーム完了時 | + +### Hook 結果型 + +```rust +pub enum OnPromptSubmitResult { + Continue, + Cancel(String), +} + +pub enum PreLlmRequestResult { + Continue, + Cancel(String), +} + +pub enum PreToolCallResult { + Continue, + Skip, // ツール実行をスキップ + Abort(String), // 処理中断 + Pause, // 一時停止（resume()で再開） +} + +pub enum PostToolCallResult { + Continue, + Abort(String), +} + +pub enum OnTurnEndResult { + Finish, + ContinueWithMessages(Vec), // メッセージを追加してターン継続 + Paused, +} + +pub enum StreamHookResult { + Continue, + Abort(String), + Pause, +} +``` + +### Tool Call Context + +`PreToolCall` / `PostToolCall` は、ツール実行の文脈を含むコンテキストを受け取ります。 + +```rust +pub struct ToolCallContext { + pub call: ToolCall, // 呼び出し情報（改変可能） + pub meta: ToolMeta, // メタ情報（読み取り専用） + pub tool: Arc, // インスタンス（状態アクセス用） +} + +pub struct PostToolCallContext { + pub call: ToolCall, // 呼び出し情報 + pub result: ToolResult, // 実行結果（改変可能） + pub meta: ToolMeta, // メタ情報（読み取り専用） + pub tool: Arc, // インスタンス（状態アクセス用） +} +``` + +### ストリーミングHookのコンテキスト + +```rust +pub struct TextDeltaContext { + pub index: usize, + pub delta: String, +} + +pub struct ToolCallDeltaContext { + pub index: usize, + pub delta_json_fragment: String, +} + +pub struct StreamChunkContext { + pub event: crate::event::Event, +} + +pub struct StreamCompleteContext { + pub turn: usize, + pub event_count: usize, +} +``` + +### Hook 登録API + +```rust +worker.add_on_prompt_submit_hook(my_hook); +worker.add_pre_llm_request_hook(my_hook); +worker.add_pre_tool_call_hook(my_hook); +worker.add_post_tool_call_hook(my_hook); +worker.add_on_turn_end_hook(my_hook); +worker.add_on_abort_hook(my_hook); +worker.add_on_text_delta_hook(my_hook); +worker.add_on_tool_call_delta_hook(my_hook); +worker.add_on_stream_chunk_hook(my_hook); +worker.add_on_stream_complete_hook(my_hook); +``` + +## Worker Event API (Subscriber) + +### 背景と目的 + +Workerは内部でイベントを処理し結果を返しますが、UIへのストリーミング表示やリアルタイムフィードバックには、イベントを外部に公開する仕組みが必要です。 + +### WorkerSubscriber Trait + +`WorkerSubscriber`トレイトを実装し、`worker.subscribe()` で一括登録します。 + +```rust +pub trait WorkerSubscriber: Send { + // スコープ型（ブロックイベント用） + type TextBlockScope: Default + Send + Sync; + type ToolUseBlockScope: Default + Send + Sync; + + // === ブロックイベント（スコープ管理あり）=== + fn on_text_block( + &mut self, + scope: &mut Self::TextBlockScope, + event: &TextBlockEvent, + ) {} + + fn on_tool_use_block( + &mut self, + scope: &mut Self::ToolUseBlockScope, + event: &ToolUseBlockEvent, + ) {} + + // === 単発イベント === + fn on_usage(&mut self, event: &UsageEvent) {} + fn on_status(&mut self, event: &StatusEvent) {} + fn on_error(&mut self, event: &ErrorEvent) {} + + // === 累積イベント（Worker層で追加）=== + fn on_text_complete(&mut self, text: &str) {} + fn on_tool_call_complete(&mut self, call: &ToolCall) {} + + // === ターン制御 === + fn on_turn_start(&mut self, turn: usize) {} + fn on_turn_end(&mut self, turn: usize) {} +} +``` + +### 内部実装 + +SubscriberはWorker内部でAdapter群に分解され、各KindのHandlerとしてTimelineに登録されます。 +累積イベント（`on_text_complete`, `on_tool_call_complete`）はAdapter内でブロック終了時にバッファから合成されます。 + +- `TextBlockSubscriberAdapter`: テキストデルタを蓄積し、Stop時に`on_text_complete`を呼ぶ +- `ToolUseBlockSubscriberAdapter`: ID・名前・JSONを蓄積し、Stop時に`on_tool_call_complete`を呼ぶ +- `UsageSubscriberAdapter`, `StatusSubscriberAdapter`, `ErrorSubscriberAdapter`: 単純な転送 +- `SubscriberTurnNotifier`: ターン開始・終了の通知 + +### 使用例 + +```rust +struct StreamPrinter; + +impl WorkerSubscriber for StreamPrinter { + type TextBlockScope = (); + type ToolUseBlockScope = (); + + fn on_text_block(&mut self, _: &mut (), event: &TextBlockEvent) { + if let TextBlockEvent::Delta(text) = event { + print!("{}", text); + } + } + + fn on_text_complete(&mut self, text: &str) { + println!("\n--- Complete: {} chars ---", text.len()); + } +} + +worker.subscribe(StreamPrinter); +let result = worker.run("Hello!").await?; +``` + +## Worker API + +### 生成と設定 (Mutable状態のみ) + +```rust +// 生成（ビルダーパターン） +let worker = Worker::new(client) + .system_prompt("You are a helpful assistant.") + .max_tokens(4096) + .temperature(0.7) + .top_p(0.9) + .top_k(40) + .stop_sequence("\n\n") + .with_config(request_config) + .with_item(item) + .with_items(items); + +// バリデーション（プロバイダの対応確認） +let worker = worker.validate()?; + +// ミューテーション +worker.set_system_prompt("..."); +worker.set_max_tokens(4096); +worker.set_temperature(0.7); +worker.set_top_p(0.9); +worker.set_top_k(40); +worker.add_stop_sequence("\n\n"); +worker.clear_stop_sequences(); +worker.set_request_config(config); + +// 履歴操作 +worker.push_item(item); +worker.extend_history(items); +worker.set_history(items); +worker.clear_history(); +worker.history_mut(); // &mut Vec + +// ツール登録 +worker.register_tool(factory)?; +worker.register_tools(factories)?; +``` + +### 全状態で利用可能 + +```rust +// 実行 +let result = worker.run("user input").await?; +let result = worker.resume().await?; + +// キャンセル +worker.cancel(); +let sender = worker.cancel_sender(); + +// 参照 +worker.history(); // &[Item] +worker.get_system_prompt(); // Option<&str> +worker.turn_count(); // usize +worker.request_config(); // &RequestConfig +worker.last_run_interrupted(); // bool +worker.is_cancelled(); // bool + +// ツールサーバー +worker.tool_server_handle(); // ToolServerHandle + +// Timeline直接アクセス +worker.timeline_mut(); // &mut Timeline + +// イベント購読 +worker.subscribe(my_subscriber); + +// Hook登録 +worker.add_on_prompt_submit_hook(hook); +worker.add_pre_llm_request_hook(hook); +worker.add_pre_tool_call_hook(hook); +worker.add_post_tool_call_hook(hook); +worker.add_on_turn_end_hook(hook); +worker.add_on_abort_hook(hook); +worker.add_on_text_delta_hook(hook); +worker.add_on_tool_call_delta_hook(hook); +worker.add_on_stream_chunk_hook(hook); +worker.add_on_stream_complete_hook(hook); +``` + +## エラー型 + +```rust +pub enum WorkerError { + Client(ClientError), + Tool(ToolError), + Hook(HookError), + Aborted(String), + Cancelled, + ConfigWarnings(Vec), +} + +pub enum ToolRegistryError { + DuplicateName(String), +} +``` + +## 公開エクスポート + +`lib.rs` から以下がre-exportされています。 + +```rust +pub use message::{ContentPart, Item, Message, Role}; +pub use worker::{ToolRegistryError, Worker, WorkerConfig, WorkerError, WorkerResult}; +``` + +モジュール: +- `pub mod event` +- `pub mod hook` +- `pub mod llm_client` +- `pub mod state` +- `pub mod subscriber` +- `pub mod timeline` +- `pub mod tool` +- `pub mod tool_server` diff --git a/docs/llm_client_reqs.md b/docs/llm_client_reqs.md new file mode 100644 index 00000000..fec2d913 --- /dev/null +++ b/docs/llm_client_reqs.md @@ -0,0 +1,17 @@ +# insomniaが求めるllmクライアントライブラリ + +- 前提： + a. userメッセージを追加しなくてもagentの途中ママ投げれば、AIはそれを自身の生成途中と認識して普通に継続する。 + b. KVキャッシュは速度・効率の面で有利で、基本的にコンテキストを重ねた後での事後的なコンテキストの改変はKVキャッシュヒット率を大幅に下げる + c. ツール・フックの基本的なスキーマ自動化を提供する + +以上を前提とし、私が求める性能： +- メッセージの送信と生成のResume、一時停止/再開が必要。 +- 暗黙的にキャッシュを保証し、キャッシュを破壊しうる操作を明示的にブロックせずとも、いつの間にかキャッシュ破壊してた状態にはしたくない。 + +## 上層の要件 + +おそらく最下層の抽象化レイヤーではなく、その上でやるべき事 + +- Hooksの実装 + - 送信コンテンツの操作やエージェントの生成にしたがって発生するイベントを適切に処理できる仕組み