C.W.K.
Stream
Lesson 03 of 04 · published

Multi-turn tool loop

~24 min · tools, loop, agent

Level 0Downloader
0 XP0/41 lessons0/11 achievements
0/120 XP to next level120 XP to go0% complete

네 부분 흐름

  1. User turn — 사람이 뭔가 물어봄.
  2. Assistant turn (tool_calls 포함) — 모델이 "이거 답하려면 get_weather(Tokyo) 호출 필요" 라고 함.
  3. Tool turn — 아빠 코드가 함수 실행하고 conversation에 {"role": "tool", "content": "..."} 추가.
  4. Assistant turn (final) — 모델이 tool 결과 가지고 답.

Single round 아니라 loop

모델이 tool 결과 본 후 또 tool 호출할 수 있어. Loop: messages 보내 → 응답에 tool_calls 있으면 실행하고 append → 없으면 답 반환. 고정 max는 없는데, 실전에선 5–10 iteration이면 거의 다 커버; runaway 막으려면 cap 두기.

Replace 아니라 Append

Conversation은 전체 message history. 매 iteration마다 append: assistant tool_calls turn, 그다음 tool 결과 turn. 다음 turn에서 모델이 제대로 reasoning하려면 전체 history 필요. Assistant turn 교체하면 모델 plan 손실.

Code

Production-grade tool loop·python
import httpx, json, glob

OLLAMA = "http://localhost:11434/api/chat"

def get_weather(city: str, unit: str = "celsius") -> str:
    # Stub — 실제 API 호출로 교체
    return json.dumps({"city": city, "temp": 22, "unit": unit, "condition": "sunny"})

def search_files(pattern: str, directory: str = ".") -> str:
    matches = glob.glob(f"{directory}/{pattern}")
    return json.dumps({"files": matches[:10], "count": len(matches)})

REGISTRY = {"get_weather": get_weather, "search_files": search_files}

def chat_with_tools(model: str, user_message: str, tools: list,
                    max_iters: int = 8) -> str:
    messages = [{"role": "user", "content": user_message}]

    for _ in range(max_iters):
        resp = httpx.post(OLLAMA, json={
            "model": model, "messages": messages,
            "tools": tools, "stream": False,
        }, timeout=120.0).json()

        msg = resp["message"]
        messages.append(msg)

        if not msg.get("tool_calls"):
            return msg["content"]

        for call in msg["tool_calls"]:
            name = call["function"]["name"]
            args = call["function"]["arguments"]
            try:
                result = REGISTRY[name](**args) if name in REGISTRY \
                         else json.dumps({"error": f"Unknown tool: {name}"})
            except Exception as e:
                result = json.dumps({"error": str(e)})
            messages.append({"role": "tool", "content": result})

    return f"<<max iterations ({max_iters}) reached>>"

# 사용 — 두 tool 순차 필요
print(chat_with_tools(
    "qwen2.5:7b",
    "Find all .py files under '.' and tell me the weather in Seoul.",
    tools=[
        {"type": "function", "function": {
            "name": "get_weather",
            "description": "Current weather for a city.",
            "parameters": {"type": "object",
                "properties": {"city": {"type": "string"},
                               "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}},
                "required": ["city"]}}},
        {"type": "function", "function": {
            "name": "search_files",
            "description": "Search files by glob pattern under a directory.",
            "parameters": {"type": "object",
                "properties": {"pattern": {"type": "string"},
                               "directory": {"type": "string"}},
                "required": ["pattern"]}}},
    ],
))

External links

Exercise

chat_with_tools(model, user_message, tools, max_iters) 구현. Prompt 셋으로 test: tool 호출 0개 필요한 거, 1개 필요한 거, 순차 2개 필요한 거. 각각의 iteration count 로깅.

Progress

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

댓글 0

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

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