0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@0903osaTeam Tea & Water

第三章：人間とAIの協調ループ（Human-in-the-Loop: HITL

0
Last updated at Posted at 2026-06-25

人間とAIの協調ループ（Human-in-the-Loop: HITL）

AIエージェントの自律性が高まる一方で、システムの決定を完全にAIに委ねることには、ハルシネーション（幻覚）や意図しない動作、セキュリティ上の脆弱性といった重大なリスクが伴います。特に「メールの送信」「決済処理」「本番コードのデプロイ」といった、不可逆なアクション（Side Effect）を伴うタスクにおいては、人間が介在するシステム設計が必須となります。

本章では、自律性と安全性のバランスを取るための Human-in-the-Loop (HITL) のグラデーション設計、UXにおける割り込みの接点、承認・修正・監査のライフサイクル、そしてハルシネーションを前提としたUX設計パターンについて、実装レベルのアーキテクチャとコード例を交えて解説します。

3.1 完全自律（No Loop）と人間介入（HITL）のグラデーション設計

エージェントシステムを設計する際、自動化の度合いを「0か100か」で考えるべきではありません。タスクのリスク（失敗した際の影響度）と複雑さに応じて、人間とAIの役割分担をグラデーション（段階的）に設計する必要があります。

自動化レベルのグラデーションモデル

一般に、HITLシステムは以下の3つのトポロジーに分類されます。

  1. Human-in-the-Loop (HITL - 介入・承認)
    • 動作原理: エージェントが特定のアクション（例: データベースの更新、メール送信）を実行する直前に、処理を必ず一時停止し、人間の明示的な「承認（Approve）」を待ちます。
    • 適したユースケース: 金融取引、本番環境への変更、顧客への直接連絡など、失敗のコストが極めて高いタスク。
  2. Human-on-the-Loop (HOTL - 監視・割り込み)
    • 動作原理: エージェントは基本的に完全自律で動作し続けますが、実行ログやステートを人間がダッシュボード上でリアルタイムに監視し、問題が発生した際に「割り込み（Interrupt）」や「緊急停止（Emergency Stop）」を行うことができます。
    • 適したユースケース: 大量のWebスクレイピング、ドキュメントの自動整理、長時間実行されるデータパイプライン。
  3. Human-in-command (HIC - 主導・協調)
    • 動作原理: 人間がタスクの分解や大局的な意思決定の大部分をコントロールし、AIは「数式の計算」「プログラミングコードの個別パーツの実装」など、完全に制御されたサンドボックス内での定型作業のみを担当します。
    • 適したユースケース: 複雑なシステム設計、法的なドキュメントのドラフト作成。

意思決定マトリクス

プロジェクトマネージャー（PM）やシステム設計者は、以下のマトリクスを基準に自動化レベルを決定します。

アクションの影響範囲 失敗の回復性 推奨されるアプローチ 具体的な実装方針
極めて高い (決済、インフラ破壊) 不可逆 (回復困難) HITL (ゲート型) 特定ノードの実行前に必ず breakpoint を設定し、人間がUI上で「承認」を押すまで状態を凍結。
中程度 (メール下書き、社内通知) 可逆 (修正が容易) HITL (レビュー型) AIが成果物（ドラフト）を作成し、人間が「修正」を加えた上で送信ボタンを自ら押す。
低い (要約、データ分析) 可逆 (再実行可能) HOTL / 完全自律 エージェントに完全な自律走行を許容しつつ、エラーや異常出力を検知した際のみアラート通知。

3.2 ユーザー体験（UX）における「割り込み」と「フィードバック」の接点設計

HITLを支えるエンジニアリング上の核心は、「状態の保存（チェックポイント）」 と 「実行の一時停止と再開」 です。非同期で動作するエージェントに対して、人間のユーザーがストレスなく割り込みやフィードバックを行えるように設計する必要があります。

シーケンス設計：人間割り込みのライフサイクル

エージェントが非同期タスクを実行し、途中でユーザーの承認を求める際の標準的なシーケンスは以下の通りです。

エンジニアリング上の課題：ステートの永続化

Webアプリケーションでこれを行う場合、エージェントを実行するスレッドやプロセスをメモリ上でスリープさせて待つことはできません。なぜなら、人間のレビューには数分〜数日かかることがあるからです。
したがって、以下の要件を満たす必要があります。

  1. チェックポインタ（Checkpointer）: 各ノードの実行が終わるたびに、エージェントの内部状態（メモリ、変数、メッセージ履歴）をリレーショナルデータベース（PostgreSQL等）やKey-Valueストア（Redis等）にシリアル化（JSON等）して保存する。
  2. スレッドセーフな再起動: ユーザーが承認リクエストを送信した際、一時停止されたスレッドID（または thread_id ）を指定して、その時点のDBの状態からエージェントを瞬時に再構築し、次のノードを実行する。

3.3 承認・修正・監査（Approve, Correct, Audit）のライフサイクル

ここでは、エンジニアが実際にシステムに組み込むための「承認・修正・監査」のトリプル・パターンについて、LangGraphのステートマシン設計をベースにしたPythonコード例で示します。

具体例として、「AIがSQLクエリを生成し、実行前に人間がレビュー・修正・実行する」 というシナリオをモデリングします。

実装例：人間によるSQLレビューと修正ループ

このコードでは、人間が「SQLをそのまま実行する（Approve）」「SQLをエディタで直接書き換える（Correct）」「フィードバックを戻してAIに再生成させる（Audit / Feedback）」の3つのアクションをとれるように設計しています。

from typing import TypedDict, Optional, Literal
import json

# =====================================================================
# 1. 状態の定義
# =====================================================================
class SqlTaskState(TypedDict):
    question: str
    generated_sql: Optional[str]
    human_feedback: Optional[str]
    human_action: Optional[Literal["approve", "correct", "reject"]]
    execution_result: Optional[str]
    status: str

# =====================================================================
# 2. 各ノード（処理）の定義
# =====================================================================
def sql_generator_node(state: SqlTaskState) -> dict:
    """AIがSQLを生成するノード"""
    print("\n[Node] AI: SQLの生成/修正を行います。")
    question = state["question"]
    feedback = state.get("human_feedback")
    
    # 本来はここでLLM APIを叩いて生成する
    if not feedback:
        # 初回生成
        generated_sql = "SELECT id, name, salary FROM users WHERE salary > 100000;"
        print(f"  -> 生成されたSQL: {generated_sql}")
    else:
        # フィードバックを受けた再生成
        print(f"  -> フィードバックを確認: '{feedback}'")
        generated_sql = "SELECT id, name, salary FROM users WHERE salary > 100000 AND status = 'active';"
        print(f"  -> 修正されたSQL: {generated_sql}")
        
    return {
        "generated_sql": generated_sql,
        "human_feedback": None,  # フィードバックをクリア
        "human_action": None,    # アクションをクリアして無限ループを防止
        "status": "pending_review"
    }

def human_review_node(state: SqlTaskState) -> dict:
    """
    ブレークポイントとして機能するモックノード。
    本番環境では、ここで実行が一時停止し、Web UIからのAPI呼び出しを待ちます。
    """
    print("\n[Node] Human Review: 人間の入力待ちです...")
    # このノード自体は何もしない。状態は保存され、実行はここで「中断」される。
    return {}

def executor_node(state: SqlTaskState) -> dict:
    """承認されたSQLを実行するノード"""
    print("\n[Node] Database: SQLを実行します。")
    sql = state["generated_sql"]
    print(f"  -> 実行中: {sql}")
    
    # 実行結果のモック
    result = "3 records found: [(1, 'Alice', 120000), (3, 'Bob', 150000)]"
    return {
        "execution_result": result,
        "status": "completed"
    }

# =====================================================================
# 3. エッジ（遷移ルール）の定義
# =====================================================================
def check_review_result(state: SqlTaskState) -> Literal["generate", "execute", "wait"]:
    """レビュー結果に基づいて次のノードを決定するエッジ"""
    action = state.get("human_action")
    
    if action == "approve" or action == "correct":
        return "execute"
    elif action == "reject":
        return "generate"
    
    # アクションが決まっていない場合はレビューノードで待機
    return "wait"

# =====================================================================
# 4. 実行エンジンと状態割り込みのシミュレーション
# =====================================================================
class HITLGraphRunner:
    def __init__(self):
        self.state = {}

    def run(self, state: SqlTaskState) -> SqlTaskState:
        self.state = state
        
        # 簡易ループ
        while True:
            status = self.state.get("status", "start")
            
            if status == "start" or self.state.get("human_action") == "reject":
                # AIによるSQL生成を実行
                output = sql_generator_node(self.state)
                self.state.update(output)
                
            elif self.state.get("status") == "pending_review":
                # レビュー結果を評価
                next_step = check_review_result(self.state)
                
                if next_step == "wait":
                    print("  [System] 人間の入力を待つため、ステートを保存して一時停止します。")
                    # ループを抜け、現在のステートを呼び出し元に返す (一時停止の再現)
                    return self.state
                    
                elif next_step == "execute":
                    # SQLを実行ノードに回す
                    output = executor_node(self.state)
                    self.state.update(output)
                    
            elif status == "completed":
                print("\n[System] プロセスが完了しました。")
                break
                
        return self.state

# =====================================================================
# 5. シナリオテスト（Web API連携のモック）
# =====================================================================
if __name__ == "__main__":
    runner = HITLGraphRunner()
    
    # --- フェーズ 1: ユーザーがタスクを要求し、AIがSQLを生成して一時停止する ---
    print("=== シナリオ: 開始 ===")
    current_state: SqlTaskState = {
        "question": "給料が10万より多いアクティブなユーザーを教えて",
        "generated_sql": None,
        "human_feedback": None,
        "human_action": None,
        "execution_result": None,
        "status": "start"
    }
    
    current_state = runner.run(current_state)
    # ここでエージェントは一時停止し、StateはDBに保存されていると仮定
    
    # --- フェーズ 2: ユーザーがWeb UIでSQLを確認し、修正指示（Audit）を出す ---
    print("\n--- ユーザーアクション: SQLの却下とフィードバック ---")
    # DBからステートをロードし、ユーザーのフィードバックを注入
    current_state["human_action"] = "reject"
    current_state["human_feedback"] = "ステータスが 'active' のユーザーだけに絞り込んでください。"
    
    # エージェント実行を再開
    current_state = runner.run(current_state)
    # AIが再生成し、再び 'pending_review' 状態で一時停止する
    
    # --- フェーズ 3: ユーザーがSQLを確認し、手動で微修正して承認する（Correct） ---
    print("\n--- ユーザーアクション: 手動修正して実行承認 ---")
    # ユーザーがエディタ上でSQLを微修正した想定 (LIMITを追加)
    modified_sql = current_state["generated_sql"] + " LIMIT 10;"
    
    current_state["generated_sql"] = modified_sql
    current_state["human_action"] = "correct"
    
    # エージェント実行を再開
    current_state = runner.run(current_state)
    
    print("\n最終実行結果:", current_state["execution_result"])

3.4 AIのハルシネーションを前提としたUXパターン（段階的公開と人間による上書き）

自律型AIを搭載したプロダクトを設計する上で最も重要なパラダイムシフトは、「AIは間違える（ハルシネーションを起こす）という前提に立つこと」 です。エラーをプログラマティックにゼロにすることは不可能であるため、UI/UXデザインによってその悪影響を無効化します。

パターン1：段階的公開（Gradual Exposure）と「下書き（Draft）」の明示

AIが生成したドキュメントやコード、アクションの実行結果を、即座にエンドユーザーに公開したり、実行環境にデプロイしたりしてはいけません。

[ AI 生成 ] ──> [ 下書き (Draft) ] ──> [ 人間が修正 (Correct) ] ──> [ 公開/実行 ]
                      ▲ (レビュー画面で差分をハイライト)
  • 下書きマーク: AIが生成した要素には、必ず「AI Draft（AIによる下書き）」というバッジを付け、確定していない状態であることをレビューアに視覚的に伝えます。
  • 差分表示 (Diff Highlight): 人間が「過去のドラフト」と「AIの提案」の差分、または「AIの提案」と「人間の修正箇所」の差分を瞬時に確認できる GitライクなUIを用意します。

パターン2：人間による強制上書き（Human Override）のデータスキーマ

AIが誤った推論を続けて修正ループが収束しない場合や、例外的なビジネスルールを適用したい場合、人間がStateの一部のフィールドを強制的に上書きし、AIによる書き換えを「ロック（禁止）」する仕組みをデータスキーマ上に設計します。

推奨されるStateのデータモデル設計（JSON）

{
  "project_id": "proj-1029",
  "summary": "AIが生成したサマリーテキスト...",
  "budget": 500000,
  "meta_info": {
    "summary": {
      "updated_by": "ai",
      "confidence": 0.85,
      "is_locked": false
    },
    "budget": {
      "updated_by": "human_user_45",
      "confidence": 1.0,
      "is_locked": true
    }
  }
}
  • is_locked フラグ: is_locked: true に設定されたフィールドは、その後のエージェントの推論ループ内（LLMへのプロンプト生成や、State更新ノードのロジック）で、LLMが決して上書きしてはならない「確定値（定数）」として扱われます。これにより、人間が一度手入力で修正した値をAIが再度ハルシネーションで書き換えてしまう「手戻り」を防ぐことができます。

まとめ：安全なHITLループを設計するためのエンジニア・デザイナーの心得

  • AIの動作を同期的に待たない: HITLは非同期処理として設計し、ステートの完全永続化とスレッドセーフな再起動を標準とする。
  • すべての変更に履歴と作成者を: updated_by: "ai" と updated_by: "human" を切り分け、人間の主権を常にAIの上に置く。
  • ロック機構の実装: 人間が修正した値はロックし、AIループによる「再書き換えの悲劇」を完全にシャットアウトする。
0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?