정답 코드 보기 (solution.py)
#!/usr/bin/env python3
"""Lab 6 — 돌고 있는 루프 조종하기: 리마인더 큐 + 할 일 목록.
긴 작업 중 두뇌는 목표를 잊고 표류합니다. 하네스는 다음 user 쪽 메시지(=tool_result
메시지)에 <system-reminder> 블록을 슬쩍 끼워 넣어 방향을 잡아줍니다.
- 사용자가 실행 중에 던진 말 (steer)
- 하네스가 감지한 사실 (예: "파일이 바깥에서 바뀌었음")
- 매 라운드 다시 보여주는 할 일 목록 (todo)
주의: 끼워 넣는 위치는 '배열 끝'입니다. 시스템 프롬프트를 고치면 캐시가 깨지니까요.
"""
import os
import sys
import threading
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
TODO_TOOL = {
"name": "todo_write",
"description": "할 일 목록 전체를 교체한다. 한 줄에 하나, 끝난 항목은 'x ' 로 시작.",
"input_schema": {"type": "object",
"properties": {"items": {"type": "string"}},
"required": ["items"]},
}
def reminder_block(text):
return {"type": "text", "text": f"<system-reminder>\n{text}\n</system-reminder>"}
class Steering:
def __init__(self):
self._lock = threading.Lock() # 다른 스레드(입력, 백그라운드 작업)도 remind 할 수 있음
self._queue = []
self.todos = [] # [(done: bool, text: str)]
def remind(self, text):
with self._lock:
self._queue.append(text)
def steer(self, user_text):
"""사용자가 루프 실행 중에 끼어든 말."""
self.remind(f"사용자가 작업 도중 이렇게 말했습니다: {user_text}")
def todo_write(self, items):
"""'x 끝난일\\n할일' 형식 문자열을 파싱해 self.todos 를 교체하고 확인 메시지를 반환."""
# >>> TODO: 줄마다 'x ' 접두사로 완료 여부를 판단해 (bool, text) 리스트로 만드세요 (빈 줄 무시)
todos = []
for line in items.splitlines():
line = line.strip()
if not line:
continue
done = line.startswith("x ")
todos.append((done, line[2:].strip() if done else line))
self.todos = todos
return f"todo 갱신: {sum(d for d, _ in todos)}/{len(todos)} 완료"
# <<< TODO
def render_todos(self):
return "\n".join(f"[{'x' if d else ' '}] {t}" for d, t in self.todos)
def inject(self, blocks):
"""run_loop 의 inject 훅. tool_result 블록 리스트 '뒤에' 리마인더를 붙인다.
1) 큐에 쌓인 리마인더들을 한 블록으로 합쳐 붙이고 큐를 비운다
2) 할 일 목록이 있고 미완료 항목이 있으면 "현재 할 일:\\n<render_todos>" 블록도 붙인다
리마인더가 하나도 없으면 아무것도 붙이지 않는다."""
# >>> TODO: 잠금 안에서 큐를 비우고, 위 규칙대로 reminder_block 을 blocks 에 append 하세요
with self._lock:
pending, self._queue = self._queue, []
if pending:
blocks.append(reminder_block("\n".join(pending)))
if any(not d for d, _ in self.todos):
blocks.append(reminder_block("현재 할 일:\n" + self.render_todos()))
# <<< TODO
def executor(self, base_execute):
"""todo_write 는 여기서 처리하고 나머지는 base_execute 로."""
def execute(name, args):
if name == "todo_write":
return self.todo_write(args["items"])
return base_execute(name, args)
return execute
if __name__ == "__main__":
s = Steering()
s.todo_write("x 저장소 둘러보기\n테스트 고치기")
s.steer("README도 같이 고쳐줘")
blocks = [{"type": "tool_result", "tool_use_id": "t1", "content": "ok"}]
s.inject(blocks)
for b in blocks:
print(b.get("text") or b)
채점 테스트 보기 (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
from common.loop import run_loop # noqa: E402
lab = load_target(__file__)
class Lab6(unittest.TestCase):
def test_todo_parse(self):
s = lab.Steering()
msg = s.todo_write("x 첫째\n\n둘째\nx 셋째")
self.assertEqual(s.todos, [(True, "첫째"), (False, "둘째"), (True, "셋째")])
self.assertIsInstance(msg, str)
def test_inject_drains_queue_once(self):
s = lab.Steering()
s.remind("A")
s.steer("B")
blocks = []
s.inject(blocks)
self.assertEqual(len(blocks), 1)
self.assertIn("<system-reminder>", blocks[0]["text"])
self.assertIn("A", blocks[0]["text"])
self.assertIn("B", blocks[0]["text"])
again = []
s.inject(again)
self.assertEqual(again, [], "한 번 전달한 리마인더는 큐에서 사라져야 합니다")
def test_todos_reshown_until_done(self):
s = lab.Steering()
s.todo_write("일1")
b = []
s.inject(b)
self.assertEqual(len(b), 1)
self.assertIn("[ ] 일1", b[0]["text"])
s.todo_write("x 일1")
b = []
s.inject(b)
self.assertEqual(b, [], "모두 끝나면 다시 보여줄 필요 없음")
def test_in_loop(self):
s = lab.Steering()
brain = ScriptedBrain([
reply(tool_use("todo_write", {"items": "조사\n수정"}, id="T")),
reply(tool_use("run_command", {"command": "x"}, id="R")),
reply(text("완료")),
])
msgs = [{"role": "user", "content": "go"}]
def base(name, args):
s.steer("천천히 해") # 도구 실행 중 사용자가 끼어듦
return "ok"
run_loop(brain, msgs, [lab.TODO_TOOL], "sys", execute=s.executor(base), inject=s.inject)
second_user = msgs[4]["content"]
self.assertEqual(second_user[0]["type"], "tool_result",
"tool_result 가 먼저, 리마인더는 그 뒤에 와야 합니다")
joined = "".join(b.get("text", "") for b in second_user)
self.assertIn("천천히 해", joined)
self.assertIn("조사", joined)
if __name__ == "__main__":
unittest.main()