🧮 技術調査 - GPT-5.6三階層のタスク別計算予算と実行基盤

python -m pip install --upgrade openai
export OPENAI_API_KEY="sk-..."

migration_inventory:
  workflow: pr-review
  current_model: gpt-5.5
  current_effort: medium
  candidate_configs:
    - model: gpt-5.6-terra
      effort: medium
    - model: gpt-5.6-terra
      effort: low
    - model: gpt-5.6-sol
      effort: medium
  eval_metrics:
    - correctness
    - missing_tests
    - security_findings
    - latency_ms
    - total_tokens
    - estimated_cost_usd

# tier-routing-policy.yaml 例
policies:
  - task_class: bulk_classification
    model: gpt-5.6-luna
    reasoning:
      effort: none
    max_output_tokens: 1200
    fallback: halt_on_uncertain

  - task_class: standard_agentic_work
    model: gpt-5.6-terra
    reasoning:
      effort: medium
    max_output_tokens: 12000
    fallback: retry_once_then_escalate

  - task_class: high_stakes_final_review
    model: gpt-5.6-sol
    reasoning:
      effort: high
      mode: standard
    max_output_tokens: 32000
    fallback: human_review_only

{
  "type": "function",
  "name": "get_inventory",
  "description": "Return inventory for a sku.",
  "parameters": {
    "type": "object",
    "properties": {
      "sku": { "type": "string" }
    },
    "required": ["sku"],
    "additionalProperties": false
  },
  "output_schema": {
    "type": "object",
    "properties": {
      "sku": { "type": "string" },
      "available_units": { "type": "number" }
    },
    "required": ["sku", "available_units"],
    "additionalProperties": false
  },
  "allowed_callers": ["programmatic"]
}

POST /v1/responses
OpenAI-Beta: responses_multi_agent=v1
Content-Type: application/json

{
  "model": "gpt-5.6-sol",
  "input": "Review this diff with independent agents for correctness, security, and tests.",
  "multi_agent": {
    "enabled": true,
    "max_concurrent_subagents": 3
  }
}

from openai import OpenAI

client = OpenAI()

TIER_POLICY = {
    "bulk": {
        "model": "gpt-5.6-luna",
        "reasoning": {"effort": "none"},
        "max_output_tokens": 1200,
    },
    "standard": {
        "model": "gpt-5.6-terra",
        "reasoning": {"effort": "medium"},
        "max_output_tokens": 12000,
    },
    "critical": {
        "model": "gpt-5.6-sol",
        "reasoning": {"effort": "high"},
        "max_output_tokens": 32000,
    },
}

def run_task(task_class: str, prompt: str, cache_key: str):
    policy = TIER_POLICY[task_class]
    return client.responses.create(
        model=policy["model"],
        reasoning=policy["reasoning"],
        input=[{"role": "user", "content": prompt}],
        max_output_tokens=policy["max_output_tokens"],
        prompt_cache_key=cache_key,
    )

response = client.responses.create(
    model="gpt-5.6-sol",
    reasoning={"mode": "pro", "effort": "high"},
    input="Review this database migration plan for data-loss failure modes.",
    max_output_tokens=25000,
)

if response.status == "incomplete":
    reason = response.incomplete_details.reason
    if reason == "max_output_tokens":
        # 自動リトライは 1 回に限定する
        response = client.responses.create(
            model="gpt-5.6-sol",
            reasoning={"mode": "standard", "effort": "medium"},
            input="Summarize the partial findings and list unresolved checks.",
            previous_response_id=response.id,
            max_output_tokens=32000,
        )
    else:
        raise RuntimeError(f"unexpected incomplete reason: {reason}")

import json
from openai import OpenAI

client = OpenAI()

def get_inventory(sku: str):
    return {"sku": sku, "available_units": 42}

def get_demand(sku: str):
    return {"sku": sku, "requested_units": 31}

implementations = {
    "get_inventory": get_inventory,
    "get_demand": get_demand,
}

tools = [
    {
        "type": "function",
        "name": "get_inventory",
        "description": "Return sku and available_units.",
        "parameters": {
            "type": "object",
            "properties": {"sku": {"type": "string"}},
            "required": ["sku"],
            "additionalProperties": False,
        },
        "output_schema": {
            "type": "object",
            "properties": {
                "sku": {"type": "string"},
                "available_units": {"type": "number"},
            },
            "required": ["sku", "available_units"],
            "additionalProperties": False,
        },
        "allowed_callers": ["programmatic"],
    },
    {
        "type": "function",
        "name": "get_demand",
        "description": "Return sku and requested_units.",
        "parameters": {
            "type": "object",
            "properties": {"sku": {"type": "string"}},
            "required": ["sku"],
            "additionalProperties": False,
        },
        "output_schema": {
            "type": "object",
            "properties": {
                "sku": {"type": "string"},
                "requested_units": {"type": "number"},
            },
            "required": ["sku", "requested_units"],
            "additionalProperties": False,
        },
        "allowed_callers": ["programmatic"],
    },
    {"type": "programmatic_tool_calling"},
]

input_items = [{"role": "user", "content": "Compare inventory with demand for sku_123."}]

while True:
    response = client.responses.create(
        model="gpt-5.6-terra",
        store=False,
        include=["reasoning.encrypted_content"],
        input=input_items,
        tools=tools,
    )
    input_items.extend(item.model_dump(exclude_none=True) for item in response.output)
    calls = [item for item in response.output if item.type == "function_call"]
    if not calls:
        print(response.output_text)
        break
    for call in calls:
        result = implementations[call.name](**json.loads(call.arguments))
        input_items.append({
            "type": "function_call_output",
            "call_id": call.call_id,
            "output": json.dumps(result),
            "caller": call.caller.model_dump() if call.caller else None,
        })

from openai import OpenAI

client = OpenAI()

response = client.beta.responses.create(
    model="gpt-5.6-sol",
    input=(
        "Review this pull request with three agents: correctness, security, "
        "and missing tests. Reconcile duplicate findings and return a prioritized review."
    ),
    multi_agent={
        "enabled": True,
        "max_concurrent_subagents": 3,
    },
    betas=["responses_multi_agent=v1"],
)

{
  "model": "gpt-5.6-terra",
  "prompt_cache_key": "tenant-acme-support-v1",
  "prompt_cache_options": { "mode": "explicit", "ttl": "30m" },
  "input": [
    {
      "type": "message",
      "role": "user",
      "content": [
        {
          "type": "input_text",
          "text": "Stable support policy and tool instructions...",
          "prompt_cache_breakpoint": { "mode": "explicit" }
        },
        {
          "type": "input_text",
          "text": "Current user question goes here."
        }
      ]
    }
  ]
}

def emit_usage(workflow: str, task_class: str, response):
    usage = response.usage
    input_details = getattr(usage, "input_tokens_details", None)
    output_details = getattr(usage, "output_tokens_details", None)
    telemetry.emit(
        workflow=workflow,
        task_class=task_class,
        model=response.model,
        status=response.status,
        input_tokens=usage.input_tokens,
        output_tokens=usage.output_tokens,
        total_tokens=usage.total_tokens,
        cached_tokens=getattr(input_details, "cached_tokens", 0),
        cache_write_tokens=getattr(input_details, "cache_write_tokens", 0),
        reasoning_tokens=getattr(output_details, "reasoning_tokens", 0),
    )

budget_kill_switch:
  workflow: pr-review
  monthly_cost_ceiling_usd: 300
  daily_cost_ceiling_usd: 25
  single_run_token_ceiling: 250000
  fallback_budget:
    max_retries: 1
    max_fallbacks_per_run: 1
  on_breach:
    action: halt_new_requests
    notify: platform-oncall
    preserve_partial_outputs: true

def guard_budget(workflow: str, projected_cost_usd: float):
    spent = usage_store.month_to_date_cost(workflow)
    ceiling = policy_store.monthly_ceiling(workflow)
    if spent + projected_cost_usd > ceiling:
        circuit_breaker.open(workflow)
        raise RuntimeError(f"budget ceiling exceeded for {workflow}")

{
  "type": "function",
  "name": "cancel_order",
  "description": "Cancel a customer order after explicit approval.",
  "parameters": {
    "type": "object",
    "properties": { "order_id": { "type": "number" } },
    "required": ["order_id"],
    "additionalProperties": false
  },
  "allowed_callers": ["direct"]
}

const cancelOrder = tool({
  name: "cancel_order",
  description: "Cancel a customer order.",
  parameters: z.object({ orderId: z.number() }),
  needsApproval: true,
  async execute({ orderId }) {
    return `Cancelled order ${orderId}`;
  },
});

let result = await run(agent, "Cancel order 123.");
if (result.interruptions?.length) {
  const state = result.state;
  for (const interruption of result.interruptions) {
    state.approve(interruption);
  }
  result = await run(agent, state);
}

<tool_orchestration>
Use Programmatic Tool Calling only for inventory-demand comparison.
Call get_inventory and get_demand concurrently.
Return exactly JSON with sku, available_units, requested_units, shortage_units.
Retry transient failures at most one time.
Do not call order-changing tools.
Use direct tool calls for approval before any inventory-changing action.
</tool_orchestration>

# 外部 orchestration 例
prep = client.responses.create(
    model="gpt-5.6-luna",
    reasoning={"effort": "low"},
    input="Extract candidate issues from these logs.",
)

draft = client.responses.create(
    model="gpt-5.6-terra",
    reasoning={"effort": "medium"},
    input=f"Draft remediation plan from candidates:\n{prep.output_text}",
)

final = client.responses.create(
    model="gpt-5.6-sol",
    reasoning={"effort": "high"},
    input=f"Review and risk-rank this plan:\n{draft.output_text}",
)

-- 例: workflow ごとの cache read/write 比を見る
select
  workflow,
  sum(cached_tokens) as cache_reads,
  sum(cache_write_tokens) as cache_writes,
  sum(cached_tokens) * 1.0 / nullif(sum(cache_write_tokens), 0) as read_write_ratio
from llm_usage
where model like 'gpt-5.6%'
group by workflow
order by read_write_ratio asc;

fallback_policy:
  allowed:
    - from: gpt-5.6-sol
      to: gpt-5.6-terra
      only_if: "reversible and not final approval"
    - from: gpt-5.6-terra
      to: gpt-5.6-luna
      only_if: "classification or preprocessing"
  denied:
    - task_class: "payment approval"
    - task_class: "security final signoff"
    - task_class: "data migration go no-go"