# Chat Completions 함수 호출을 Responses API로 이전하는 Python 검증 절차

> Responses API 이전은 모든 함수 호출을 앱에서 검증·실행하고, 원래 call_id로 결과를 반환한 뒤 최종 답변까지 받아야 끝납니다. Python 비동기 예제로 전체 루프를 먼저 오프라인에서 확인하고, 업무 결과가 불명확하면 재실행을 멈춥니다.

- URL: https://blog.laozhang.ai/ko/posts/chat-completions-to-responses-api-function-calling-migration
- Published: 2026-07-28
- Updated: 2026-10-08
- Author: LaoZhang AI Team (https://blog.laozhang.ai/ko/about)
- Topic: API 가이드
- Tags: Responses API, Chat Completions, 함수 호출, call_id, Python

---
Chat Completions에서 이미 돌아가는 함수 호출을 Responses API로 이전하려면 **요청 형식과 도구 실행 루프를 함께 바꿔야 합니다.** 모델이 반환한 `function_call`은 실행 요청입니다. 앱이 함수명·인자·권한을 검사하고 실제 함수를 실행한 다음, 원래 `call_id`를 가진 `function_call_output`을 돌려줘야 모델이 결과를 바탕으로 답변할 수 있습니다. HTTP 요청 성공이나 “처리했습니다”라는 문장만으로 업무 완료를 판정하면 안 됩니다.

먼저 아래 Python 예제를 오프라인에서 실행해 보세요. 가상 반품 수거 요청 세 개를 받아 등록 두 건을 만들고, 각 호출의 결과를 반환한 뒤 최종 답변을 받는 전체 흐름입니다. 실제 OpenAI 클라이언트나 택배 서비스에 연결하지 않습니다. 기존 앱에 붙일 때는 transport와 업무 저장소를 교체하고, 마지막의 전환 확인표를 실제 환경에서 통과해야 합니다.

[OpenAI의 이전 가이드](https://developers.openai.com/api/docs/guides/migrate-to-responses)는 신규 프로젝트에 Responses를 권장하면서 Chat Completions도 계속 지원한다고 설명합니다. 기존 서비스를 한 번에 바꾸기보다 같은 업무의 호출·결과·복구 경로를 대조할 수 있는 단위로 이전하세요.

## 기존 코드에서 바꿔야 할 여섯 지점

함수 이름과 업무 규칙은 유지할 수 있지만, Chat 메시지를 Responses 입력에 그대로 섞을 수는 없습니다.

| Chat Completions에서 사용한 값 | Responses에서 사용할 값 | 앱에서 확인할 점 |
| --- | --- | --- |
| `client.chat.completions.create` | `client.responses.create` | 실제 연결 경로의 지원 여부 |
| `messages` | `input` | 문자열만 이어 붙이지 않고 관련 Item 보존 |
| `tools[].function` 안의 정의 | `tools[]`에 평탄하게 놓은 함수 정의 | `type`, `name`, `parameters`, `strict` |
| `choices[0].message.tool_calls` | `output` 전체의 `function_call` | 호출이 0개·1개·여러 개일 수 있음 |
| `role: "tool"`, `tool_call_id` | `function_call_output`, `call_id` | 반환된 원래 `call_id` 유지 |
| 첫 assistant 메시지의 텍스트 | 도구 결과 반환 후 최종 메시지 | 업무 결과와 최종 문장을 별도로 확인 |

[함수 호출 문서](https://developers.openai.com/api/docs/guides/function-calling#handling-function-calls)의 호출 Item에는 함수명, JSON 문자열인 `arguments`, `call_id`가 있습니다. Item 자체의 `id`는 다른 식별자입니다. 배열의 첫 요소가 텍스트라고 가정하거나, `id`를 `call_id` 대신 결과에 넣지 마세요.

아래 함수는 `strict: True`를 명시하고 객체의 모든 속성을 `required`에 넣으며 `additionalProperties: False`로 닫습니다. 현재 Responses에서 **strict를 생략하면 가능한 스키마는 strict로 정규화하고, 호환되지 않으면 엄격한 스키마 보장 없이 호출하는 방식으로 돌아갑니다.** 명시적으로 `strict: True`를 보냈는데 요구사항이 맞지 않으면 거절됩니다. 항상 strict가 된다고 가정하는 것도, 생략하면 무조건 오류라고 설명하는 것도 부정확합니다. [현재 strict 규칙](https://developers.openai.com/api/docs/guides/function-calling#strict-mode)을 기준으로 결정하세요.

strict는 인자의 형태를 제한할 뿐 주문 소유권이나 수거 가능 여부를 보증하지 않습니다. 함수 실행 없이 최종 JSON만 필요하다면 [Structured Outputs와 Function Calling의 선택 기준](https://blog.laozhang.ai/ko/posts/structured-outputs-vs-function-calling)을 먼저 확인하세요.

## Python 비동기 예제: 호출 세 개, 가상 수거 등록 두 건

다음 전체 코드를 `pickup_loop.py`로 저장하고 Python 3.10 이상에서 실행합니다. 표준 라이브러리만 사용하므로 SDK 설치나 API 키가 필요 없습니다.

```bash
python3 pickup_loop.py
```

`Session`의 사용자 범위·요청 키·완료 대상은 인증된 앱이 정하는 값입니다. 모델에게 사용자나 업무 권한을 선택하게 하지 않습니다. 예제 날짜와 주문 번호는 가상 데이터이며, 동시 실행 2개·업무 제한 2초·전체 제한 30초·최대 6라운드는 이 예제를 위한 값입니다.

```python
import asyncio
import copy
import json
from dataclasses import dataclass, field

TOOLS = [{
    "type": "function", "name": "register_return_pickup",
    "description": "Register a pickup for an authorized return order.",
    "strict": True,
    "parameters": {
        "type": "object",
        "properties": {
            "order_id": {"type": "string"},
            "pickup_date": {"type": "string", "enum": ["2026-10-12", "2026-10-13"]},
        },
        "required": ["order_id", "pickup_date"],
        "additionalProperties": False,
    },
}]
INSTRUCTIONS = (
    "반품 수거를 등록하려면 register_return_pickup을 호출하세요. "
    "도구 결과에 없는 등록 상태를 만들지 마세요. "
    "ok=false이면 자동 재시도하지 말고 오류를 설명하세요."
)

class StopLoop(RuntimeError):
    pass

@dataclass
class Session:
    # 인증된 앱이 결정하는 값이며 모델 인자가 아닙니다.
    tenant: str = "demo-shop"
    request_key: str = "return-request-44"
    allowed_orders: frozenset = frozenset({"ORD-7391", "ORD-7392"})
    history: list = field(default_factory=lambda: [{
        "role": "user", "content": "ORD-7391과 ORD-7392를 10월 13일에 수거해 주세요."
    }])
    expected_orders: frozenset = frozenset({"ORD-7391", "ORD-7392"})
    calls: dict = field(default_factory=dict)
    jobs: dict = field(default_factory=dict)
    registrations: dict = field(default_factory=dict)
    receipts: dict = field(default_factory=dict)
    limit: asyncio.Semaphore = field(default_factory=lambda: asyncio.Semaphore(2))


def parse_args(raw, session):
    def unique_keys(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError("duplicate_json_key")
            value[key] = item
        return value
    value = json.loads(raw, object_pairs_hook=unique_keys)
    if not isinstance(value, dict) or set(value) != {"order_id", "pickup_date"}:
        raise ValueError("invalid_fields")
    if not all(isinstance(v, str) for v in value.values()):
        raise ValueError("invalid_type")
    if value["order_id"] not in session.allowed_orders:
        raise ValueError("order_not_authorized")
    if value["pickup_date"] not in {"2026-10-12", "2026-10-13"}:
        raise ValueError("date_not_available")
    return value


async def register_return_pickup(args, session):
    # 가상 업무 저장소입니다. 외부 택배 서비스에는 연결하지 않습니다.
    await asyncio.sleep(0)
    key = (session.tenant, session.request_key, args["order_id"])
    receipt = {"pickup_id": "demo-" + args["order_id"], **args, "status": "registered"}
    session.registrations[key] = receipt
    return receipt

HANDLERS = {"register_return_pickup": register_return_pickup}


async def run_job(handler, args, session):
    try:
        async with session.limit:
            result = await asyncio.wait_for(handler(args, session), timeout=2)
        return {"ok": True, "data": result}
    except asyncio.CancelledError:
        raise
    except Exception:
        # 시간 초과·연결 오류만으로 외부 쓰기 실패를 확정할 수 없습니다.
        return {"ok": False, "error": "business_outcome_unknown"}


async def dispatch(call, session):
    cid = call["call_id"]
    handler = HANDLERS.get(call["name"])
    if handler is None:
        payload = {"ok": False, "error": "function_not_allowed"}
    else:
        try:
            args = parse_args(call["arguments"], session)
        except (ValueError, TypeError):
            payload = {"ok": False, "error": "arguments_rejected"}
        else:
            key = (session.tenant, session.request_key, args["order_id"])
            signature = json.dumps(args, sort_keys=True)
            if key not in session.jobs:
                # await 전에 pending 작업을 등록해 동시 중복 실행을 막습니다.
                session.jobs[key] = (signature, asyncio.create_task(run_job(handler, args, session)))
            saved_signature, task = session.jobs[key]
            if saved_signature != signature:
                payload = {"ok": False, "error": "operation_key_conflict"}
            else:
                payload = await asyncio.shield(task)
    session.receipts[cid] = payload
    return {"type": "function_call_output", "call_id": cid,
            "output": json.dumps(payload, ensure_ascii=False)}


def inspect_output(response, session):
    if response.get("status") != "completed" or response.get("error"):
        raise StopLoop("response_not_completed: " + str(response.get("incomplete_details")))
    items = response.get("output")
    if not isinstance(items, list):
        raise StopLoop("missing_output_array")
    fresh, signatures = {}, dict(session.calls)
    texts, refused = [], False
    for item in items:
        kind = item.get("type")
        if kind not in {"message", "reasoning", "function_call"}:
            raise StopLoop("unsupported_item: " + str(kind))
        if kind == "function_call":
            cid = item.get("call_id")
            if not isinstance(cid, str) or not cid:
                raise StopLoop("missing_call_id")
            if not isinstance(item.get("name"), str) or not isinstance(item.get("arguments"), str):
                raise StopLoop("malformed_call")
            if (item.get("status") not in {None, "completed"}
                or item.get("async") is True or item.get("namespace")
                or item.get("caller") not in (None, {"type": "direct"})):
                raise StopLoop("unsupported_call_mode")
            sig = (item["name"], item["arguments"])
            if cid in signatures and signatures[cid] != sig:
                raise StopLoop("call_id_conflict")
            if cid not in signatures:
                fresh[cid] = item
            signatures[cid] = sig
        elif kind == "message":
            if item.get("status") not in {None, "completed"}:
                raise StopLoop("partial_message")
            for part in item.get("content", []):
                refused |= part.get("type") == "refusal"
                if part.get("type") == "output_text" and item.get("phase") != "commentary":
                    texts.append(part["text"])
    if refused:
        raise StopLoop("model_refusal")
    if not fresh and any(i.get("type") == "function_call" for i in items):
        raise StopLoop("duplicate_call_without_new_progress")
    session.calls = signatures
    return list(fresh.values()), "".join(texts).strip()


async def migrate(create, model, session, reasoning=None):
    deadline = asyncio.get_running_loop().time() + 30
    try:
        for _ in range(6):
            remaining = deadline - asyncio.get_running_loop().time()
            if remaining <= 0:
                raise StopLoop("deadline_exceeded")
            request = {"model": model, "instructions": INSTRUCTIONS,
                       "input": copy.deepcopy(session.history), "tools": TOOLS,
                       "parallel_tool_calls": True, "store": False,
                       "max_output_tokens": 1200}
            if reasoning is not None:
                request["reasoning"] = reasoning
            response = await asyncio.wait_for(create(**request), timeout=remaining)
            calls, text = inspect_output(response, session)
            # 원본 Item 전체를 보존합니다. phase와 encrypted_content도 유지됩니다.
            session.history.extend(copy.deepcopy(response["output"]))
            if not calls:
                if not text:
                    raise StopLoop("final_text_missing")
                if not session.receipts:
                    return {"status": "no_action", "text": text, "receipts": {}}
                status = "tool_failed"
                if all(r["ok"] for r in session.receipts.values()):
                    actual = {r["data"]["order_id"] for r in session.receipts.values()}
                    status = "done" if actual == session.expected_orders else "task_incomplete"
                return {"status": status, "text": text, "receipts": session.receipts}
            outputs = await asyncio.wait_for(
                asyncio.gather(*(dispatch(c, session) for c in calls)),
                timeout=max(0.001, deadline - asyncio.get_running_loop().time()))
            session.history.extend(outputs)
            if any(r.get("error") == "business_outcome_unknown" for r in session.receipts.values()):
                raise StopLoop("business_outcome_unknown: reconcile before resuming")
        raise StopLoop("round_limit_exceeded")
    except (asyncio.TimeoutError, asyncio.CancelledError):
        # 취소된 네트워크 요청 또는 남은 업무의 결과를 먼저 대조해야 합니다.
        for _, task in session.jobs.values():
            if not task.done():
                task.cancel()
        await asyncio.gather(*(task for _, task in session.jobs.values()), return_exceptions=True)
        raise


class FakeResponses:
    def __init__(self):
        self.requests = []

    async def create(self, **request):
        self.requests.append(copy.deepcopy(request))
        if len(self.requests) == 1:
            calls = [{"type": "function_call", "id": "fc-" + str(i),
                      "call_id": "call-" + str(i), "name": "register_return_pickup",
                      "arguments": json.dumps({"order_id": order, "pickup_date": "2026-10-13"}),
                      "status": "completed"}
                     for i, order in enumerate(["ORD-7391", "ORD-7392", "ORD-7391"])]
            return {"status": "completed", "output": [
                {"type": "reasoning", "id": "rs-demo", "summary": [], "encrypted_content": "synthetic"},
                {"type": "message", "role": "assistant", "phase": "commentary",
                 "content": [{"type": "output_text", "text": "수거 가능 여부를 확인하겠습니다."}]},
                *calls]}
        outputs = [i for i in request["input"] if i.get("type") == "function_call_output"]
        assert {i["call_id"] for i in outputs} == {"call-0", "call-1", "call-2"}
        assert all(json.loads(i["output"])["ok"] for i in outputs)
        return {"status": "completed", "output": [{
            "type": "message", "role": "assistant", "phase": "final_answer",
            "content": [{"type": "output_text", "text": "두 주문의 10월 13일 수거가 등록되었습니다."}]}]}


async def offline_demo():
    fake, session = FakeResponses(), Session()
    result = await migrate(fake.create, "offline-fixture", session)
    assert result["status"] == "done"
    assert len(session.registrations) == 2  # 원본 call_id는 세 개, 가상 등록은 두 건
    assert fake.requests[1]["input"][1]["encrypted_content"] == "synthetic"
    assert fake.requests[1]["input"][2]["phase"] == "commentary"
    print(json.dumps(result, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    asyncio.run(offline_demo())
```

출력의 `status`가 `done`이고 세 `call_id`의 receipt가 모두 성공이면, 가상 응답에 대한 앱 로직이 끝난 것입니다. `call-0`과 `call-2`는 같은 주문·요청을 가리키므로 가상 등록을 공유하지만, 도구 결과는 각 원래 ID로 따로 반환합니다. 최종 완료 판단에는 요청한 두 주문의 실제 receipt도 필요합니다. 하나만 등록했는데 모델이 두 건을 처리했다고 말해도 `task_incomplete`가 됩니다.

![가상 예제의 호출 세 개가 업무 키에 따라 수거 등록 두 건을 공유하고 원래 call_id별로 결과를 반환하는 관계도](https://blog.laozhang.ai/posts/ko/chat-completions-to-responses-api-function-calling-migration/img/call-identity-and-business-records.webp)

### 오류를 최종 성공으로 바꾸지 않는 방법

함수명 허용 목록과 인자 검사를 통과하지 못하면 handler를 실행하지 않고 `ok: false`를 돌려줍니다. 모델이 이 오류를 설명할 수는 있지만 결과 상태는 `tool_failed`입니다. 함수 호출 없이 최종 문장만 온 경우도 `no_action`으로 구분합니다.

`failed`, `incomplete`, `queued`, `in_progress`, `cancelled` 응답은 응답 완료를 기다리는 이 루프에서 실행 대상으로 받지 않습니다. `call_id` 누락·충돌, 지원하지 않는 Item, 미완료 호출, 거절 응답, 최종 텍스트 누락도 중단 조건입니다. [Responses 상태·스트리밍 문서](https://developers.openai.com/api/docs/guides/streaming-responses)에 맞춰 오류와 부분 응답을 성공 경로와 분리한 앱 정책입니다.

이 예제는 일반 사용자 정의 함수만 다룹니다. Python에서 `asyncio`를 사용한다는 이유로 API의 별도 비동기 도구 기능을 구현한 것은 아닙니다. `async: true`, namespace, program caller 같은 다른 호출 방식은 명시적으로 중단합니다. 도입하려면 그 기능의 별도 실행·재개 계약이 필요합니다.

### 메모리 중복 방지와 영속 멱등성은 다릅니다

`session.jobs`는 handler를 기다리기 **전에** pending task를 등록합니다. 같은 프로세스에서 두 호출이 동시에 들어와도 같은 업무 키에 대해 handler를 두 번 시작하지 않습니다. 동일한 업무 키에 다른 날짜를 넣으면 기존 결과를 덮어쓰지 않고 `operation_key_conflict`를 반환합니다.

하지만 프로세스가 재시작되면 이 딕셔너리는 사라집니다. 실제 수거 등록은 `(tenant, request_key, order_id)` 같은 안정적인 업무 키를 DB의 unique constraint로 관리하고, 키 확인·업무 변경·결과 기록을 같은 트랜잭션에 넣어야 합니다. 외부 택배 시스템이라면 영속 outbox와 해당 시스템의 멱등 키, 결과 조회·대조 절차가 필요합니다. 이 설계는 앱에서 구현할 사항이며 OpenAI가 제공하는 정확히 한 번 실행 보장이 아닙니다.

원래 `call_id`는 모델 호출과 도구 결과를 연결합니다. 새 모델 요청에서 ID가 달라져도 같은 반품 요청일 수 있으므로 비즈니스 키로 대신 사용할 수 없습니다. `parallel_tool_calls: False`로 바꾸는 것도 중복 쓰기 방지책이 아닙니다. 그 설정은 한 턴에 호출을 0개 또는 1개로 제한합니다. [`tool_choice: "required"`도 호출 요청을 요구할 뿐](https://developers.openai.com/api/docs/guides/function-calling#tool-choice) 실행 성공을 보증하지 않습니다.

handler 예외·시간 초과 때는 `business_outcome_unknown`으로 멈춥니다. 응답이 없다는 이유만으로 택배 시스템에 등록이 안 됐다고 단정할 수 없기 때문입니다. 외부 상태와 영속 기록을 확인하기 전에는 같은 쓰기를 자동으로 재실행하지 마세요.

## 기존 Responses 클라이언트에 연결하기

`migrate`는 비동기 `create(**request)`가 전체 응답 딕셔너리를 반환하도록 받습니다. 이미 앱에서 관리하는 Python Responses 클라이언트가 있다면 다음 adapter로 연결할 수 있습니다. 클라이언트 생성, 인증 설정과 실행 호출은 기존 앱에서 맡습니다.

```python
from pickup_loop import migrate, Session

async def run_with_existing_client(client, model, session, reasoning=None):
    async def create(**request):
        response = await client.responses.create(**request)
        return response.model_dump(exclude_unset=True)
    return await migrate(create, model, session, reasoning=reasoning)
```

이 adapter는 문서 형태에 맞춘 연결 예시이며, 이 글에서는 실제 SDK의 수락이나 서버 응답을 검증하지 않았습니다. 기존 앱의 SDK 버전·모델·route로 요청과 replay를 확인해야 합니다. `output_text`는 Python SDK가 텍스트를 모아 주는 편의 속성이지, raw REST의 전체 상태를 대신하는 필드가 아닙니다. 위 코드는 전체 `output`을 보존하고 최종 메시지의 텍스트를 별도로 추출합니다.

이전 과정에서 모델까지 자동으로 바꾸지 마세요. [현재 모델별 API·매개변수 호환성](https://developers.openai.com/api/docs/guides/latest-model#gpt-6-astra-update-api-and-model-parameters)에 따르면 GPT-6 Astra와 GPT-6.1 Sol의 함수 호출에는 Responses가 필요합니다. GPT-6 Sol·Luna의 Chat 함수 호출에는 `reasoning_effort: none` 조건이 있습니다. Responses에서는 `reasoning.effort`를 사용하며, Astra·6.1 Sol은 `none`을 지원하지 않습니다. reasoning이 `none`이 아닌 설정에 기존 sampling·logprob 옵션을 그대로 넘기지도 마세요. 기존 모델과 실제 effort를 보존할 수 있는 호환 조합부터 확인합니다.

## 다음 요청에 어떤 상태를 남길 것인가

위 코드는 수동 replay를 사용합니다. `response.output` 전체와 이번 배치의 도구 결과를 다음 `input`에 추가하며, 매번 `instructions`를 보냅니다. 텍스트 메시지만 추출해 이력을 다시 만들면 reasoning Item과 assistant 메시지의 `phase`를 잃을 수 있습니다. [현재 상태 관리 문서](https://developers.openai.com/api/docs/guides/conversation-state)는 암호화된 reasoning Item을 기본 반환하며 관련 전체 Item의 보존을 안내합니다. `commentary` 메시지는 이력에 남지만 이 예제의 최종 답변으로는 사용하지 않습니다.

다른 상태 방식은 보관·복구 조건에 따라 선택합니다.

| 방식 | 다음 요청 | 놓치기 쉬운 조건 |
| --- | --- | --- |
| 수동 replay | 관련 전체 이력과 새 도구 결과 | 앱이 이력을 보관하고 문맥 크기를 관리 |
| `previous_response_id` | 이전 응답 ID와 새 결과 | 이전 top-level instructions는 승계되지 않음 |
| `conversation` | 지속 대화 ID와 새 입력 | 완료 후 입력·출력이 자동 추가되며 보관 정책 별도 |

[`previous_response_id`와 `conversation`은 같은 요청에 함께 지정할 수 없습니다](https://developers.openai.com/api/reference/resources/responses/methods/create). 이미 ID로 참조하는 이력까지 전체로 중복 보내지 마세요. `previous_response_id`는 HTTP에서도 사용하며, WebSocket 연결 내부 캐시와는 복구 조건이 다릅니다. 요청 payload가 작아져도 이전 문맥의 입력 토큰 비용이 없어지는 것은 아닙니다.

`store=False`는 일반 Response 저장을 끄는 옵션입니다. 조직을 Zero Data Retention 대상으로 만들거나 앱·중간 제공자의 로그까지 지우지 않습니다. 일반 저장 응답의 기본 30일 보관과 달리 Conversation Item에는 같은 30일 TTL이 적용되지 않습니다. [데이터 보관 문서](https://developers.openai.com/api/docs/guides/your-data)에서 조직의 승인된 데이터 제어와 실제 사용하는 기능을 함께 확인하세요.

## 스트리밍: arguments.done만 보고 함수를 실행하지 않기

스트리밍에서도 원래 `call_id`는 실제 호출 Item에서 가져옵니다. [`response.function_call_arguments.done`의 현재 스키마](https://developers.openai.com/api/reference/resources/responses/streaming-events#response.function_call_arguments.done)에는 `arguments`, `item_id`, `output_index`, `sequence_number`, `type`이 있으며 전체 Item이나 `call_id`가 들어 있다고 가정하면 안 됩니다.

다음 collector는 보수적으로 **전체 `response.completed`까지 모은 뒤** 동일한 `migrate` 루프에 응답을 넘깁니다. delta는 실행하지 않고, 인자 완료·Item 완료·전체 응답 완료를 따로 확인합니다. `response.output_item.done`의 [완료 Item](https://developers.openai.com/api/reference/resources/responses/streaming-events#response.output_item.done)과 최종 응답이 다르면 중단합니다.

`stream_collect.py`로 저장하세요. 같은 디렉터리의 `pickup_loop.py`에서 오류 클래스를 가져오며, 입력 `events`는 JSON 딕셔너리를 내보내는 비동기 iterator입니다. SDK 이벤트 객체를 사용한다면 먼저 해당 버전의 직렬화 방식으로 변환해야 합니다.

```python
import json
from pickup_loop import StopLoop


async def collect_stream(events):
    seen, slots = {}, {}
    async for event in events:
        seq = event["sequence_number"]
        encoded = json.dumps(event, sort_keys=True)
        if seq in seen:
            if seen[seq] != encoded:
                raise StopLoop("event_sequence_conflict")
            continue
        seen[seq] = encoded
        kind = event["type"]
        if kind in {"error", "response.failed", "response.incomplete"}:
            raise StopLoop("stream_failed: " + kind)
        if kind == "response.output_item.added" and event["item"]["type"] == "function_call":
            item = event["item"]
            index = event["output_index"]
            if index in slots:
                raise StopLoop("duplicate_stream_slot")
            slots[index] = {"item": item, "buf": item["arguments"], "args_done": False, "done": False}
        elif kind in {"response.function_call_arguments.delta", "response.function_call_arguments.done"}:
            slot = slots.get(event["output_index"])
            if slot is None or slot["item"]["id"] != event["item_id"]:
                raise StopLoop("unmatched_arguments")
            if slot["args_done"]:
                raise StopLoop("arguments_after_done")
            if kind.endswith(".delta"):
                slot["buf"] += event["delta"]
            else:
                if slot["buf"] != event["arguments"]:
                    raise StopLoop("arguments_mismatch")
                slot["args_done"] = True
        elif kind == "response.output_item.done" and event["item"]["type"] == "function_call":
            slot = slots.get(event["output_index"])
            item = event["item"]
            if (slot is None or not slot["args_done"] or slot["done"]
                or any(item.get(k) != slot["item"].get(k) for k in ("id", "call_id", "name"))
                or item["arguments"] != slot["buf"]):
                raise StopLoop("completed_item_mismatch")
            slot["item"], slot["done"] = item, True
        elif kind == "response.completed":
            response = event["response"]
            if response.get("status") != "completed" or response.get("error"):
                raise StopLoop("terminal_response_failed")
            final_calls = {i: item for i, item in enumerate(response["output"])
                           if item["type"] == "function_call"}
            if set(final_calls) != set(slots):
                raise StopLoop("terminal_call_set_mismatch")
            for index, item in final_calls.items():
                if not slots[index]["done"] or slots[index]["item"] != item:
                    raise StopLoop("terminal_item_mismatch")
            return response  # reasoning·message·phase도 포함한 최종 전체 응답
    raise StopLoop("stream_ended_without_completed")
```

stream transport의 `create`가 `await collect_stream(events)`를 반환하면 비스트리밍과 같은 검증·실행·결과 반환 루프를 사용할 수 있습니다. 네트워크 이벤트 대기에도 deadline을 적용해야 합니다. 위 `migrate`는 transport 전체를 `asyncio.wait_for`로 감싸므로 스트리밍 완료를 영원히 기다리지 않습니다.

이 collector의 중복 이벤트 검사와 완료 후 실행 정책은 앱 구현 예시입니다. 서버가 끊긴 스트림을 자동 복구하거나 이벤트 재전송을 보장한다는 뜻이 아닙니다. `failed`, `incomplete`, 오류 이벤트 또는 완료 이벤트 없는 EOF에서는 handler를 실행하지 않습니다. 업무 실행 후 최종 답변을 받다가 끊겼다면 아래 복구 경로로 넘어갑니다.

![전체 응답 완료 후 실행하는 스트리밍 예제와 업무 실행·결과 확인 상태에 따른 세 가지 복구 경로를 설명한 개념도](https://blog.laozhang.ai/posts/ko/chat-completions-to-responses-api-function-calling-migration/img/stream-completion-and-recovery.webp)

## 최종 답변이 끊겨도 수거를 다시 등록하지 않기

복구할 때 먼저 마지막으로 확인된 단계부터 구분합니다.

1. **완료된 호출 Item을 받기 전:** 부분 JSON을 실행하지 않았는지 확인합니다. 실행이 없었다면 저장한 입력에서 모델 요청을 다시 시작할 수 있습니다.
2. **업무 실행 중 결과가 불명확:** 영속 업무 키로 택배 시스템과 결과 기록을 조회합니다. 등록됐는지 확인 전에는 쓰기를 다시 하지 않습니다.
3. **업무 결과는 확정됐지만 모델 답변이 없음:** 저장해 둔 원본 호출·관련 전체 Item·각 `function_call_output`으로 이어갑니다. 복구 요청의 목적은 이미 확정된 결과의 설명입니다.

모델 요청 retry와 업무 처리 retry는 별도 정책입니다. 예제의 메모리 `Session`만 저장했다고 프로세스 장애 뒤 복구가 된다고 주장할 수 없습니다. 운영에서는 인증 범위, 업무 키, 원래 호출과 출력, 확정·불명확 상태를 영속 저장하되 보관 정책에 맞게 민감 정보를 줄입니다.

`max_output_tokens`는 reasoning과 사용자에게 보이는 출력을 함께 제한합니다. [`max_tool_calls`는 내장 도구 호출 제한](https://developers.openai.com/api/reference/resources/responses/methods/create)이며 위 사용자 정의 handler의 라운드 수나 업무 처리 비용을 제한하지 않습니다. 앱의 deadline·동시 실행 수·최대 라운드·업무 결과 기록을 별도로 관리해야 합니다.

## 운영 전환 전에 대조할 결과

가상 예제에서는 여러 호출, 같은 업무의 중복, 인자·권한 거절, ID 충돌, 응답 실패, 빈 답변, 합성 스트림의 EOF·미완료·중복 이벤트를 오프라인으로 확인했습니다. 이는 앱 분기 검증이며 실제 모델 응답, SDK 수락, 외부 수거 등록이나 재시작 후 멱등성의 증거가 아닙니다.

실제 전환에서는 동일한 인증 범위와 고정 업무 키를 사용해 기존 경로와 새 경로를 대조하세요. 실제 쓰기를 두 경로에서 중복 실행하는 대신, 테스트 저장소·읽기 전용 대조·승인된 제한 실행 중 업무에 맞는 방법을 선택합니다.

| 확인 항목 | 통과 판단 | 통과하지 못했을 때 |
| --- | --- | --- |
| 요청·모델 호환 | 실제 route가 flat 함수 스키마와 해당 모델 설정을 수락 | 전환을 멈추고 endpoint·설정을 확인 |
| 실행 권한 | 잘못된 주문·함수·인자가 handler까지 가지 않음 | 앱 검증과 인증 범위 수정 |
| 호출 연결 | 모든 호출의 결과가 원래 `call_id`에 연결 | 임의 ID 생성 없이 원본을 복구 |
| 여러 호출 | 배치 전체 결과가 다음 요청 전에 모임 | collector·동시 실행 정책 수정 |
| 업무 중복 | 재시작·중복 요청 뒤에도 같은 업무 키의 확정 등록 한 건 | 영속 키·트랜잭션·대조 절차부터 수정 |
| 실제 완료 | 필요한 주문 receipt와 최종 답변 모두 확인 | 부분 업무·모델 실패를 별도 상태로 남김 |
| 스트리밍 | 완료·실패·EOF와 취소를 구분하고 부분 인자 미실행 | 비스트리밍 경로를 유지 |
| 복구·롤백 | 기존 확정 업무를 다시 쓰지 않고 답변·상태 복구 | 전환 확대 중단 |

Azure나 다른 호환 제공자를 사용하면 base URL, 인증, deployment·모델 ID, 보관, 상태 연결과 이벤트 지원을 해당 제공자의 계약으로 확인해야 합니다. OpenAI 형식의 요청이나 HTTP 200만으로 모든 기능의 호환성을 판정할 수 없습니다.

롤백도 엔드포인트 플래그 하나로 끝나지 않습니다. 미리 확인한 **모델·엔드포인트·매개변수·상태·업무 결과**의 조합으로 돌아갈 수 있어야 합니다. Responses 전용 함수 호출 모델을 그대로 Chat으로 돌리는 것은 동작하는 롤백이 아닙니다. 이미 완료한 수거의 업무 키와 결과 기록도 전환 중 유지하세요.

## 자주 묻는 질문

### 함수 결과만 받으면 이전이 끝난 건가요?

아닙니다. 앱에서 업무 결과가 확정되고, 각 호출에 같은 `call_id`의 결과를 반환하며, 사용자가 받을 최종 답변까지 확인해야 합니다. 실패 설명이 자연스러워도 업무 성공으로 처리하지 않습니다.

### parallel_tool_calls를 끄면 중복 등록을 막을 수 있나요?

아닙니다. [이 옵션은 한 턴의 호출을 0개 또는 1개로 제한합니다](https://developers.openai.com/api/docs/guides/function-calling#parallel-function-calling). 네트워크 retry·모델 재요청·프로세스 재시작에 따른 중복 업무는 영속 키와 업무 트랜잭션으로 처리해야 합니다.

### store=False면 이전 reasoning과 phase를 빼도 되나요?

아닙니다. 수동 replay에서는 관련 전체 출력 Item을 유지해야 합니다. [현재 상태 관리 가이드](https://developers.openai.com/api/docs/guides/conversation-state)는 암호화된 reasoning과 assistant `phase`를 보존하도록 안내합니다. 저장 옵션과 다음 요청에 필요한 문맥은 별개입니다.

### Chat Completions를 반드시 중단해야 하나요?

[Chat Completions는 계속 지원됩니다](https://developers.openai.com/api/docs/guides/migrate-to-responses). 다만 사용하는 모델의 함수 호출 지원은 별도로 확인해야 합니다. 기존 앱이 호환 모델에서 정상 동작한다면, 새 루프와 복구 절차가 확인될 때까지 기존 경로를 유지할 수 있습니다.

## 참고 자료

이 글이 링크한 외부 페이지를 본문에 나온 순서대로 정리했습니다. 마지막 업데이트: 2026-10-08.

- [OpenAI의 이전 가이드](https://developers.openai.com/api/docs/guides/migrate-to-responses) (developers.openai.com)
- [함수 호출 문서](https://developers.openai.com/api/docs/guides/function-calling) (developers.openai.com)
- [Responses 상태·스트리밍 문서](https://developers.openai.com/api/docs/guides/streaming-responses) (developers.openai.com)
- [현재 모델별 API·매개변수 호환성](https://developers.openai.com/api/docs/guides/latest-model) (developers.openai.com)
- [현재 상태 관리 문서](https://developers.openai.com/api/docs/guides/conversation-state) (developers.openai.com)
- [previous_response_id와 conversation은 같은 요청에 함께 지정할 수 없습니다](https://developers.openai.com/api/reference/resources/responses/methods/create) (developers.openai.com)
- [데이터 보관 문서](https://developers.openai.com/api/docs/guides/your-data) (developers.openai.com)
- [response.function_call_arguments.done의 현재 스키마](https://developers.openai.com/api/reference/resources/responses/streaming-events) (developers.openai.com)