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

Tool Runner: 재사용 가능한 루프 추상

~14 min · tool-runner, abstraction, library

Level 0Observer
0 XP0/64 lessons0/13 achievements
0/150 XP to next level150 XP to go0% complete

대부분 루프가 동일

프로젝트 셋 후엔 루프가 동일하다는 거 알아챌 거 — assistant 콘텐츠 append, stop_reason 분기, tool_use 블록 dispatch(가능한 곳 병렬), tool_results append, 반복. 이걸 Tool Runner로 추출 — (tools, handlers, max_iters) 받아 루프 도는 클래스나 함수.

Tool Runner가 소유하는 것

소유 — tool registry, dispatcher(concurrency 가진), iteration limit, error handling, observability hook. 소유 X — tool 정의나 핸들러(도메인 코드와 살아), 프롬프트(caller 일), 모델 selection(caller 일).

cwkPippa-shape Tool Runner

cwkPippa Claude 베리언트는 Agent SDK의 built-in tool support 사용, but ChatGPT랑 Gemini 베리언트는 Tool Runner 패턴 구현. 같은 registry, 같은 dispatch 모양, just 다른 upstream API. 패턴이 프로바이더 사이 이동에 살아남아.

원칙: 셋째 도구 후 루프 추출. 더 일찍은 premature; 더 늦은 건 같은 코드 다시 쓰는 거.

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

External links

Exercise

프로젝트 tool 루프를 Tool Runner로 추출. Observability hook 셋 추가 — pre-call, post-call, on-error. 기존 로깅에 wire — 새 의존성 X.
Hint
Runner 추출이 어려우면 핸들러가 루프 상태에 의존; 그게 먼저 고칠 버그.

Progress

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

댓글 0

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

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