# 機能設計書 92-機械学習モデル

## 概要

本ドキュメントは、Apache Flink の機械学習モデルサポート機能について設計内容を記載する。flink-models モジュールは、Flink の Table API/SQL から外部の機械学習モデル（現時点では OpenAI API）を呼び出すための基盤を提供する。

### 本機能の処理概要

flink-models は、Flink のストリーム/バッチ処理パイプライン内から外部の機械学習モデルを呼び出し、推論結果を取得する機能を提供する。現在のバージョンでは OpenAI API への接続をサポートし、チャット補完（Chat Completion）と埋め込み（Embedding）の 2 種類のタスクに対応している。AsyncPredictFunction を使用した非同期処理により、高スループットな推論処理を実現する。

**業務上の目的・背景**：近年の AI/ML の発展により、ストリーム処理パイプライン内でリアルタイムに機械学習モデルを活用するニーズが高まっている。外部 LLM サービス（OpenAI 等）との連携により、自然言語処理、テキスト分類、埋め込みベクトル生成などをストリーミングデータに対してリアルタイムに適用できる。

**機能の利用シーン**：
- ストリーミングデータに対するリアルタイムテキスト分類
- 顧客問い合わせの自動応答生成
- ドキュメントの埋め込みベクトル生成（セマンティック検索用）
- センチメント分析、エンティティ抽出
- Table API/SQL からの宣言的な ML モデル呼び出し

**主要な処理内容**：
1. ModelProviderFactory による OpenAI クライアントの初期化
2. 入力データのトークン数制限チェックとコンテキストオーバーフロー処理
3. OpenAI API への非同期リクエスト送信
4. レスポンスの RowData 形式への変換
5. エラーハンドリング（リトライ、フェイルオーバー、無視）

**関連システム・外部連携**：
- OpenAI API（Chat Completions、Embeddings エンドポイント）
- Flink Table API/SQL
- jtokkit ライブラリ（トークンカウント）

**権限による制御**：OpenAI API キーによる認証が必要。

## 関連画面

本機能は UI 画面と直接関連しない基盤機能である。

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | - |

## 機能種別

外部 API 連携 / 機械学習推論 / 非同期データ処理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| endpoint | String | Yes | OpenAI API エンドポイント URL | 有効な URL |
| api-key | String | Yes | OpenAI API キー | 非空文字列 |
| model | String | Yes | モデル名（例: gpt-3.5-turbo, text-embedding-ada-002） | 非空文字列 |
| max-context-size | Integer | No | コンテキスト最大トークン数 | 正の整数 |
| context-overflow-action | Enum | No | コンテキストオーバーフロー時の動作（デフォルト: TRUNCATED_TAIL） | 列挙値 |
| error-handling-strategy | Enum | No | エラーハンドリング戦略（デフォルト: RETRY） | RETRY, FAILOVER, IGNORE |
| retry-num | Integer | No | リトライ回数（デフォルト: 100） | 正の整数 |
| retry-fallback-strategy | Enum | No | リトライ失敗時のフォールバック（デフォルト: FAILOVER） | FAILOVER, IGNORE |
| system-prompt | String | No | システムプロンプト（デフォルト: "You are a helpful assistant."） | - |
| temperature | Double | No | 出力のランダム性（0.0-1.0） | 0.0-2.0 |
| top-p | Double | No | トークン選択の確率閾値 | 0.0-1.0 |
| stop | String | No | 停止シーケンス（カンマ区切り） | - |
| max-tokens | Long | No | 生成トークンの最大数 | 正の整数 |
| presence-penalty | Double | No | 新規トピックへの誘導ペナルティ | -2.0-2.0 |
| n | Long | No | 生成する選択肢の数（デフォルト: 1） | 正の整数 |
| seed | Long | No | 再現性のためのシード値 | - |
| response-format | Enum | No | レスポンス形式（TEXT, JSON_OBJECT） | 列挙値 |
| dimension | Long | No | 埋め込みベクトルの次元数 | 正の整数 |

### 入力データソース

- Flink Table の入力カラム（単一の VARCHAR 型カラム）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| output | VARCHAR / ARRAY<FLOAT> | チャット補完の場合は文字列、埋め込みの場合は浮動小数点配列 |
| error-string | STRING (メタデータ) | エラー発生時のエラーメッセージ |
| http-status-code | INT (メタデータ) | HTTP ステータスコード |
| http-headers-map | MAP<STRING, ARRAY<STRING>> (メタデータ) | HTTP レスポンスヘッダー |

### 出力先

- Flink Table の出力カラム

## 処理フロー

### 処理シーケンス

```
1. OpenAIModelProviderFactory.createModelProvider()
   └─ エンドポイント URL に基づいて適切な関数を選択

2. AbstractOpenAIModelFunction.open()
   └─ OpenAI クライアントの初期化
   └─ コンテキスト制限用のエンコーディング初期化

3. asyncPredict() 呼び出し
   └─ 入力の NULL チェック
   └─ コンテキストオーバーフロー処理

4. asyncPredictInternal() 実行
   └─ API リクエストパラメータの構築
   └─ OpenAI API への非同期リクエスト送信

5. レスポンス変換
   └─ ChatCompletion/Embedding レスポンスを RowData に変換

6. エラーハンドリング
   └─ エラー発生時は戦略に従って処理

7. close() 呼び出し
   └─ OpenAI クライアントのリリース
```

### フローチャート

```mermaid
flowchart TD
    A[入力 RowData 受信] --> B{入力が NULL?}
    B -->|Yes| C[空リスト返却]
    B -->|No| D[コンテキスト制限処理]
    D --> E{トークン数超過?}
    E -->|Yes| F{overflow action}
    E -->|No| G[asyncPredictInternal 呼び出し]
    F -->|TRUNCATED| H[トークン切り詰め]
    F -->|SKIPPED| C
    H --> G
    G --> I[API リクエスト構築]
    I --> J[OpenAI API 呼び出し]
    J --> K{レスポンス成功?}
    K -->|Yes| L[RowData に変換]
    K -->|No| M{error strategy}
    M -->|RETRY| N[リトライ]
    M -->|FAILOVER| O[例外スロー]
    M -->|IGNORE| P[エラー情報付きで返却]
    N --> J
    L --> Q[結果返却]
    P --> Q
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | 単一入力カラム | 入力スキーマは単一の VARCHAR カラムのみ許可 | 常時 |
| BR-02 | 単一出力カラム | 出力スキーマは単一の物理カラムのみ許可 | 常時 |
| BR-03 | コンテキスト制限 | max-context-size 設定時はトークン数をチェック | max-context-size 設定時 |
| BR-04 | リトライ制御 | rate limiting 対応のため最大 100 回（デフォルト）リトライ | error-handling-strategy が RETRY の場合 |
| BR-05 | クライアント共有 | 同一 baseUrl/apiKey の組み合わせでクライアントを共有 | 常時 |

### 計算ロジック

- トークンカウント: jtokkit ライブラリを使用してモデル固有のエンコーディングでカウント
- コンテキスト切り詰め: 先頭または末尾から超過トークンを削除

## データベース操作仕様

本機能はデータベースを使用しない。外部 API との通信を行う。

### 操作別データベース影響一覧

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | 外部 API を使用 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| OpenAIServiceException | API エラー | OpenAI API からエラーレスポンス | error-handling-strategy に従う |
| IllegalArgumentException | バリデーションエラー | スキーマ不正、パラメータ不正 | 設定を修正 |
| RuntimeException | 実行時エラー | FAILOVER 戦略選択時 | ジョブ失敗 |

### リトライ仕様

- デフォルトリトライ回数: 100 回（OpenAI の rate limiting 対応）
- リトライはクライアント内部で実行
- リトライ失敗後は retry-fallback-strategy に従う

## トランザクション仕様

- 各推論リクエストは独立した非同期処理
- トランザクション制御なし（外部 API 呼び出し）

## パフォーマンス要件

- 非同期処理（AsyncPredictFunction）により高スループットを実現
- OpenAI クライアントの参照カウント管理による効率的なリソース利用
- コンテキスト制限によりトークン超過を防止

## セキュリティ考慮事項

- API キーは設定パラメータとして渡される（環境変数からの取得を推奨）
- 入力データは OpenAI API に送信される（データプライバシーに注意）
- HTTPS による通信暗号化

## 備考

- flink-model-openai は Experimental 機能として提供されている
- OpenAI API 互換のエンドポイント（Azure OpenAI 等）でも使用可能
- トークンカウントには jtokkit ライブラリを使用

---

## コードリーディングガイド

本機能を理解するために参照すべきファイルと、推奨する読み解き順序を以下に示す。

### 推奨読解順序

#### Step 1: データ構造を理解する

まず、設定オプションとエラーハンドリングの構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | OpenAIOptions.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/OpenAIOptions.java` | 設定オプションの定義 |
| 1-2 | ContextOverflowAction.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/ContextOverflowAction.java` | コンテキストオーバーフロー処理の列挙型 |

**読解のコツ**: OpenAIOptions は ConfigOption を使用して各パラメータを定義している。@Documentation.Section アノテーションでドキュメント生成用のカテゴリを指定している。

#### Step 2: エントリーポイントを理解する

ModelProviderFactory によるプロバイダー生成を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | OpenAIModelProviderFactory.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/OpenAIModelProviderFactory.java` | ファクトリの実装 |

**主要処理フロー**:
1. **32行目**: IDENTIFIER として "openai" を定義
2. **35-51行目**: createModelProvider() でエンドポイントに基づき適切な関数を選択
3. **43-48行目**: embeddings エンドポイントなら EmbeddingModelFunction、chat/completions なら ChatModelFunction
4. **59-65行目**: requiredOptions() で必須オプションを定義
5. **68-86行目**: optionalOptions() でオプションパラメータを定義

#### Step 3: 抽象クラスを理解する

共通の処理ロジックを持つ抽象クラスを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | AbstractOpenAIModelFunction.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/AbstractOpenAIModelFunction.java` | 基底クラスの実装 |

**主要処理フロー**:
- **79-102行目**: コンストラクタでオプション解析とスキーマ検証
- **104-110行目**: open() でクライアント初期化とエンコーディング初期化
- **113-127行目**: asyncPredict() で NULL チェックとコンテキスト制限処理
- **143-168行目**: validateSingleColumnSchema() で入出力スキーマの検証
- **197-224行目**: handleErrorsAndRespond() でエラーハンドリング戦略に基づく処理
- **227-262行目**: ErrorHandlingStrategy と RetryFallbackStrategy の列挙型定義
- **268-336行目**: ErrorMessageMetadata でエラー情報のメタデータ定義

#### Step 4: 具体的なモデル関数を理解する

チャット補完と埋め込みの具体的な実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | OpenAIChatModelFunction.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/OpenAIChatModelFunction.java` | チャット補完の実装 |
| 4-2 | OpenAIEmbeddingModelFunction.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/OpenAIEmbeddingModelFunction.java` | 埋め込みの実装 |

**OpenAIChatModelFunction の主要処理フロー**:
- **44行目**: ENDPOINT_SUFFIX として "chat/completions" を定義
- **53-64行目**: コンストラクタでモデル名、システムプロンプト、出力カラムインデックスを設定
- **86-106行目**: asyncPredictInternal() で ChatCompletionCreateParams を構築して API 呼び出し
- **108-126行目**: convertToRowData() でレスポンスを RowData に変換

**OpenAIEmbeddingModelFunction の主要処理フロー**:
- **43行目**: ENDPOINT_SUFFIX として "embeddings" を定義
- **49-60行目**: コンストラクタでモデル名、次元数、出力カラムインデックスを設定
- **82-92行目**: asyncPredictInternal() で EmbeddingCreateParams を構築して API 呼び出し
- **94-114行目**: convertToRowData() で埋め込みベクトルを Float 配列に変換

#### Step 5: ユーティリティクラスを理解する

クライアント管理とトークン処理のユーティリティを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | OpenAIUtils.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/OpenAIUtils.java` | クライアント管理 |
| 5-2 | ContextOverflowAction.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/ContextOverflowAction.java` | トークン制限処理 |

**OpenAIUtils の主要処理フロー**:
- **42-62行目**: createAsyncClient() で参照カウント付きクライアントを生成/取得
- **64-77行目**: releaseAsyncClient() で参照カウントを減らし、0 になったらクローズ

**ContextOverflowAction の主要処理フロー**:
- **43-109行目**: TRUNCATED_TAIL/HEAD、SKIPPED 等の列挙値定義
- **130-146行目**: initializeEncodingForContextLimit() でモデル用エンコーディングを初期化
- **156-169行目**: processTokensWithLimit() でトークン制限を適用

### プログラム呼び出し階層図

```
OpenAIModelProviderFactory.createModelProvider()
    |
    +-- OpenAIChatModelFunction / OpenAIEmbeddingModelFunction
            |
            +-- AbstractOpenAIModelFunction (基底クラス)
                    |
                    +-- open()
                    |       |
                    |       +-- OpenAIUtils.createAsyncClient()
                    |       +-- ContextOverflowAction.initializeEncodingForContextLimit()
                    |
                    +-- asyncPredict()
                    |       |
                    |       +-- ContextOverflowAction.processTokensWithLimit()
                    |       +-- asyncPredictInternal()
                    |               |
                    |               +-- client.chat().completions().create()
                    |               |   または
                    |               +-- client.embeddings().create()
                    |               |
                    |               +-- convertToRowData()
                    |                       |
                    |                       +-- handleErrorsAndRespond() (エラー時)
                    |
                    +-- close()
                            |
                            +-- OpenAIUtils.releaseAsyncClient()
```

### データフロー図

```
[入力]                      [処理]                           [出力]

RowData               AbstractOpenAIModelFunction        RowData
(VARCHAR) -------> asyncPredict() ----------------> (VARCHAR/ARRAY<FLOAT>)
    |                      |
    |                      v
    |              コンテキスト制限チェック
    |                      |
    |                      v
    |              asyncPredictInternal()
    |                      |
    |                      v
    |              API リクエスト構築
    |                      |
    |                      v
    +---------------> OpenAI API <---------------+
                           |                     |
                           v                     |
                    API レスポンス               |
                           |                     |
                           v                     |
                    convertToRowData() --------->|
                           |                     |
                           v                     |
                    handleErrorsAndRespond() <---+
                           |                    (エラー時)
                           v
                    結果 RowData
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| OpenAIModelProviderFactory.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/` | ソース | ファクトリクラス |
| AbstractOpenAIModelFunction.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/` | ソース | 基底関数クラス |
| OpenAIChatModelFunction.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/` | ソース | チャット補完関数 |
| OpenAIEmbeddingModelFunction.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/` | ソース | 埋め込み関数 |
| OpenAIOptions.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/` | ソース | 設定オプション定義 |
| OpenAIUtils.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/` | ソース | ユーティリティクラス |
| ContextOverflowAction.java | `flink-models/flink-model-openai/src/main/java/org/apache/flink/model/openai/` | ソース | コンテキストオーバーフロー処理 |
| pom.xml | `flink-models/flink-model-openai/` | 設定 | Maven ビルド設定 |
| org.apache.flink.table.factories.Factory | `flink-models/flink-model-openai/src/main/resources/META-INF/services/` | 設定 | SPI 登録ファイル |
| pom.xml | `flink-models/` | 設定 | 親モジュール Maven 設定 |
