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

python3 pickup_loop.py

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())

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)

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")