C.W.K.
Stream
Lesson 06 of 07 · published

SDK 없이 Multi-Turn Tool Loop

~22 min · raw-tool-loop, no-sdk

Level 0Tokenizer
0 XP0/54 lessons0/10 achievements
0/120 XP to next level120 XP to go0% complete

개념적 loop 는 SDK 와 동일 — request build → POST → response parse → if tool_calls: 실행 + append + POST 다시 → else: 텍스트 반환. 차이는 매 step 이 dev 가 instrument 가능하다는 것.

Hook 박을 곳

  • Raw request body log (wire 검증)
  • Raw response body log (debug)
  • Parsed tool_call 결정 log (eval)
  • Handler return 값 log (replay)

SDK 가 hide 하는 거 보임

매 step 의 raw shape — request 의 정확한 JSON, response 의 typed/untyped, tool result 의 wrapping (function_call_output vs tool-role message). 한 번 직접 wire 해 보면 SDK 의 magic 이 사라짐.

Production 가치

Per-iteration breakpoint hook(state) 같은 것 — SDK 로는 불가. Customer bug 를 captured wire trace 에서 reproduce, LLM gateway 와 통합 (gateway 가 'well-behaved HTTP client' 동작 기대), per-call instrumentation 다 raw level 에서.

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

External links

Exercise

트랙 5 의 tool loop 를 raw httpx 로 완전 재구현. 같은 3 fake tool, 같은 iteration cap. SDK 로는 어려운 기능 1 개 추가 — per-iteration breakpoint hook(state) 가 mid-loop state 검사/변형 가능.

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.