하네스 워크북 Lab 03 교재 0/11

Lab 03 · 약 25분

에이전트 루프

1 개념

원문: 04-the-agent-loop.md

무엇이 깨지나

실제 작업은 도구를 여러 번, 이전 결과를 보고 이어서 써야 한다.

패치

stop_reason == "tool_use" 인 동안 [호출 → 실행 → 결과 덧붙이기]를 반복. 이게 '에이전트'의 전부다. 병렬 tool_use 결과는 하나의 user 메시지에 모은다. MAX_ROUNDS 로 폭주를 막는다.

비용

라운드마다 배열 전체를 다시 보낸다 → 비용이 라운드 수의 제곱처럼 커진다 (→ Lab 4).

할 일 (exercise.py)

python run_tests.py 3 -v

생각해볼 것

2 따라 해보기

비유 → 짧은 코드 → 실제 출력. 아래 출력은 미리 실행해 둔 결과이고, ▶ 실행을 누르면 브라우저 안 파이썬으로 다시 돌립니다(첫 실행 때 약 10MB 내려받음).

예제 1 생각 → 행동 → 관찰, 라운드마다 찍어 보기

에이전트 강의에서 말하는 Thought → Action → Observation 순환은 코드로는 while stop_reason == "tool_use" 한 줄입니다.

아래 두뇌는 대본(ScriptedBrain)대로 말하지만, 도구는 진짜로 실행됩니다. 목록 보기 → 파일 읽기 → 답하기, 세 라운드를 지켜보세요.

import solution as lab
from common.brain import ScriptedBrain, reply, text, tool_use

script = ScriptedBrain([
    reply(text("어떤 문서가 있는지 먼저 볼게요."),
          tool_use("run_command", {"command": "ls data/procurement/docs"}, id="t1")),
    reply(text("대금지급 규정이 있네요. 읽어 보겠습니다."),
          tool_use("read_file", {"path": "data/procurement/docs/06_대금지급_규정.md"}, id="t2")),
    reply(text("규정상 어음 지급은 하지 않습니다. (06_대금지급_규정.md)")),
])
rounds = 0


def narrated(request):
    global rounds
    rounds += 1
    r = script(request)
    thought = "".join(b["text"] for b in r["content"] if b["type"] == "text")
    print(f"🧠 라운드 {rounds}  생각: {thought}")
    print(f"   stop_reason = {r['stop_reason']}")
    return r


def observed(name, args):
    out = lab.execute_tool(name, args)
    print(f"   🔧 행동: {name}  →  👀 관찰: {out.strip().splitlines()[0]} …")
    return out


messages = [{"role": "user", "content": "어음으로 지급해도 돼?"}]
answer = lab.run_turn(narrated, messages, execute=observed, log=lambda *a: None)
print("\n최종 답:", answer)
print("배열 길이:", len(messages), "(user 1 + 왕복 2번×2 + 마지막 답 1)")
실행 결과
🧠 라운드 1  생각: 어떤 문서가 있는지 먼저 볼게요.
   stop_reason = tool_use
   🔧 행동: run_command  →  👀 관찰: 01_발주_가이드.md …
🧠 라운드 2  생각: 대금지급 규정이 있네요. 읽어 보겠습니다.
   stop_reason = tool_use
   🔧 행동: read_file  →  👀 관찰: # 대금지급 규정 …
🧠 라운드 3  생각: 규정상 어음 지급은 하지 않습니다. (06_대금지급_규정.md)
   stop_reason = end_turn

최종 답: 규정상 어음 지급은 하지 않습니다. (06_대금지급_규정.md)
배열 길이: 6 (user 1 + 왕복 2번×2 + 마지막 답 1)

💡 바꿔 보기: 두 번째 대본의 파일을 05_대사_통제.md 로 바꾸면 관찰 결과가 어떻게 달라지나요?

예제 2 고집 센 모델과 안전장치 MAX_ROUNDS

모델이 끝없이 도구만 부르면? 루프는 영원히 돌고 요금은 계속 나갑니다. 그래서 라운드 상한이 필요합니다. 상한에 걸리면 하네스가 정중하게 턴을 끝냅니다.

import solution as lab
from common.brain import ScriptedBrain, reply, tool_use

stubborn = ScriptedBrain([reply(tool_use("run_command", {"command": "echo 한 번만 더"}, id=f"s{i}"))
                          for i in range(10)])
messages = [{"role": "user", "content": "아무거나 해 봐"}]
print(lab.run_turn(stubborn, messages, max_rounds=3,
                   execute=lambda n, a: "ok", log=print))
print("두뇌 호출 횟수:", len(stubborn.requests))
실행 결과
  [tool] run_command({'command': 'echo 한 번만 더'})
  [tool] run_command({'command': 'echo 한 번만 더'})
  [tool] run_command({'command': 'echo 한 번만 더'})
[harness] MAX_ROUNDS(3) 도달 — 이번 턴을 중단합니다.
두뇌 호출 횟수: 3

💡 바꿔 보기: max_rounds 를 5 로 바꿔 보세요.

예제 3 도구 3개면 코딩 에이전트 (Thorsten Ball 의 'How to Build an Agent' 축소판)

Thorsten Ball 은 "에이전트에는 비밀이 없다 — LLM + 루프 + 도구 몇 개"라고 보여 줬습니다. 파일 쓰기와 읽기만으로도 '만들고 → 확인하는' 작업이 됩니다.

import os
import tempfile
import solution as lab
from common.brain import ScriptedBrain, reply, text, tool_use

os.chdir(tempfile.mkdtemp())          # 빈 작업 폴더에서
brain = ScriptedBrain([
    reply(text("메모 파일을 만들게요."),
          tool_use("write_file", {"path": "todo.md", "content": "- [ ] 발주서 검토\n- [ ] 송장 대사"}, id="w1")),
    reply(text("잘 써졌는지 확인합니다."),
          tool_use("run_command", {"command": "cat todo.md"}, id="c1")),
    reply(text("todo.md 에 할 일 2개를 적었습니다.")),
])
messages = [{"role": "user", "content": "할 일 메모 파일 만들어 줘"}]
print(lab.run_turn(brain, messages))
print("\n실제로 만들어진 파일:\n" + open("todo.md", encoding="utf-8").read())
실행 결과
  [tool] write_file({'path': 'todo.md', 'content': '- [ ] 발주서 검토\n- [ ] 송장 대사'})
  [tool] run_command({'command': 'cat todo.md'})
todo.md 에 할 일 2개를 적었습니다.

실제로 만들어진 파일:
- [ ] 발주서 검토
- [ ] 송장 대사

💡 바꿔 보기: 대본에 run_command 로 ls 하는 라운드를 하나 더 넣어 보세요.

3 확인 퀴즈

Q1 루프를 멈추는 조건은?

Q2 라운드가 늘수록 비용이 빠르게 커지는 이유는?

4 실습: exercise.py 채우기

TODO 표시 부분을 채우고 채점하세요. 입력 내용은 이 기기에 자동 저장됩니다.

5 정답과 테스트

정답 코드 보기 (solution.py)
#!/usr/bin/env python3
"""Lab 3 — 에이전트 루프: 호출 → 실행 → 덧붙이기 → 반복.

Lab 2는 도구를 한 번만 허용했습니다. 실제 작업은 "파일 목록 보기 → 읽기 → 고치기 →
테스트" 처럼 여러 번 필요하죠. 해법은 while 루프 하나입니다.
"에이전트"란 결국 stop_reason 이 tool_use 인 동안 도는 이 루프입니다.

실행:  python labs/03_agent_loop/solution.py
"""
import os
import sys

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
from common.brain import get_brain  # noqa: E402
from common.toolbox import BASE_TOOLS, execute_tool  # noqa: E402

SYSTEM = ("너는 로컬 머신에서 일하는 엔지니어링 에이전트다. "
          "도구로 작업을 끝낸 뒤 한 일을 요약하라.")
MAX_ROUNDS = 50


def run_turn(brain, messages, tools=BASE_TOOLS, system=SYSTEM,
             execute=execute_tool, max_rounds=MAX_ROUNDS, log=print):
    """사용자 한 턴 = 작업이 끝날 때까지 필요한 만큼의 모델/도구 왕복.

    규칙
    1. 매 라운드: 두뇌 호출 → 응답 content 전체를 assistant 메시지로 추가
    2. stop_reason 이 "tool_use" 가 아니면 → 응답 텍스트를 반환하고 끝
    3. 응답 속 '모든' tool_use 를 실행 → tool_result 들을 '하나의' user 메시지로 추가
    4. max_rounds 를 넘으면 멈추고 "[harness]" 로 시작하는 문자열 반환 (폭주 방지)
    """
    # >>> TODO: 위 규칙대로 루프를 구현하세요 (log 로 도구 호출을 출력하면 보기 좋습니다)
    for _ in range(max_rounds):
        response = brain({"system": system, "messages": messages, "tools": tools})
        messages.append({"role": "assistant", "content": response["content"]})
        if response["stop_reason"] != "tool_use":
            return "".join(b["text"] for b in response["content"] if b["type"] == "text")
        results = []
        for b in response["content"]:
            if b["type"] == "tool_use":
                log(f"  [tool] {b['name']}({str(b['input'])[:100]})")
                results.append({"type": "tool_result", "tool_use_id": b["id"],
                                "content": execute(b["name"], b["input"])})
        messages.append({"role": "user", "content": results})
    return f"[harness] MAX_ROUNDS({max_rounds}) 도달 — 이번 턴을 중단합니다."
    # <<< TODO


def main():
    brain, messages = get_brain(), []
    while True:
        try:
            user_text = input("\nyou> ").strip()
        except EOFError:
            break
        if user_text in ("/quit", ""):
            break
        messages.append({"role": "user", "content": user_text})
        print(f"\nassistant> {run_turn(brain, messages)}")


if __name__ == "__main__":
    main()
채점 테스트 보기 (test_lab.py)
import os
import sys
import unittest

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
from common.labkit import load_target  # noqa: E402
from common.brain import ScriptedBrain, reply, text, tool_use  # noqa: E402

lab = load_target(__file__)
quiet = lambda *a, **k: None  # noqa: E731


def fake_exec(name, args):
    return f"ran {name} {args.get('command', '')}"


class Lab3(unittest.TestCase):
    def run_script(self, script, **kw):
        brain = ScriptedBrain(script)
        msgs = [{"role": "user", "content": "작업해줘"}]
        out = lab.run_turn(brain, msgs, execute=fake_exec, log=quiet, **kw)
        return brain, msgs, out

    def test_plain_answer_one_call(self):
        brain, msgs, out = self.run_script([reply(text("그냥 답"))])
        self.assertEqual(out, "그냥 답")
        self.assertEqual(len(brain.requests), 1)
        self.assertEqual(len(msgs), 2)

    def test_multi_round(self):
        brain, msgs, out = self.run_script([
            reply(tool_use("run_command", {"command": "ls"}, id="t1")),
            reply(text("생각 중"), tool_use("run_command", {"command": "cat x"}, id="t2")),
            reply(text("끝!")),
        ])
        self.assertEqual(out, "끝!")
        self.assertEqual(len(brain.requests), 3)
        self.assertEqual([m["role"] for m in msgs],
                         ["user", "assistant", "user", "assistant", "user", "assistant"])
        self.assertEqual(msgs[2]["content"][0]["tool_use_id"], "t1")
        self.assertEqual(msgs[4]["content"][0]["content"], "ran run_command cat x")
        # 세 번째 요청엔 앞의 모든 왕복이 담겨 있어야 한다
        self.assertEqual(len(brain.requests[2]["messages"]), 5)
        self.assertIn("tools", brain.requests[0])

    def test_parallel_tools_one_user_message(self):
        brain, msgs, out = self.run_script([
            reply(tool_use("run_command", {"command": "a"}, id="p1"),
                  tool_use("run_command", {"command": "b"}, id="p2")),
            reply(text("ok")),
        ])
        results = msgs[2]["content"]
        self.assertEqual([r["tool_use_id"] for r in results], ["p1", "p2"],
                         "병렬 tool_use 결과는 '하나의' user 메시지에 모두 담아야 합니다")

    def test_max_rounds(self):
        forever = [reply(tool_use("run_command", {"command": "again"})) for _ in range(5)]
        brain, msgs, out = self.run_script(forever, max_rounds=3)
        self.assertEqual(len(brain.requests), 3)
        self.assertTrue(out.startswith("[harness]"))


if __name__ == "__main__":
    unittest.main()

📚 더 읽을거리

교재 전체 목록 →