Assistants APIからResponses APIへ移行する実装手順【停止日・状態・ロールバック】

Assistants APIからResponses APIへの移行は、endpoint名の置換ではありません。最短の安全策は、新規会話だけをResponsesへ流し、instructionsをアプリのコードへ移し、会話状態とtool実行の責任を明示してから、既存Threadを必要時に扱うことです。

先に停止日を二つ区別してください。

• Assistants APIは2026年8月26日に停止します。
• reusable prompt objectsは2026年6月3日にdeprecatedとなり、/v1/promptsは2026年11月30日に停止します。

現行のAssistants移行ガイドにはAssistant → Promptという対応が残っていますが、長期運用する新規実装でPrompt objectを移行先にしてはいけません。instructionsはGitで版管理し、Responses呼び出し時にinstructionsとinputとして送ります。停止日はOpenAIの Assistants API deprecationと reusable prompts deprecationを別々に確認します。

この記事の対象は公式OpenAI Assistants APIからResponses APIへの移行です。Chat Completionsからの移行、Azure OpenAI、Yandex Cloud、第三者のOpenAI互換providerは、endpoint、API version、保存、tool対応が異なるため対象外です。

まず15分で移行の設計図を作る

既存コードから次の六つだけを抜き出します。

1. Assistantに保存したinstructions、model設定、tools
2. Threadを自社ユーザーやtenantへ結び付けるキー
3. Runの作成、polling、requires_action処理
4. Retrieval/File Searchのvector storeとmetadata filter
5. write系toolの重複防止、権限確認、監査ログ
6. 正常系・失敗系の代表リクエストと現在のtokens、latency、cost

対応先は次の通りです。

OpenAIの Assistants migration guideはobjectの差分確認に使えます。ただし、Prompt objectの寿命、運用上のrollback、toolの副作用までは自社の設計判断が必要です。

同じ配送照会をPythonでbefore / after比較する

比較する契約は一つです。入力は追跡番号JP-104の状況は？、toolは必須の文字列tracking_idを受け取るlookup_delivery、期待結果はJP-104を一度照会し、その結果だけで日本語回答を返すことです。unknown tool、不正な引数、元のcall IDを失ったoutput、side effectの重複、tenant混在、MAX_TOOL_ROUNDS内に最終回答がない場合は失敗です。

次の一つのPythonファイルは同じ入力、parameter schema、dispatcherを共有し、旧側のAssistant/Thread/Run/requires_actionと、新側のcode-managed instructions/application-managed state/Response/application tool loopを比較します。

python
import json
import os
from openai import OpenAI

client = OpenAI()
MODEL = os.environ["OPENAI_MODEL"]
ASSISTANT_ID = os.environ["OPENAI_ASSISTANT_ID"]
MAX_TOOL_ROUNDS = 5
USER_INPUT = "追跡番号JP-104の状況は？"

PARAMETERS = {
    "type": "object",
    "properties": {"tracking_id": {"type": "string"}},
    "required": ["tracking_id"],
    "additionalProperties": False,
}


def lookup_delivery(tracking_id: str, tenant_id: str) -> dict:
    return {
        "tracking_id": tracking_id,
        "status": "out_for_delivery",
        "tenant_id": tenant_id,
    }


def run_before() -> str:
    # OPENAI_ASSISTANT_ID側にも同じ名前とPARAMETERSを設定済みであること。
    completed_call_ids: set[str] = set()
    assistant = client.beta.assistants.retrieve(ASSISTANT_ID)
    matching_tools = [
        tool for tool in assistant.tools
        if tool.type == "function" and tool.function.name == "lookup_delivery"
    ]
    if (
        len(matching_tools) != 1
        or matching_tools[0].function.parameters != PARAMETERS
    ):
        raise RuntimeError("Stored Assistant tool schema differs from contract")

    thread = client.beta.threads.create()
    client.beta.threads.messages.create(
        thread_id=thread.id,
        role="user",
        content=USER_INPUT,
    )
    run = client.beta.threads.runs.create_and_poll(
        thread_id=thread.id,
        assistant_id=ASSISTANT_ID,
    )

    for _ in range(MAX_TOOL_ROUNDS):
        if run.status == "completed":
            messages = client.beta.threads.messages.list(
                thread_id=thread.id,
                limit=1,
            )
            return messages.data[0].content[0].text.value
        if run.status != "requires_action":
            raise RuntimeError(f"Legacy Run failed: {run.status}")

        outputs = []
        for call in run.required_action.submit_tool_outputs.tool_calls:
            if call.id in completed_call_ids:
                raise RuntimeError(f"Duplicate tool call: {call.id}")
            if call.function.name != "lookup_delivery":
                raise RuntimeError(f"Unknown tool: {call.function.name}")
            args = json.loads(call.function.arguments)
            result = lookup_delivery(args["tracking_id"], "tenant_demo")
            completed_call_ids.add(call.id)
            outputs.append({
                "tool_call_id": call.id,
                "output": json.dumps(result, ensure_ascii=False),
            })

        run = client.beta.threads.runs.submit_tool_outputs_and_poll(
            thread_id=thread.id,
            run_id=run.id,
            tool_outputs=outputs,
        )

    raise RuntimeError("Legacy Run exceeded MAX_TOOL_ROUNDS")


def run_after() -> str:
    tools = [{
        "type": "function",
        "name": "lookup_delivery",
        "description": "Return delivery status visible to the current tenant.",
        "parameters": PARAMETERS,
        "strict": True,
    }]
    items = [{"role": "user", "content": USER_INPUT}]
    completed_call_ids: set[str] = set()

    for _ in range(MAX_TOOL_ROUNDS):
        response = client.responses.create(
            model=MODEL,
            instructions=(
                "日本語で回答する。配送情報を推測せず、"
                "必要ならlookup_deliveryを使う。"
            ),
            tools=tools,
            input=items,
            store=False,
        )
        items.extend(response.output)
        calls = [x for x in response.output if x.type == "function_call"]
        if not calls:
            return response.output_text

        for call in calls:
            if call.call_id in completed_call_ids:
                raise RuntimeError(f"Duplicate tool call: {call.call_id}")
            if call.name != "lookup_delivery":
                raise RuntimeError(f"Unknown tool: {call.name}")
            args = json.loads(call.arguments)
            result = lookup_delivery(args["tracking_id"], "tenant_demo")
            completed_call_ids.add(call.call_id)
            items.append({
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": json.dumps(result, ensure_ascii=False),
            })

    raise RuntimeError("Responses loop exceeded MAX_TOOL_ROUNDS")


print({"before": run_before(), "after": run_after()})

この比較コードと後述のTypeScript実装は、同じ入力、schema、実行責任、最大ラウンド、期待結果と失敗条件を共有します。Pythonは旧新の対応確認、TypeScriptはResponses側の実装詳細を読むためのものです。

どちらも現在のOpenAI公式例に合わせたsource-aligned codeですが、このrunでは対象projectのcredentialsと読者の現在のSDKを使って実行していません。stagingでAssistant設定とPARAMETERSの一致、SDK resource、model/tool対応、実際のoutput itemを確認してから切り替えてください。

TypeScriptでRunを明示的なtool loopへ置き換える

次の例は、同じ配送照会契約をTypeScriptで展開した構成です。custom functionはOpenAI側ではなく自社backendが実行します。model名を固定せず、環境ごとにOPENAI_MODELを設定します。

ts
import OpenAI from "openai";

const openai = new OpenAI();
const model = process.env.OPENAI_MODEL;

if (!model) {
  throw new Error("OPENAI_MODEL is required");
}

const tools = [
  {
    type: "function" as const,
    name: "lookup_delivery",
    description: "Return delivery status visible to the current tenant.",
    parameters: {
      type: "object",
      properties: {
        tracking_id: { type: "string" },
      },
      required: ["tracking_id"],
      additionalProperties: false,
    },
    strict: true,
  },
];

async function lookupDelivery(trackingId: string, tenantId: string) {
  // Replace with a tenant-scoped, read-only database query.
  return {
    tracking_id: trackingId,
    status: "out_for_delivery",
    tenant_id: tenantId,
  };
}

export async function answerWithTools(
  userText: string,
  tenantId: string,
): Promise<string> {
  const inputItems: OpenAI.Responses.ResponseInput = [
    { role: "user", content: userText },
  ];
  const completedCallIds = new Set<string>();
  const maxToolRounds = 5;

  for (let round = 0; round < maxToolRounds; round += 1) {
    const response = await openai.responses.create({
      model,
      instructions:
        "日本語で回答する。配送情報を推測せず、必要ならlookup_deliveryを使う。",
      input: inputItems,
      tools,
      store: false,
    });

    // function_callだけでなく、reasoning関連を含む全output itemを保持する。
    inputItems.push(...response.output);
    const calls = response.output.filter(
      (item) => item.type === "function_call",
    );

    if (calls.length === 0) {
      return response.output_text;
    }

    for (const call of calls) {
      if (completedCallIds.has(call.call_id)) {
        throw new Error(`Duplicate tool call blocked: ${call.call_id}`);
      }

      const args = JSON.parse(call.arguments) as { tracking_id: string };
      const result =
        call.name === "lookup_delivery"
          ? await lookupDelivery(args.tracking_id, tenantId)
          : { error: `Unsupported tool: ${call.name}` };

      completedCallIds.add(call.call_id);
      inputItems.push({
        type: "function_call_output",
        call_id: call.call_id,
        output: JSON.stringify(result),
      });
    }
  }

  throw new Error("Tool loop exceeded maxToolRounds");
}

answerWithTools("追跡番号JP-104の状況は？", "tenant_demo")
  .then(console.log)
  .catch(console.error);

ここで外してはいけない点は四つです。

• call_idは変更せず、対応するfunction_call_outputへ返す。
• reasoning modelを使う場合も、function callだけを抜き出さずresponse.output全体を次のinputへ残す。
• instructionsは各Responseへ明示する。特にprevious_response_idでつないでも、前回のinstructionsは自動継承されない。
• write系toolではcall_idをDBのidempotency keyとして保存する。プロセス内のSetだけでは再起動やretryを防げない。

関数のschema、複数tool call、結果の返却方法は Function calling guideで確認できます。built-in tool向けの制限値だけに頼らず、custom tool loopにはアプリ独自の最大ラウンド、timeout、認可、監査を置きます。

会話状態は三択で決める

Responses APIでは、用途に合わせて状態の所有者を選べます。「新しいからConversation」と一律には決まりません。

1. Conversation

複数端末、長期サポート、非同期workerなど、同じ会話を継続して参照したい場合に向きます。Conversationのitemsは通常のResponse objectの30日保持とは別に扱われます。自社DBにはtenant_id、user_id、conversation_id、retention区分を一緒に保存し、別tenantのIDを受け入れないようにします。

python
import os
from openai import OpenAI

client = OpenAI()
conversation = client.conversations.create()

response = client.responses.create(
    model=os.environ["OPENAI_MODEL"],
    conversation=conversation.id,
    instructions="Reply in Japanese and never expose another tenant's data.",
    input="この会話の移行テストを開始してください。",
)

print(conversation.id)
print(response.output_text)

2. previous_response_id

短い直列会話で、直前のResponseへつなぐだけなら実装量を減らせます。ただしResponse objectsはデフォルトで30日保存され、チェーン内の過去input tokensも課金対象です。previous_response_idとconversationは同時に指定できません。

3. 自社DB + store=False

保存期間、削除、監査、移植性を自社で統制したい場合に向きます。上のTypeScript例のように必要なitemsを再送します。テキストだけを保存するとtoolやreasoningの文脈が壊れるため、APIへ戻す完全なitemsと、画面表示用の正規化データを分けて持つのが安全です。

保持期間と課金を含む仕様は Conversation state guideで確認してください。選定結果は「楽だから」ではなく、retention、delete、tenant isolation、復旧、token costの五項目で記録します。

三種類のtool実行責任を分ける

現在の MCP and Connectors guideではremote MCPはtool type mcpを使い、remote serverへデータを送る前にデフォルトでapprovalを求めます。custom function loopをそのままMCPへ移したり、serverと送信データを評価せずapprovalを無効化したりしないでください。

File Searchは検索品質まで移行する

Responsesのfile_searchはhosted toolです。custom functionと違い、アプリが検索を実行してfunction_call_outputを返す必要はありません。

python
import os
from openai import OpenAI

client = OpenAI()

response = client.responses.create(
    model=os.environ["OPENAI_MODEL"],
    instructions="登録文書だけを根拠に日本語で答え、出典を示す。",
    input="返品可能な期間を教えてください。",
    tools=[
        {
            "type": "file_search",
            "vector_store_ids": [os.environ["OPENAI_VECTOR_STORE_ID"]],
            "max_num_results": 8,
        }
    ],
    include=["file_search_call.results"],
)

print(response.output_text)
for item in response.output:
    if item.type == "file_search_call":
        print(item.results)

移行テストでは回答文だけでなく、次を採点します。

実際のrequest shapeと結果のinclude方法は File Search guideに合わせます。

既存Threadは一括変換しない

ThreadsからConversationsへの自動移行機能はありません。まず新規会話の入口だけをResponsesへ切り替え、既存Threadは開かれたときだけ処理するほうが、個人データと検証範囲を抑えられます。

既存Threadを再開するときは、次のいずれかをプロダクト要件として決めます。

• 旧会話はread-only表示し、新しいResponses会話を開始する。
• 直近の必要なmessagesだけを新しいinputへ変換する。
• 長い履歴は検証可能なsummaryと参照IDへ縮約する。
• 法的・監査上必要な場合だけ、全itemsを型ごとに変換する。

過去のtool resultを新しい「実行要求」として再送してはいけません。支払い、送信、登録などのside effectが二重に発生します。旧thread_idと新conversation_idまたはresponse_idの対応は監査用に残します。

Run pollingの移行先を分ける

対話UIならstreamingでdeltaを表示し、完了eventでResponse全体を確定します。接続が切れた場合に、表示だけを再開するのか処理を再実行するのかを区別します。

数分かかる生成など、接続を維持しない処理だけは background modeを候補にします。response idをjob tableへ保存し、状態をpollします。すべての旧Runをbackgroundへ置き換える必要はありません。

切替前の受け入れテスト

文章の一致率ではなく、タスクの結果を比べます。旧Assistants経路と新Responses経路に同じgolden setを通し、次を記録してください。

• 必須事実、禁止事項、構造化出力schema
• 呼ばれたtool、arguments、実行回数、最終結果
• File Searchが選んだ文書とcitation
• multi-turnで保持された情報と削除後の挙動
• input/output tokens、model call数、tool round数、scenario cost
• P50/P95 latency、timeout、4xx/5xx、空回答率

write toolを旧経路と新経路で同時実行するshadow testは危険です。比較時はread-only tool、dry-run、または副作用を発生させないrecorded resultを使います。

Feature flagで段階的に切り替える

推奨順序は、社内利用 → 1% → 5% → 25% → 50% → 100%です。各段階で最低観測件数と停止条件を先に決めます。

text
if new_session:
    route = feature_flag("responses_rollout", tenant_id)
    pin route to session
else:
    route = session.saved_route

一つの会話を途中でAssistantsとResponsesの間に往復させないでください。state format、tool result、retentionの意味が違うためです。

go条件

• golden setの必須ケースがすべて合格
• write side effectの重複がゼロ
• tenant越境がゼロ
• File Searchの根拠精度が基準以上
• P95、error rate、scenario costが事前予算内
• 監視画面から旧thread_idと新IDを追跡できる

rollback条件

• 認可漏れ、重複実行、誤った書き込みが一件でも発生
• 必須toolの欠落や不正argumentsが閾値超過
• 状態消失、別会話の混入、削除不能
• latency、5xx、costが連続してstop threshold超過

rollbackではfeature flagを戻し、新しく始まる会話を旧経路へ戻します。すでにResponsesで開始した会話は新経路へpinしたまま、停止または完了させます。Response/Conversationのデータを即時削除せず、retention policyに従って原因を調査します。

このrollback先としてAssistants APIを使えるのは2026年8月26日までです。停止後は閉じたendpointへ戻れません。したがってproduction cutoverは停止日の直前ではなく、問題を修正して再試験できる余裕を持って完了させます。

日本語情報の鮮度とprovider境界

日本語のHelp記事や自動翻訳ページに古い予定が残る場合があります。日付が衝突したら、現在の英語deprecationsとAPI referenceを基準にし、確認日をrunbookへ残します。2026年8月26日と2026年11月30日は別製品面の停止日です。

また、この手順は公式OpenAI API向けです。OpenAI互換のrequest bodyが通るだけでは、Conversation、File Search、background、stream event、保存期間の互換性は証明できません。Azure OpenAIなど別providerでは、そのproviderのendpoint、API version、region、data policyで別の受け入れ試験を実施します。

project key、API側の請求、最初のResponses呼び出しがまだなら、先に OpenAI APIキーの取得と動作確認を完了してください。ChatGPTの契約とAPIの利用条件は別です。

完了判定

移行完了はresponses.create()が200を返した時点ではありません。

• productionで新しいAssistant、Thread、Runを作成していない
• instructionsとtool schemaがコードレビュー対象になっている
• reusable Promptを新しい恒久依存にしていない
• 会話ごとに状態方式とrouteが固定されている
• tool loopに認可、idempotency、timeout、最大ラウンドがある
• File Searchを文書、filter、citation、空振りで検証した
• golden setと運用指標が合格し、rollback訓練も成功した
• 旧Assistants経路の停止とデータ保持・削除計画が決まっている

この条件がそろって初めて、Assistantsのobjectを置き換えたのではなく、Responses上で同じ業務結果を再現できたと判断できます。