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

Chat Completions의 함수 호출(Function Calling)을 Responses API로 옮길 때 완료 기준은 URL 변경이 아니다. response.output의 모든 function_call을 앱이 찾아서 허용 목록과 인자를 검증하고, 업무 처리를 한 번만 실행하고, 각각의 원래 call_id로 function_call_output을 반환한 뒤 최종 사용자 답변을 받아야 이전이 끝난다.

운영 코드를 건드리기 전에 같은 요청을 두 경로에서 비교할 staging fixture를 만든다. 기존 경로의 함수명·인자·업무 결과·부작용 횟수·최종 문장을 baseline으로 저장하고, Responses 경로가 동일한 의미를 재현하는지 확인한다. provider가 /v1/responses, typed Item 또는 tool result 반환을 지원한다는 실제 증거가 없다면 그 지점에서 중단한다.

15분 리허설에서 먼저 증명할 것

다음 네 가지를 한 trace에서 볼 수 있어야 한다.

1. 모델이 반환한 호출과 앱이 돌려준 결과가 같은 call_id로 연결된다.
2. 함수명과 JSON이 잘못되면 handler를 실행하지 않고 구조화 오류를 돌려준다.
3. 한 라운드의 call이 여러 개면 모두 끝난 뒤 다음 Responses 요청을 보낸다.
4. 네트워크 retry가 발생해도 동일한 반품 수거 요청은 한 번만 등록된다.

아래 하네스는 국내 전자상거래에서 흔한 반품 수거 등록을 예로 든다. 주소나 결제 정보는 모델에 주지 않고, 이미 검증된 주문 ID와 수거일만 도구 인자로 받는다. 실제 운영에서는 사용자 권한과 주문 상태를 handler 내부에서 다시 확인해야 한다.

비동기 Python 하네스: 반품 수거를 한 번만 등록하기

bash
python -m pip install -U openai
export OPENAI_API_KEY="project-api-key"
export OPENAI_MODEL="your-tool-capable-model"
export REQUEST_KEY="return-ORD-7391"
python migrate_return_pickup.py

다음을 migrate_return_pickup.py로 저장한다.

python
import asyncio
import hashlib
import json
import os

from openai import AsyncOpenAI


client = AsyncOpenAI()
model = os.environ.get("OPENAI_MODEL")
request_key = os.environ.get("REQUEST_KEY")

if not model or not request_key:
    raise RuntimeError("Set OPENAI_MODEL and a stable REQUEST_KEY.")

TOOLS = [
    {
        "type": "function",
        "name": "register_return_pickup",
        "description": "Register one pickup date for an eligible return order.",
        "strict": True,
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "Verified order ID, for example ORD-7391.",
                },
                "pickup_date": {
                    "type": "string",
                    "enum": ["2026-07-30", "2026-07-31"],
                },
            },
            "required": ["order_id", "pickup_date"],
            "additionalProperties": False,
        },
    }
]

ALLOWED_DATES = {"2026-07-30", "2026-07-31"}
PICKUPS = {}
RESULT_LEDGER = {}
LEDGER_LOCK = asyncio.Lock()


def parse_arguments(raw):
    try:
        value = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise ValueError("invalid_json") from exc

    if not isinstance(value, dict):
        raise ValueError("arguments_must_be_object")
    if set(value.keys()) != {"order_id", "pickup_date"}:
        raise ValueError("unexpected_or_missing_field")
    if not isinstance(value["order_id"], str) or not value["order_id"].startswith("ORD-"):
        raise ValueError("invalid_order_id")
    if value["pickup_date"] not in ALLOWED_DATES:
        raise ValueError("pickup_date_not_available")
    return value


def ledger_key(tool_name, arguments):
    normalized = json.dumps(
        {
            "request_key": request_key,
            "tool": tool_name,
            "order_id": arguments["order_id"],
            "pickup_date": arguments["pickup_date"],
        },
        sort_keys=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()


async def register_return_pickup(arguments):
    pickup_id = f"{request_key}:{arguments['order_id']}"
    PICKUPS[pickup_id] = {
        "pickup_id": pickup_id,
        "order_id": arguments["order_id"],
        "pickup_date": arguments["pickup_date"],
        "status": "registered",
    }
    return PICKUPS[pickup_id]


HANDLERS = {"register_return_pickup": register_return_pickup}


def tool_output(call_id, payload):
    return {
        "type": "function_call_output",
        "call_id": call_id,
        "output": json.dumps(payload, ensure_ascii=False),
    }


async def dispatch(call):
    if not call.call_id:
        raise RuntimeError("Protocol error: missing call_id; stop the continuation.")

    handler = HANDLERS.get(call.name)
    if handler is None:
        return tool_output(
            call.call_id,
            {"ok": False, "error": "function_not_allowed"},
        )

    try:
        arguments = parse_arguments(call.arguments)
    except ValueError as exc:
        return tool_output(
            call.call_id,
            {"ok": False, "error": str(exc)},
        )

    key = ledger_key(call.name, arguments)
    try:
        # 데모에서는 side effect와 ledger 기록을 하나의 임계 구역으로 묶는다.
        # 운영에서는 DB unique constraint + transaction/outbox로 교체한다.
        async with LEDGER_LOCK:
            if key not in RESULT_LEDGER:
                result = await handler(arguments)
                RESULT_LEDGER[key] = {"ok": True, "data": result}
            payload = RESULT_LEDGER[key]
    except Exception:
        payload = {"ok": False, "error": "pickup_service_failed"}

    return tool_output(call.call_id, payload)


async def main():
    instructions = (
        "당신은 반품 접수 도우미입니다. "
        "수거일 확정 전 register_return_pickup을 호출하세요. "
        "도구 결과에 없는 상태나 일정을 만들지 마세요. "
        "ok=false이면 재실행하지 말고 사용자가 확인할 항목을 설명하세요."
    )
    input_items = [
        {
            "role": "user",
            "content": "주문 ORD-7391 반품을 2026년 7월 31일에 수거해 주세요.",
        }
    ]

    for _round in range(6):
        response = await client.responses.create(
            model=model,
            instructions=instructions,
            input=input_items,
            tools=TOOLS,
            parallel_tool_calls=True,
            store=False,
        )

        if response.status == "incomplete":
            details = getattr(response, "incomplete_details", None)
            raise RuntimeError(
                f"Trace {response.id}: incomplete response; "
                f"do not accept partial text. Details: {details!r}"
            )

        # text만 복사하지 않고 reasoning을 포함한 모든 output Item을 재전송한다.
        input_items.extend(response.output)
        calls = [
            item
            for item in response.output
            if item.type == "function_call"
        ]

        if not calls:
            if not response.output_text:
                raise RuntimeError(
                    f"Trace {response.id}: final text is missing."
                )
            print(response.output_text)
            return

        # asyncio.gather는 현재 라운드의 모든 결과를 모은 뒤 continuation한다.
        outputs = await asyncio.gather(*(dispatch(call) for call in calls))
        input_items.extend(outputs)

    raise RuntimeError("Tool loop exceeded 6 rounds; inspect the trace before retry.")


if __name__ == "__main__":
    asyncio.run(main())

오류 계약은 코드와 동일하다. function_not_allowed, 깨진 JSON, 누락 필드, 불가능한 날짜, pickup service exception은 원래 call_id를 가진 구조화 function_call_output으로 돌아간다. 모델은 그 실패를 사용자에게 설명할 수 있지만 앱은 자동으로 다른 함수를 찾거나 부작용을 재시도하지 않는다. call_id 자체가 없으면 결과를 안전하게 연결할 수 없으므로 trace를 남기고 continuation을 중단한다.

asyncio.gather는 call 여러 개를 한 배치로 수집한다. 결과 배열 위치가 아니라 각 output의 call_id가 상관관계를 보장한다. 예제의 asyncio.Lock은 한 프로세스 안에서만 check와 기록을 묶는다. 운영에서는 stable REQUEST_KEY에 대한 데이터베이스 unique constraint와 업무 변경을 같은 트랜잭션으로 처리하거나, 외부 택배 시스템이면 transactional outbox와 사후 reconciliation을 둬야 한다. 모델 요청을 새로 시작하면 call_id가 바뀔 수 있으므로 business key가 별도로 필요하다.

Chat 형식을 어디까지 걷어내야 하나

Chat Completions 객체를 Responses 입력에 부분적으로 섞지 않는다. 교체 지점은 다음과 같다.

OpenAI 공식 Responses migration 문서는 output이 message만의 배열이 아니라 typed Items라는 점과 평탄한 function tool 형태를 설명한다. 공식 Function Calling 문서에 따라 custom function의 허가·인자 검증·실행·오류·retry 책임은 모델이 아니라 애플리케이션에 있다.

strict는 생략 시 동작을 추측하지 말고 명시적으로 선택한다. strict: True를 쓸 때는 위처럼 필요한 필드를 required에 넣고 additionalProperties: False로 닫은 뒤, 실제 모델과 SDK 버전에서 schema를 검증한다. 애플리케이션 함수를 실행하지 않고 schema에 맞는 JSON만 필요하다면 Structured Outputs와 function calling 비교에서 계약을 먼저 나눈다.

상태 선택은 코드 편의보다 보관 조건이 먼저다

이 글의 하네스는 manual replay다. store=False를 지정하고, 응답의 일부가 아니라 response.output 전체와 각 tool output을 다음 input에 이어 붙인다. reasoning Item을 텍스트로 바꾸거나 버리면 안 된다.

다른 두 방식은 다음 조건에서만 선택한다.

• previous_response_id: 서버 측 응답 상태가 허용되고 짧은 continuation이 필요할 때 쓴다. 이전 top-level instructions는 자동 승계되지 않으므로 매 요청에서 다시 보낸다. conversation과 동시에 지정하지 않는다.
• Conversations: 세션·기기·작업을 넘는 장기 대화에 쓴다. retention, 접근 권한, 삭제 절차를 먼저 정한다.

OpenAI Conversation state 가이드에 따르면 Responses는 기본적으로 응답을 저장한다. previous_response_id를 사용해도 이전 문맥의 input token 비용이 사라지지 않는다. 규제 또는 stateless 요구가 있다면 조직의 현재 data control 조건도 별도로 확인해야 한다.

스트리밍 장애를 세 시점으로 나눠라

Chat Completions의 문자열 delta parser를 그대로 쓰면 함수 인자가 완성되기 전에 실행하거나 call 자체를 놓칠 수 있다. 현재 Responses streaming 가이드를 기준으로 다음 시점을 분리한다.

• 인자 수집 중: response.function_call_arguments.delta를 Item별 buffer에 추가하고 절대 실행하지 않는다.
• 인자 완료: response.function_call_arguments.done 뒤 전체 JSON을 parse·검증하고 allowlisted handler만 실행한다.
• 응답 종료: response.completed, response.failed, response.incomplete를 서로 다른 상태로 기록한다.

인자 done 전에 연결이 끊기면 미완성 call을 버린다. side effect 이후 최종 답변 stream만 끊겼다면 ledger와 택배 시스템 상태를 조회한 뒤 답변만 복구한다. 수거 등록을 다시 실행해서는 안 된다.

한국어 provider 문서와 OpenAI 계약을 분리하기

본문 코드는 OpenAI direct Python SDK를 기준으로 한다. Azure OpenAI는 base URL, 인증, deployment name과 지원 범위를 별도로 관리하므로 Microsoft 한국어 Responses 마이그레이션 가이드를 따라야 한다. 다른 Responses-compatible provider도 모델명, billing, 저장, streaming events, parallel call 지원이 독립 계약이다. provider별 staging probe 없이 “drop-in”이라고 결론 내리지 않는다.

연구의 Google 페이지는 hl=ko&gl=KR로 확인했지만 실제 물리적 캡처 지역은 알 수 없었다. 따라서 이 글은 한국 현지 수요 강도나 계정 이용 가능성을 확정하지 않는다. 첫 Responses 요청과 공식 지역 조건이 아직 확인되지 않았다면 한국어 OpenAI API 키 설정 가이드부터 진행한다.

배포 문 앞에서 보는 증거표

전환 중 429가 발생하면 quota 부족과 rate limit을 섞지 말고 한국어 OpenAI API quota 오류 가이드에서 billing·사용량·속도 제한을 분리한다.

다음 행동은 Python 하네스를 고정된 REQUEST_KEY로 staging에서 두 번 실행하는 것이다. ledger와 실제 수거 등록이 각각 한 건이고 모든 call_id가 맞으며 최종 문장이 나온 뒤에만 내부 traffic slice를 Responses로 전환한다. Chat Completions는 계속 지원되므로 feature flag와 rollback을 유지한 채 점진적으로 확대할 수 있다.