세 번째 도구부터 공통 루프를 Tool Runner로 뽑아

도구가 달라도 반복 구조는 같아

도우미 콘텐츠를 이력에 넣고, stop_reason을 확인하고, tool_use를 가능하면 병렬로 실행하고, tool_result를 덧붙여 다시 부르는 흐름은 계속 반복돼. 도구가 셋쯤 되면 이 흐름을 tools, handlers, max_iters를 받는 Tool Runner로 추출할 가치가 생겨.

공통 실행과 도메인 책임을 섞지 마

Runner는 도구 등록부, 동시 실행 배분, 반복 한도, 오류 처리, 관측 훅을 맡아. 도구 정의와 처리기는 각 도메인 가까이에 남기고, 프롬프트와 모델 선택도 호출자가 결정해. 공통화할 것은 반복 제어이지 모든 정책이 아니야.

제공자가 달라도 패턴은 살아남아

cwkPippa의 Claude 구현은 Agent SDK가 제공하는 도구 지원을 쓰지만, ChatGPT와 Gemini 구현에는 같은 Runner 개념이 필요해. 상류 API 형식은 달라도 등록·실행·결과 회수의 책임은 이어져. 제공자 중립이라는 구호보다 실제로 반복되는 제어 흐름을 추출한 결과야.

원칙: 셋째 도구에서 공통 루프를 뽑아. 그 전에는 성급하고, 그 뒤에는 복사 코드가 자라기 시작해.

Code

Minimal Tool Runner·python

import asyncio, json
from dataclasses import dataclass, field
from typing import Callable, Awaitable

@dataclass
class ToolRunner:
    client: AsyncAnthropic
    model: str
    tools: list[dict]
    handlers: dict[str, Callable[..., Awaitable[dict]]]
    max_iters: int = 10
    on_tool_call: Callable[[str, dict], None] | None = None

    async def _dispatch(self, blocks):
        async def one(b):
            if self.on_tool_call:
                self.on_tool_call(b.name, b.input)
            try:
                out = await self.handlers[b.name](**b.input)
                return {"type": "tool_result", "tool_use_id": b.id, "content": json.dumps(out)}
            except Exception as e:
                return {"type": "tool_result", "tool_use_id": b.id, "is_error": True, "content": str(e)}
        return await asyncio.gather(*(one(b) for b in blocks))

    async def run(self, system: str, user: str) -> str:
        messages = [{"role": "user", "content": user}]
        for _ in range(self.max_iters):
            resp = await self.client.messages.create(
                model=self.model, max_tokens=2048, system=system,
                tools=self.tools, messages=messages,
            )
            messages.append({"role": "assistant", "content": resp.content})
            if resp.stop_reason != "tool_use":
                return next(b.text for b in resp.content if b.type == "text")
            blocks = [b for b in resp.content if b.type == "tool_use"]
            messages.append({"role": "user", "content": await self._dispatch(blocks)})
        raise RuntimeError("max_iters reached")

도구가 달라도 반복 구조는 같아

공통 실행과 도메인 책임을 섞지 마

제공자가 달라도 패턴은 살아남아

Code

External links

Exercise

Progress

댓글 0