# 機能設計書 78-インジェストパイプライン

## 概要

本ドキュメントは、GiselleワークフローシステムのRAG（Retrieval-Augmented Generation）パイプラインにおけるインジェストパイプライン機能の設計を記述する。この機能は、ドキュメントをベクトルストアに取り込む一連の処理を統合したパイプラインを提供する。

### 本機能の処理概要

**業務上の目的・背景**：RAGシステムにおいて、ドキュメントをベクトルストアに登録するには、ドキュメント読み込み、チャンク分割、埋め込み生成、ストア保存という複数のステップが必要となる。この機能は、これらのステップを統合したパイプラインを提供し、差分インジェスト（変更があったドキュメントのみ処理）や並列処理によって効率的なドキュメント取り込みを実現する。

**機能の利用シーン**：
- GitHubリポジトリのコードをベクトルストアに初期登録する場面
- ドキュメント更新時に変更分のみを再インジェストする場面
- 大量のドキュメントを並列で効率的に処理する場面

**主要な処理内容**：
1. ドキュメントローダーからメタデータを非同期で読み込み
2. バージョン管理により変更があったドキュメントのみを処理対象に選定
3. 各ドキュメントのコンテンツを読み込み、チャンク分割、埋め込み生成
4. チャンクストアにチャンクと埋め込みを保存
5. 孤立したドキュメント（削除されたもの）をクリーンアップ
6. 処理進捗とエラー情報を通知

**関連システム・外部連携**：
- DocumentLoader: ドキュメントの読み込み
- Chunker: テキストのチャンク分割
- Embedder: 埋め込み生成
- ChunkStore: チャンクデータの保存

**権限による制御**：埋め込みAPIキーおよびデータベース接続権限が必要。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 31 | GitHub Vector Store設定画面 | 補助機能 | リポジトリデータのベクトルストアへの取り込み処理 |
| 32 | ドキュメントVector Store設定画面 | 補助機能 | ドキュメントデータのベクトルストアへの取り込み処理 |

## 機能種別

データ変換・バッチ処理

## 入力仕様

### 入力パラメータ（createPipeline関数）

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| documentLoader | DocumentLoader<TDocMetadata> | Yes | ドキュメントローダー | - |
| chunkStore | ChunkStore<TChunkMetadata> | Yes | チャンクストア | - |
| documentKey | (metadata: TDocMetadata) => string | Yes | ドキュメントキー生成関数 | - |
| documentVersion | (metadata: TDocMetadata) => string | Yes | ドキュメントバージョン生成関数 | - |
| metadataTransform | (metadata: TDocMetadata) => TChunkMetadata | Yes | メタデータ変換関数 | - |
| embeddingProfileId | EmbeddingProfileId | Yes | 埋め込みプロファイルID | 有効なプロファイルIDであること |
| metadataVersion | (metadata: TDocMetadata) => string | No | メタデータバージョン生成関数 | - |
| chunker | ChunkerFunction | No | チャンカー（デフォルト: createDefaultChunker） | - |
| embeddingComplete | EmbeddingCompleteCallback | No | 埋め込み完了コールバック | - |
| maxBatchSize | number | No | 最大バッチサイズ（デフォルト: 100） | - |
| maxRetries | number | No | 最大リトライ回数（デフォルト: 3） | - |
| retryDelay | number | No | リトライ遅延（デフォルト: 1000ms） | - |
| parallelLimit | number | No | 並列処理数（デフォルト: 15） | - |
| onProgress | (progress: IngestProgress) => void | No | 進捗コールバック | - |
| onError | (error: IngestError) => void | No | エラーコールバック | - |
| headers | Record<string, string> | No | AI Gateway用HTTPヘッダー | - |

### 入力データソース

- DocumentLoaderからのドキュメントメタデータとコンテンツ
- ChunkStoreからの既存ドキュメントバージョン情報

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| IngestResult.totalDocuments | number | 処理対象ドキュメント総数 |
| IngestResult.successfulDocuments | number | 成功したドキュメント数 |
| IngestResult.failedDocuments | number | 失敗したドキュメント数 |
| IngestResult.metadataOnlyUpdates | number | メタデータのみ更新したドキュメント数 |
| IngestResult.errors | Array<{document, error}> | エラー情報の配列 |

### 出力先

- ChunkStoreへのチャンク保存
- 呼び出し元への処理結果返却
- コールバックへの進捗・エラー通知

## 処理フロー

### 処理シーケンス

```
1. パイプライン初期化
   ├─ 埋め込みプロファイルの検証
   ├─ APIキーの取得（Gateway or Provider）
   └─ Embedderの初期化

2. バージョントラッキング初期化
   ├─ ChunkStoreから既存ドキュメントバージョンを取得
   └─ VersionTrackerを作成

3. ドキュメント準備（prepareDocuments）
   ├─ メタデータを非同期で読み込み
   ├─ ドキュメントキーとバージョンを計算
   ├─ 更新要否を判定（needsEmbedding, needsMetadataUpdate）
   └─ 処理対象ドキュメントをyield

4. バッチ処理（processBatch）
   ├─ 並列でドキュメントを読み込み
   ├─ processDocumentで各ドキュメントを処理
   └─ 進捗を通知

5. ドキュメント処理（processDocument）
   ├─ embedContentでチャンク分割＋埋め込み生成
   └─ ChunkStoreにinsert

6. 孤立ドキュメント処理
   ├─ 削除されたドキュメントキーを取得
   └─ ChunkStoreからdeleteBatch

7. 結果返却
   └─ IngestResultを返却
```

### フローチャート

```mermaid
flowchart TD
    A[開始] --> B{プロファイル検証}
    B -->|NG| C[ConfigurationError]
    B -->|OK| D{APIキー取得}
    D -->|なし| E[ConfigurationError]
    D -->|あり| F[Embedder初期化]
    F --> G[既存バージョン取得]
    G --> H[VersionTracker作成]

    subgraph メインループ
        I[メタデータ読み込み] --> J{更新要否判定}
        J -->|埋め込み必要| K[yield ドキュメント]
        J -->|メタデータのみ| L[updateMetadata]
        J -->|更新不要| I
        K --> M[バッチ処理]
        M --> N[ドキュメント読み込み]
        N --> O[embedContent]
        O --> P[ChunkStore.insert]
        P --> I
    end

    H --> I
    L --> I
    I --> Q{全ドキュメント処理完了?}
    Q -->|No| I
    Q -->|Yes| R[孤立ドキュメント取得]
    R --> S{孤立あり?}
    S -->|Yes| T[deleteBatch]
    S -->|No| U[結果返却]
    T --> U
    C --> V[終了]
    E --> V
    U --> V
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | 差分インジェスト | バージョンが変更されたドキュメントのみ再処理 | 既存ドキュメントがある場合 |
| BR-02 | メタデータのみ更新 | コンテンツ未変更でメタデータのみ変更時は再埋め込みしない | metadataVersion指定時 |
| BR-03 | 孤立ドキュメント削除 | ソースに存在しなくなったドキュメントは削除 | インジェスト完了時 |
| BR-04 | リトライ | 処理失敗時は指定回数までリトライ | maxRetries > 0 |
| BR-05 | 並列処理 | parallelLimit内で並列処理 | 常時 |

### 計算ロジック

差分判定ロジック：
```typescript
const { needsEmbedding, needsMetadataUpdate } = versionTracker.checkUpdateRequirement(
    docKey, newVersion, newMetadataVersion
);
// needsEmbedding: コンテンツバージョンが変更 or 新規
// needsMetadataUpdate: メタデータバージョンのみ変更
```

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

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

ChunkStore（PostgreSQL）への操作を行う。

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| バージョン取得 | チャンクテーブル | SELECT | 既存ドキュメントバージョンの取得 |
| チャンク挿入 | チャンクテーブル | DELETE + INSERT | 既存チャンク削除後に新規挿入 |
| メタデータ更新 | チャンクテーブル | UPDATE | メタデータフィールドのみ更新 |
| 孤立削除 | チャンクテーブル | DELETE | 削除されたドキュメントのチャンク削除 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | ConfigurationError | 無効な埋め込みプロファイルID | 有効なプロファイルIDを指定 |
| - | ConfigurationError | 埋め込みAPIキーが未設定 | 環境変数にAPIキーを設定 |
| - | OperationError | ドキュメント処理エラー | ログを確認し、ドキュメント形式を検証 |

### リトライ仕様

- デフォルトで3回のリトライを実行
- リトライ遅延は1000ms（指数バックオフなし）
- リトライ失敗後はエラーを記録して次のドキュメントへ進む

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

- 各ドキュメントの挿入はトランザクション内で実行（DELETE + INSERT）
- パイプライン全体はトランザクション外で実行（部分的な成功を許容）

## パフォーマンス要件

- デフォルト並列数: 15（GitHub API制限を考慮）
- バッチサイズ: 100チャンク/バッチ
- 差分インジェストにより、更新されたドキュメントのみ処理

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

- 埋め込みAPIキーは環境変数から取得
- AI Gatewayを使用する場合は専用APIキーを使用
- ドキュメント内容は外部埋め込みAPIに送信される

## 備考

- デフォルト並列数15はGitHub API制限（5,000/h）とメモリ制約のバランス
- 初期インジェスト時はレート制限に注意が必要
- RagErrorはそのまま再スロー、その他はOperationErrorにラップ

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | types.ts | `packages/rag/src/ingest/types.ts` | IngestResult、IngestProgress、IngestError型定義 |

**読解のコツ**: IngestResultの各フィールドがパイプラインの処理結果を表現している点を理解する。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | pipeline.ts | `packages/rag/src/ingest/pipeline.ts` | createPipeline関数 |

**主要処理フロー**:
1. **66-122行目**: createPipeline関数のオプション処理とEmbedder初期化
2. **253-321行目**: ingest関数（メイン処理）
3. **164-213行目**: prepareDocuments - ドキュメント準備
4. **218-248行目**: processBatch - バッチ処理
5. **126-159行目**: processDocument - 単一ドキュメント処理

#### Step 3: ヘルパー機能を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | version-tracker.ts | `packages/rag/src/ingest/version-tracker.ts` | バージョン追跡 |
| 3-2 | embedder.ts | `packages/rag/src/ingest/embedder.ts` | embedContent関数 |
| 3-3 | utils.ts | `packages/rag/src/ingest/utils.ts` | retryOperation、createBatches |

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

```
createPipeline(options)
    │
    ├─ プロファイル検証
    ├─ APIキー取得
    └─ createEmbedderFromProfile()
           │
           └─ ingest()  [返却される関数]
                  │
                  ├─ chunkStore.getDocumentVersions()
                  │
                  ├─ createVersionTracker(existingVersions)
                  │
                  ├─ prepareDocuments()
                  │      ├─ documentLoader.loadMetadata()
                  │      ├─ documentKey(metadata)
                  │      ├─ documentVersion(metadata)
                  │      ├─ versionTracker.trackSeen(docKey)
                  │      ├─ versionTracker.checkUpdateRequirement()
                  │      └─ chunkStore.updateMetadata() [メタデータのみ更新時]
                  │
                  ├─ createBatches()
                  │
                  ├─ processBatch()
                  │      ├─ documentLoader.loadDocument()
                  │      └─ processDocument()
                  │             ├─ embedContent()
                  │             │      ├─ chunker(content)
                  │             │      └─ embedder.embedMany()
                  │             └─ chunkStore.insert()
                  │
                  ├─ versionTracker.getOrphaned()
                  │
                  └─ chunkStore.deleteBatch()
```

### データフロー図

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

DocumentLoader ────────▶ loadMetadata() ────────▶ TDocMetadata
                                                       │
                                                       ▼
                                    versionTracker.checkUpdateRequirement()
                                                       │
                         ┌─────────────────────────────┴─────────────────────────────┐
                         │                             │                             │
                    [埋め込み必要]              [メタデータのみ]                   [更新不要]
                         │                             │                             │
                         ▼                             ▼                             │
                 loadDocument()             updateMetadata()                         │
                         │                             │                             │
                         ▼                             │                             │
                   embedContent()                      │                             │
                    ├─ chunker()                       │                             │
                    └─ embedMany()                     │                             │
                         │                             │                             │
                         ▼                             │                             │
                 chunkStore.insert()                   │                             │
                         │                             │                             │
                         └─────────────────────────────┴─────────────────────────────┘
                                                       │
                                                       ▼
                                            versionTracker.getOrphaned()
                                                       │
                                                       ▼
                                            chunkStore.deleteBatch()
                                                       │
                                                       ▼
                                                IngestResult
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| pipeline.ts | `packages/rag/src/ingest/pipeline.ts` | ソース | メイン実装 |
| types.ts | `packages/rag/src/ingest/types.ts` | ソース | 型定義 |
| version-tracker.ts | `packages/rag/src/ingest/version-tracker.ts` | ソース | バージョン追跡 |
| embedder.ts | `packages/rag/src/ingest/embedder.ts` | ソース | embedContent関数 |
| utils.ts | `packages/rag/src/ingest/utils.ts` | ソース | ユーティリティ |
| index.ts | `packages/rag/src/ingest/index.ts` | ソース | エクスポート |
