Decision Scheduler — Runtime OSにおけるAI意思決定の実行を調整する仕組み

Knowledge Base Archive この記事は Chinoba Knowledge Base の技術アーカイブです。 Chinoba.orgを見る →

🎥 YouTubeでも公開しています

Runtime OS Architecture｜Centralized・Edge・Federatedで実現するAI意思決定基盤

Runtime OS Architecture｜Centralized・Edge・Federatedで実現するAI意思決定基盤

Books: Runtime OS 実践ガイド: AI Agent • Human Gate • Decision Trace* 統合する実装アーキテクチャ

通常のOS（Operating System）が、CPU・メモリ・プロセスといった計算資源を安全に協調制御するための基盤であったように、Runtime OS は、知能・人間・ルール・組織・現実世界を安全に協調実行するための基盤として考えることができます。

従来OSが、

「CPU・メモリ・プロセスの協調制御」

を行っていたのに対し、

Runtime OS は、

「意思決定・エージェント・境界・責任・人間承認の協調制御」

を行う存在です。

この観点で見ると、Runtime OS の構造は、実は従来のOSアーキテクチャと非常に強い対応関係を持っています。デバイスドライバに相当するSignal Input Layer、CPU Schedulerに対応するDecision Scheduler、Kernelに対応するDecision Runtime Kernel、権限管理に対応するBoundary Layerなど、AI時代の意思決定システムは、OS的な構造として理解できるようになっていくのです。

ここではDecision Schedulerについて詳しく検討します。

概要（CPUスケジューラ相当）

Runtime OS において、Decision Scheduler は最も重要な中核機構の一つです。

通常のOSにおいてCPU Schedulerが中心的役割を果たしているように、Runtime OS においては、Decision Scheduler が「知能と意思決定の流れ」を制御します。

従来のOSでは、CPUという限られた計算資源を、複数のプロセス間で安全かつ効率的に共有する必要があります。

例えばOSは：

  • どのプロセスを実行するか
  • どれくらいCPU時間を与えるか
  • どの順番で処理するか
  • 割り込みをどう扱うか
  • 優先度をどう決めるか

を常に判断しています。

もしSchedulerが存在しなければ、あるプロセスがCPUを独占し、システム全体は停止してしまいます。

つまりCPU Schedulerとは、

「何を、いつ、どれくらい実行するか」
を制御する仕組み

なのです。

Runtime OS でも、まったく同じ問題が発生します。

ただし、Runtime OS が扱うのはCPU時間ではありません。

Runtime OS が扱うのは：

  • 意思決定
  • Agent execution
  • Human review
  • Boundary evaluation
  • Escalation
  • Workflow coordination

です。

つまり Runtime OS において不足する資源は、

「計算能力」

ではなく、

「安全に意思決定できる能力」

なのです。

そのため Runtime OS では、

Decision Scheduler

が必要になります。

Runtime OS における Scheduler の役割

Decision Scheduler の役割は、

「どのDecisionを、どの条件で、どの順序で実行するか」

を制御することです。

具体的には：

  • どのAgentを起動するか
  • どのSignalを優先するか
  • どのBoundaryを適用するか
  • Human Gate が必要か
  • Escalation するか
  • 実行を停止するか
  • 他Decisionと競合していないか
  • 既存Policyと矛盾していないか

を判断します。

つまり Runtime OS の Scheduler は、

「意思決定スケジューラ」

なのです。

なぜ Decision Scheduler が必要なのか

AIシステムが単独で存在するだけなら、
Schedulerはそこまで重要ではありません。

しかし現実世界では：

  • 複数Agent
  • 人間
  • 業務ルール
  • 法律
  • 安全基準
  • 現場制約
  • 組織権限

が同時に存在します。

例えば製造業では：

  • 温度異常Signal
  • AI検査Agent
  • 作業員報告
  • 品質管理ルール
  • 生産計画
  • 安全Boundary

が同時に流れ込みます。

ここで問題になるのは、

「何を最優先で扱うべきか」

です。

例えば：

  • ただちに工場停止すべきか
  • Human approval を待つべきか
  • AIの推論を追加実行すべきか
  • Escalation すべきか
  • 単なるノイズとして無視すべきか

を判断する必要があります。

この判断を行うのが、
Decision Scheduler です。

CPU Scheduler との対応

通常OSでは：

Process
  ↓
CPU Scheduler
  ↓
CPU Execution

です。

Runtime OS では：

Signal
  ↓
Decision Scheduler
  ↓
Decision Execution

になります。

つまり Runtime OS における「CPU時間」に相当するものは、

意思決定実行権

です。

Decision Scheduler が扱う対象

Decision Scheduler が扱うものは非常に多いです。

1. Signal Priority

まず、Signal の優先順位を判断します。

例えば：

Signal Priority
人命リスク Critical
工場停止リスク High
品質異常 Medium
通常ログ Low

です。

2. Agent Scheduling

どのAgentを起動するかを決定します。

例えば：

Risk Analysis Agent
Vision Inspection Agent
Human Approval Agent
Policy Validation Agent
Escalation Agent

などです。

ここでは：

  • 同時実行するか
  • 順次実行するか
  • 再試行するか
  • 停止するか

を制御します。

3. Boundary Selection

どのBoundaryを適用するかを決定します。

例えば：

Safety Boundary
Financial Boundary
Medical Boundary
Privacy Boundary
Operational Boundary

などです。

例えば金融送金なら：

  • 金額
  • 国
  • 法規制
  • AML
  • 承認権限

によって適用Boundaryが変わります。

4. Human Gate Scheduling

人間承認を要求するかを決定します。

例えば：

Action Human Gate
メール返信 不要
契約変更 必要
工場停止 必須
医療判断変更 必須

です。

ここでは：

  • 誰へ承認依頼するか
  • どのレベルの人間が必要か
  • タイムアウトするか
  • override可能か

まで制御します。

5. Escalation Control

問題が大きい場合は Escalation を行います。

例えば：

AI Agent
  ↓
Supervisor Agent
  ↓
Human Operator
  ↓
Manager
  ↓
Emergency Stop

のように段階的に上げます。

これは通常OSの interrupt handling に近い構造です。

Runtime OS における「公平性」

通常OSのSchedulerは：

  • starvation防止
  • fairness
  • latency

を考慮します。

Runtime OS でも同様に：

  • 重要Decisionが無視されないか
  • Human approval が滞留していないか
  • 特定Agentが暴走していないか
  • Boundary bypass が発生していないか

を管理する必要があります。

つまり Runtime OS では、

「意思決定公平性」

が必要になります。

Runtime OS Scheduler の本質

通常OSのSchedulerは：

CPU実行順序を制御する

ものでした。

しかし Runtime OS の Scheduler は：

社会的意思決定の実行順序を制御する

ものになります。

つまり Runtime OS は：

  • AI
  • 人間
  • 組織
  • 法律
  • 安全
  • 現場

を横断して、

「何を今、実行すべきか」

を決める存在になるのです。

DTMとの関係

Decision Trace Model（DTM）的には：

Event
  ↓
Signal
  ↓
Decision Scheduler
  ↓
Decision
  ↓
Boundary
  ↓
Human
  ↓
Execution
  ↓
Trace

の中心に位置します。

つまり Decision Scheduler は、

「SignalをDecisionへ変換する中枢」

なのです。

Decision Scheduler の設計原則

したがって、Decision Scheduler は次のように定義できます。

Decision Scheduler とは、多数のSignal・Agent・Boundary・Human Gate・Policy・Escalation条件を調停し、どの意思決定を、どの順序で、どの権限で、どの安全条件のもとに実行するかを制御する Runtime OS の中核機構である。

これが、
Runtime OS における Scheduler の本質です。

具体設計案

基本構成は次のようにします。

Normalized Signal
  ↓
Decision Request Builder
  ↓
Priority Evaluator
  ↓
Policy / Boundary Resolver
  ↓
Agent Planner
  ↓
Human Gate Resolver
  ↓
Escalation Controller
  ↓
Decision Queue
  ↓
Execution Dispatcher
  ↓
Decision Trace

1. Decision Request Builder

まず、Normalized Signal から Decision Request を生成します。

Signal はまだ「入力」ですが、Decision Request は「判断対象」です。

例：

{
  "decision_request_id": "dr_001",
  "trace_id": "trace_001",
  "source_signal_id": "sig_001",
  "decision_type": "safety_response",
  "target": "machine_12",
  "requested_action": "evaluate_shutdown_need",
  "context": {
    "location": "factory_line_A",
    "risk_level": "high"
  }
}

ここで重要なのは、

Signal
↓
Decision Request

へ変換することです。

2. Priority Evaluator

次に、Decision Request の優先度を計算します。

評価軸は以下です。

risk_level
urgency
human_impact
business_impact
reversibility
legal_impact
confidence
source_trust_score
time_sensitivity

例：

{
  "priority": "critical",
  "priority_score": 0.93,
  "reason": [
    "high_risk",
    "human_safety_related",
    "low_reversibility"
  ]
}

優先度は、最初は4段階で十分です。

Critical
High
Medium
Low

3. Policy / Boundary Resolver

次に、適用すべき Boundary と Policy を決定します。

例えば：

{
  "boundaries": [
    "safety_boundary",
    "operational_boundary"
  ],
  "policies": [
    "factory_shutdown_policy",
    "human_safety_policy"
  ]
}

ここでは、次を判断します。

このDecisionは安全境界に関係するか
法規制に関係するか
人間承認が必須か
自動実行してよいか
停止条件に該当するか

4. Agent Planner

次に、どのAgentを起動するかを決めます。

例：

{
  "agent_plan": [
    {
      "agent": "risk_analysis_agent",
      "mode": "sync",
      "timeout_sec": 10
    },
    {
      "agent": "policy_validation_agent",
      "mode": "sync",
      "timeout_sec": 5
    },
    {
      "agent": "human_approval_agent",
      "mode": "async",
      "timeout_sec": 300
    }
  ]
}

Agent Planner では、

どのAgentを呼ぶか
同期か非同期か
並列か直列か
タイムアウトは何秒か
失敗時に再試行するか

を決めます。

5. Human Gate Resolver

次に、人間承認が必要かを判断します。

判断基準は以下です。

risk_score が高い
不可逆な操作
人命・医療・金融・法務に関わる
Boundary を超える可能性がある
AI confidence が低い
Policy が人間承認を要求している

例：

{
  "requires_human_gate": true,
  "approval_level": "manager",
  "approval_reason": "human_safety_related_action",
  "timeout_sec": 300,
  "on_timeout": "escalate"
}

6. Escalation Controller

次に、Escalation 条件を設定します。

例：

{
  "escalation_path": [
    "supervisor_agent",
    "human_operator",
    "plant_manager",
    "emergency_stop"
  ],
  "escalation_trigger": [
    "timeout",
    "boundary_violation",
    "agent_conflict",
    "risk_score_above_threshold"
  ]
}

Escalation は、通常OSでいう interrupt handling に近いです。

通常処理で扱えない場合に、より高い権限・安全側の処理へ移します。

7. Decision Queue

スケジューリング済みのDecisionを Queue に入れます。

critical_decision_queue
high_decision_queue
normal_decision_queue
low_decision_queue
human_pending_queue
escalation_queue

ここで重要なのは、Queueを分けることです。

すべてを1つのQueueに入れると、重要なDecisionが埋もれます。

8. Execution Dispatcher

最後に、実行先へ渡します。

{
  "dispatch_target": "execution_layer",
  "execution_mode": "human_approved_required",
  "allowed_actions": [
    "stop_machine",
    "notify_operator",
    "request_additional_inspection"
  ],
  "blocked_actions": [
    "auto_restart_machine"
  ]
}

Execution Dispatcher は、Scheduler が決めた条件を守って実行させます。

Decision Request Schema 案

最小構成はこれで良いです。

{
  "decision_request_id": "dr_001",
  "trace_id": "trace_001",
  "source_signal_ids": ["sig_001"],

  "decision_type": "safety_response",
  "target": "machine_12",
  "requested_action": "evaluate_shutdown_need",

  "priority": "critical",
  "priority_score": 0.93,

  "risk_score": 0.91,
  "urgency": "high",
  "confidence": 0.86,

  "boundaries": [
    "safety_boundary",
    "operational_boundary"
  ],

  "policies": [
    "factory_shutdown_policy"
  ],

  "agent_plan": [
    "risk_analysis_agent",
    "policy_validation_agent"
  ],

  "requires_human_gate": true,
  "approval_level": "manager",

  "escalation_path": [
    "human_operator",
    "plant_manager",
    "emergency_stop"
  ],

  "status": "scheduled",
  "created_at": "2026-05-26T10:00:00+09:00"
}

スケジューリングルール例

最初はルールベースで十分です。

if risk_score >= 0.9 and human_impact == true:
    priority = Critical
    requires_human_gate = true
    escalation_path = Emergency

elif risk_score >= 0.7:
    priority = High
    requires_human_gate = true

elif confidence < 0.5:
    priority = Medium
    agent_plan = Additional Validation

else:
    priority = Normal

MVP実装構成

FastAPI + PostgreSQL + Redis Queue なら、以下の構成が良いです。

scheduler/
  schemas.py
  request_builder.py
  priority_evaluator.py
  boundary_resolver.py
  agent_planner.py
  human_gate_resolver.py
  escalation_controller.py
  dispatcher.py
  service.py

データの流れは：

normalized_signals table
  ↓
scheduler service
  ↓
decision_requests table
  ↓
redis decision queues
  ↓
agent / human / execution layer
  ↓
decision_traces table

Pydantic イメージ

from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime

class DecisionRequest(BaseModel):
    decision_request_id: str
    trace_id: str
    source_signal_ids: List[str]

    decision_type: str
    target: str
    requested_action: str

    priority: str
    priority_score: float

    risk_score: float
    urgency: str
    confidence: float

    boundaries: List[str]
    policies: List[str]

    agent_plan: List[str]

    requires_human_gate: bool
    approval_level: Optional[str] = None

    escalation_path: List[str]

    status: str = "scheduled"
    created_at: datetime

処理イメージ

def schedule_decision(signal):
    decision_request = build_decision_request(signal)

    priority = evaluate_priority(decision_request)
    boundaries = resolve_boundaries(decision_request)
    agents = plan_agents(decision_request)
    human_gate = resolve_human_gate(decision_request, boundaries)
    escalation = build_escalation_path(decision_request, human_gate)

    decision_request.priority = priority["level"]
    decision_request.priority_score = priority["score"]
    decision_request.boundaries = boundaries
    decision_request.agent_plan = agents
    decision_request.requires_human_gate = human_gate["required"]
    decision_request.approval_level = human_gate.get("approval_level")
    decision_request.escalation_path = escalation

    save_decision_request(decision_request)
    enqueue_decision(decision_request)

    return decision_request

設計上の核心

Decision Scheduler は、単に「順番を決める機能」ではありません。

それは、

Signal
↓
Decision候補
↓
Boundary評価
↓
Human Gate判定
↓
Escalation判断
↓
Execution許可

をつなぐ中枢です。

つまり設計原則はこうです。

Decision Scheduler は、多数のSignalを、優先度・安全境界・人間承認・Agent実行計画・Escalation条件を持つ Decision Request に変換し、実行可能な順序で Runtime OS 内に流す中核機構である。

この設計にすると、DTM の

Event → Signal → Decision → Boundary → Human → Log

に自然につながります。

Related Research

この記事は Chinoba Knowledge Base の一部です。

Chinoba Research
Chinoba-lab Open Source
Books and Library

コメント

タイトルとURLをコピーしました