SDK 없이 Multi-Turn Tool Loop

SDK가 없어도 tool loop의 원리는 같아. request를 만들고 POST한 뒤 response를 parsing해. tool_calls가 있으면 handler를 실행해 결과를 붙여 다시 POST하고, 없으면 text를 반환해. raw HTTP에서는 이 모든 단계에 관찰 지점을 넣을 수 있어.

기록할 지점

실제 request body — 요청 구조 검증용
실제 response body — debugging용
parsing한 tool_call 결정 — eval용
handler 반환값 — replay용

SDK가 감추던 구조를 확인해

정확한 request JSON, response event type, function_call_output과 tool role message의 차이가 매 단계에서 보여. 한 번 직접 연결하면 SDK가 대신하던 일을 구체적으로 이해할 수 있어.

운영 환경에서 얻는 제어력

iteration마다 hook(state)을 불러 중간 상태를 검사하거나 변경할 수 있어. 저장한 wire trace로 사용자 장애를 재현하고, 표준 HTTP client를 기대하는 LLM gateway와 연결하며, 호출별 측정값도 남길 수 있어.

Code

Raw tool loop end to end·python

import os, json, httpx, asyncio

TOOLS_MAP = {"get_weather": get_weather}
URL = "https://api.openai.com/v1/responses"
HEADERS = {
    "Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
    "Content-Type": "application/json",
}

async def agent_loop(user_message: str, max_iterations: int = 10):
    input_items = [{"role": "user", "content": user_message}]

    async with httpx.AsyncClient(timeout=None) as client:
        for i in range(max_iterations):
            body = {
                "model": "gpt-4.1",
                "input": input_items,
                "tools": tool_schemas,
                "stream": True,
            }
            text = ""
            tool_calls = []
            event_lines = []

            async with client.stream("POST", URL, headers=HEADERS, json=body) as resp:
                resp.raise_for_status()
                async for line in resp.aiter_lines():
                    if line == "":
                        if not event_lines: continue
                        event_type, data_json = None, None
                        for l in event_lines:
                            if l.startswith("event:"): event_type = l[6:].strip()
                            elif l.startswith("data:"): data_json = json.loads(l[5:].strip())
                        event_lines.clear()
                        if event_type == "response.output_text.delta":
                            text += data_json["delta"]
                            print(data_json["delta"], end="", flush=True)
                        elif event_type == "response.function_call_arguments.done":
                            tool_calls.append(data_json)
                    else:
                        event_lines.append(line)

            if not tool_calls:
                print()
                return text

            # Execute tools and add results
            for tc in tool_calls:
                result = TOOLS_MAP[tc["name"]](**json.loads(tc["arguments"]))
                input_items.append({"type": "function_call_output",
                    "call_id": tc["call_id"], "output": json.dumps(result)})

asyncio.run(agent_loop("What's the weather in Tokyo?"))

기록할 지점

SDK가 감추던 구조를 확인해

운영 환경에서 얻는 제어력

Code

External links

Exercise

Progress

댓글 0