Devin.KR

여러 에이전트 나눠 쓰기 - 작성자·검토자·병렬의 비용

개발자KR 조회 0

이 장에서 배우는 것

앞 장에서 반복해서 사용할 절차와 팀 규칙을 파일로 정리했다. 이번에는 그 절차를 여러 에이전트에게 나누어 맡긴다. 작성자가 초안을 만들고, 검토자가 완료 기준에 따라 문제를 찾으며, 조율자가 수정과 확정을 관리하는 작은 하네스를 만든다. 에이전트 수를 늘리는 일보다 각 역할이 어떤 입력을 받고 무엇을 결정할 수 있는지 정하는 일이 먼저다.

여러 에이전트는 같은 질문에 답을 여러 개 받는 장치로도 쓸 수 있지만, 이 장에서는 작업 책임을 나누는 방법에 집중한다. 모델 호출은 규칙 기반 가짜 모델로 대신한다. 결과가 매번 같으므로 역할을 나눈 효과와 호출이 늘어난 이유를 직접 확인할 수 있다.

  • 작성·검토·조율의 책임과 결과 형식을 구분한다.
  • 검토자에게 작성자의 목표와 다른 검사 기준을 준다.
  • 병렬 실행이 시간을 줄이는 조건과 재작업을 늘리는 조건을 판단한다.
  • 공유 파일의 쓰기를 한곳으로 모으고 결과를 합치는 규칙을 정한다.
  • 수정안을 확정하기 전에 테스트와 변경 비교를 수행하고 호출 횟수를 출력한다.

문제 상황

작은 문서 정리 하네스가 안내문 한 편을 만든다고 하자. 안내문에는 제목, 목표, 작업 순서, 검증 방법, 완료 조건이 있어야 한다. 작성 에이전트는 읽기 쉬운 문장을 만드는 데 집중한다. 그 결과 제목과 작업은 잘 적었지만, 완료 조건을 “파일 저장”이라고만 쓸 수 있다. 저장했다는 사실은 문서가 요구사항을 만족한다는 증거가 아니다.

이때 작성자에게 “다시 잘 확인하라”고만 요청하면 처음과 비슷한 답이 돌아오기 쉽다. 검토 역할을 따로 두더라도 같은 요청문과 같은 판단 기준을 주면 차이가 작다. 필요한 것은 다른 이름을 붙인 에이전트가 아니라, 다른 책임과 검사 항목을 가진 에이전트다.

문제는 파일에서도 생긴다. 작성자와 검토자가 모두 같은 안내문을 직접 수정하면, 검토자가 고친 완료 조건이 작성자의 다음 저장에 덮일 수 있다. 두 작업이 각자 성공했다고 보고해도 최종 파일은 이전 상태로 돌아갈 수 있다. 호출을 동시에 실행했다는 사실만으로 협업이 성립하지는 않는다.

이 장의 하네스는 두 에이전트가 파일을 직접 고치지 않게 한다. 작성자는 문서 후보를 반환하고, 검토자는 후보에 대한 판정을 반환한다. 조율자는 그 결과를 받아 수정 요청을 보내고, 통과한 후보만 파일로 저장한다. 파일 소유권과 완료 판정을 한곳에 모으면 흐름을 따라가기가 쉬워진다.

역할을 나누려면 판단 기준부터 나눈다

작성자(writer)는 산출물을 만드는 역할이다. 검토자(reviewer)는 산출물이 기준을 만족하는지 판단하는 역할이다. 조율자(coordinator)는 누구를 언제 호출하고 어떤 결과를 확정할지 결정하는 역할이다. 조율자가 반드시 모델일 필요는 없다. 호출 순서와 확정 규칙이 분명하다면 Python 함수가 맡는 편이 더 단순하다.

세 역할이 반환하는 결과와 책임의 경계
역할주요 입력반환 결과결정할 수 있는 것
작성자명세, 현재 후보, 수정 의견새 문서 후보후보의 문장과 구성
검토자문서 후보, 검사 기준통과 여부, 문제 코드검사 기준별 판정
조율자후보와 검토 결과확정 파일, 실행 요약재호출, 중단, 저장

작성자의 목표는 “필요한 내용을 간결하게 작성한다”다. 검토자의 기준은 “필수 항목이 존재하고, 검증 행이 정해진 검사와 연결되며, 완료 조건이 검증 통과를 요구한다”다. 두 역할은 같은 문서를 보지만, 문서를 보는 이유가 다르다. 검토자가 문장을 더 아름답게 고치는 데 집중하면 필수 항목 누락을 놓칠 수 있다.

역할별 요청문을 자연어로 표현하면 다음과 같다. 이 예제에서는 같은 내용을 Python 딕셔너리와 조건 검사로 구현한다. 외부 모델에 실제로 보내지는 않는다.

작성자에게: 문서 정리 안내문을 작성한다. 제목, 목표, 작업, 검증, 완료 항목을 사용한다. 수정 의견이 있으면 현재 후보에서 해당 항목을 고친 새 후보를 반환한다. 파일 저장 여부는 결정하지 않는다.

검토자에게: 후보를 필수 항목과 완료 기준으로 검사한다. 검증 항목은 제목과 본문의 빈 값 검사를 설명해야 한다. 완료 항목은 검증 통과 후 저장을 요구해야 한다. 실패하면 문제 코드를 반환한다. 문서를 직접 수정하지 않는다.

작성자는 후보를 만들고 검토자는 판정하며 조율자만 확정 파일을 저장한다

검토 결과는 “조금 부족하다”보다 “검증 항목 없음”처럼 수정 가능한 단위여야 한다. 이 장에서는 NO_CHECK와 WEAK_DONE이라는 문제 코드를 쓴다. 작성자는 코드를 보고 어느 행을 바꿀지 결정한다. 사람에게 설명할 문장과 프로그램이 분기할 값을 나누면, 문장 표현이 바뀌어도 흐름을 유지하기 쉽다.

역할 분리는 검토의 독립성을 보장하지는 않는다. 같은 모델을 같은 맥락으로 두 번 호출하면 비슷한 가정을 반복할 수 있다. 검토자에게 작성자의 자기평가를 먼저 보여 주면 그 평가에 끌릴 수도 있다. 따라서 검토 입력에는 후보와 검사 기준을 중심으로 넣는다. “작성자가 자신 있게 완료했다고 말했다”는 설명은 검사 근거가 아니다.

또한 검토자의 통과 판정도 최종 증거는 아니다. 이 예제의 검토자는 정해진 문서 형식을 검사한다. 문장의 의미 전체나 실제 업무의 성공 여부까지 확인하지는 않는다. 조율자는 통과한 결과에 별도의 테스트를 적용하고, 저장한 내용을 다시 읽어 확인한다.

병렬은 독립된 작업에서 시간을 줄인다

병렬 실행(parallel execution)은 여러 작업을 겹쳐 실행하는 방식이다. 작업 사이에 의존 관계가 없다면 기다리는 시간을 줄일 수 있다. 예를 들어 문서 세 편을 서로 다른 입력에서 작성하고 마지막에 목록으로 묶는 작업은 나누기 쉽다. 각 작성자가 맡은 문서와 결과 이름을 미리 정하면 서로의 진행을 기다릴 필요가 적다.

반면 작성과 검토는 보통 순서가 필요하다. 검토자는 작성 결과를 받아야 한다. 수정 작성자는 검토 의견을 받아야 한다. 아직 없는 후보를 검토할 수 없으므로 이 흐름을 동시에 시작한다고 핵심 작업이 빨라지지는 않는다. 이번 완성 코드가 순차 실행인 이유도 이 의존 관계 때문이다.

동시에 실행하기 전에 확인할 작업의 성질
작업 관계예기대 효과필요한 합치기 규칙
입력과 출력이 독립됨서로 다른 문서의 초안 작성대기 시간을 겹칠 수 있음문서 이름순으로 모으기
같은 후보를 읽기만 함형식 검사와 내용 검사검토 시간을 겹칠 수 있음모든 필수 검사 통과
앞 결과가 다음 입력임작성 후 검토해당 연결은 순차로 남음후보와 판정을 연결하기
같은 파일을 수정함같은 문단을 두 역할이 수정충돌과 재검토가 늘 수 있음수정 소유자 하나 정하기

대략적인 실행 시간을 생각해 보자. 독립된 두 작업을 순차로 실행하면 두 작업의 시간이 더해진다. 동시에 실행하면 더 오래 걸린 작업의 시간에 결과를 합치는 시간이 붙는다. 이는 지연과 자원 경쟁이 작다고 가정한 설명이다. 실제 환경에서는 실행 준비, 입력 전달, 실패 처리, 결과 정렬에도 시간이 든다.

총작업량도 따로 봐야 한다. 동시에 호출했다고 호출 횟수가 줄어들지는 않는다. 검토자를 하나 추가하면 검토 호출이 늘고, 의견이 충돌하면 작성자를 다시 호출할 수 있다. 시간을 줄이는 선택이 비용을 줄이는 선택과 항상 같지는 않다. 이 장에서는 외부 서비스 요금을 다루지 않고, 역할별 호출 횟수를 비용을 살펴보는 기초 기록으로 사용한다.

호출 횟수 하나만으로 모든 비용을 설명할 수도 없다. 짧은 후보를 읽는 호출과 긴 자료를 읽는 호출은 처리량이 다르다. 그래도 호출 횟수는 “검토자를 붙인 뒤 왜 작성 호출까지 늘었는가”를 찾는 출발점이 된다. 이번 예제는 첫 검토에서 두 문제를 발견하므로 작성 두 번, 검토 두 번을 수행한다.

파일 소유권과 결과 합치기를 정한다

같은 파일을 동시에 고치지 않게 하려면 파일별 소유자를 정하는 방법이 가장 이해하기 쉽다. 문서가 여러 개라면 각 작성자에게 서로 다른 문서를 맡긴다. 한 문서를 여러 관점으로 검토한다면 검토자는 수정 파일 대신 의견 목록을 반환한다. 실제 수정은 해당 문서의 작성자나 조율자가 적용한다.

작업을 문단별로 나눌 수도 있다. 다만 문단 사이에 공통 용어, 번호, 참조가 있다면 각 문단이 독립적이라고 보기 어렵다. 별도 후보로 작성한 뒤 한 역할이 전체 문서의 연결을 확인해야 한다. 나누는 단위가 작아질수록 합치는 책임이 사라지는 것이 아니라 더 중요해질 수 있다.

독립된 후보를 따로 받고 합치기 담당자 한 명이 공통 결과를 만든다

합치기에는 순서와 판정 규칙이 필요하다. 결과가 도착한 순서로 문서를 붙이면 실행 속도에 따라 결과가 달라질 수 있다. 작업 이름이나 명세에 정한 순서로 정렬해야 한다. 검토 결과도 먼저 통과한 검토자의 판정만 쓰지 않는다. 필수 검토가 두 개라면 둘 다 통과해야 확정한다.

서로 모순되는 의견은 조율 규칙으로 해결한다. 한 검토자는 예시를 늘리라고 하고 다른 검토자는 분량을 줄이라고 할 수 있다. 이때 수정안을 무조건 합치기보다 명세의 우선순위를 적용한다. 우선순위가 없으면 임의로 확정하지 않고 미해결 상태로 남기는 편이 판단 근거를 보존한다.

후보와 검토 결과를 연결하는 일도 필요하다. 나중 후보에 이전 후보의 통과 판정을 붙이면 검토하지 않은 문서가 확정된다. 이번 프로그램은 후보를 만든 직후 검토하고, 통과한 그 문자열을 저장하므로 연결이 단순하다. 비동기 실행으로 확장한다면 후보 번호를 함께 전달하고 판정의 후보 번호가 일치하는지 확인해야 한다.

수정 반복에는 상한을 둔다. 검토자가 계속 같은 문제를 반환하거나 작성자가 의견을 반영하지 못할 수 있다. 상한에 도달하면 확정 파일을 만들지 않고 실패를 알린다. 반복을 멈추는 일과 산출물을 완료로 처리하는 일은 서로 다른 결정이다.

완성 코드

다음 프로그램을 main.py로 저장한다. Python 3.12 이상에서 표준 라이브러리만 사용한다. 가짜 모델은 역할과 입력에 따라 정해진 후보 또는 판정을 반환한다. 모든 파일 읽기와 쓰기는 프로그램이 만든 임시 폴더 안에서 수행하며, 종료하면 임시 폴더가 정리된다.

첫 후보는 검증 항목이 없고 완료 조건이 약하다. 검토자는 두 문제를 반환한다. 작성자는 현재 후보에 두 수정 사항을 적용한다. 조율자는 재검토가 통과하면 테스트를 실행하고 최종 파일을 저장한다. 실제 병렬 실행은 넣지 않는다. 작성과 검토 사이의 의존 관계, 결과 확정, 호출 집계를 먼저 확인하는 예제다.

from collections import Counter
from difflib import unified_diff
from pathlib import Path
from tempfile import TemporaryDirectory


SPEC = {
    "title": "문서 정리 안내",
    "goal": "입력 문서를 정리한다.",
    "steps": "제목 정리, 본문 정리",
}
CHECK_LINE = "검증: 제목과 본문이 비어 있지 않은지 검사"
DONE_LINE = "완료: 검증 통과 후 파일 저장"
MAX_ROUNDS = 3


def document_errors(candidate):
    lines = candidate.splitlines()
    errors = []
    expected = {
        "제목:": f"제목: {SPEC['title']}",
        "목표:": f"목표: {SPEC['goal']}",
        "작업:": f"작업: {SPEC['steps']}",
    }
    for prefix, required in expected.items():
        matches = [line for line in lines if line.startswith(prefix)]
        if matches != [required]:
            errors.append("BAD_" + prefix.removesuffix(":"))
    checks = [line for line in lines if line.startswith("검증:")]
    if checks != [CHECK_LINE]:
        errors.append("NO_CHECK")
    done = [line for line in lines if line.startswith("완료:")]
    if done != [DONE_LINE]:
        errors.append("WEAK_DONE")
    return errors


def fake_model(role, request, calls):
    calls[role] += 1
    if role == "writer":
        if not request["feedback"]:
            spec = request["spec"]
            candidate = "\n".join([
                f"제목: {spec['title']}",
                f"목표: {spec['goal']}",
                f"작업: {spec['steps']}",
                "완료: 파일 저장",
            ]) + "\n"
        else:
            lines = request["candidate"].splitlines()
            feedback = request["feedback"]
            if "NO_CHECK" in feedback:
                lines = [
                    line for line in lines
                    if not line.startswith("검증:")
                ]
                position = next(
                    index for index, line in enumerate(lines)
                    if line.startswith("완료:")
                )
                lines.insert(position, CHECK_LINE)
            if "WEAK_DONE" in feedback:
                lines = [
                    DONE_LINE if line.startswith("완료:") else line
                    for line in lines
                ]
            candidate = "\n".join(lines) + "\n"
        return {"candidate": candidate}
    if role == "reviewer":
        issues = document_errors(request["candidate"])
        return {"approved": not issues, "issues": issues}
    raise ValueError("未知の役割".replace("未知の役割", "알 수 없는 역할"))


def require(condition, message):
    if not condition:
        raise AssertionError(message)


def run_tests(initial, final):
    require("NO_CHECK" in document_errors(initial),
            "첫 후보의 검증 누락을 찾아야 한다.")
    require("WEAK_DONE" in document_errors(initial),
            "첫 후보의 약한 완료 조건을 찾아야 한다.")
    require(document_errors(final) == [],
            "최종 후보는 모든 문서 기준을 만족해야 한다.")
    without_check = final.replace(CHECK_LINE + "\n", "")
    require("NO_CHECK" in document_errors(without_check),
            "검증 행을 제거하면 검토가 실패해야 한다.")


def coordinate(calls):
    candidate = ""
    initial = ""
    feedback = []
    history = []
    for round_number in range(1, MAX_ROUNDS + 1):
        result = fake_model("writer", {
            "spec": SPEC,
            "candidate": candidate,
            "feedback": feedback,
        }, calls)
        candidate = result["candidate"]
        if round_number == 1:
            initial = candidate
        review = fake_model("reviewer", {
            "candidate": candidate,
        }, calls)
        if review["approved"]:
            require(review["issues"] == [],
                    "통과 판정에 문제가 남아 있다.")
            history.append(f"회차 {round_number}: 검토 통과")
            return initial, candidate, history
        require(bool(review["issues"]),
                "실패 판정에는 수정 의견이 필요하다.")
        feedback = review["issues"]
        history.append(
            f"회차 {round_number}: 수정 요청 "
            + ", ".join(feedback)
        )
    raise RuntimeError("수정 상한에 도달해 확정하지 않았다.")


def main():
    calls = Counter()
    with TemporaryDirectory() as directory:
        root = Path(directory)
        initial, final, history = coordinate(calls)
        run_tests(initial, final)
        require(calls["writer"] == 2 and calls["reviewer"] == 2,
                "예상한 수정 흐름과 호출 횟수가 다르다.")
        final_path = root / "final.txt"
        final_path.write_text(final, encoding="utf-8")
        saved = final_path.read_text(encoding="utf-8")
        require(saved == final, "저장한 내용이 최종 후보와 다르다.")
        changes = list(unified_diff(
            initial.splitlines(),
            saved.splitlines(),
            fromfile="초안",
            tofile="확정안",
            lineterm="",
        ))
        for entry in history:
            print(entry)
        print("테스트: 통과")
        print("변경 비교:")
        for line in changes:
            print(line)
        print("확정 파일: final.txt")
        print("확정 내용:")
        print(saved, end="")
        print(
            f"호출 횟수: 작성자={calls['writer']}, "
            f"검토자={calls['reviewer']}, "
            f"합계={sum(calls.values())}"
        )


if __name__ == "__main__":
    main()

줄별 해설

가져오기와 상수. Counter는 역할별 호출 횟수를 센다. unified_diff는 첫 후보와 저장한 확정안의 변경을 보여 준다. Path와 TemporaryDirectory는 임시 폴더 안에서 파일을 다룬다. SPEC은 작성자가 유지해야 할 내용이며, CHECK_LINE과 DONE_LINE은 이번 예제의 고정된 검사 기준이다. MAX_ROUNDS는 작성과 검토를 묶은 회차의 상한이다.

document_errors 함수. 후보를 줄로 나누고 항목별 오류를 수집한다. 제목·목표·작업은 해당 접두어로 시작하는 행을 모아 요구한 행 하나와 비교한다. 항목이 없거나 중복되어도 실패한다. 검증과 완료도 같은 방식으로 확인한다. 단순히 문서 어딘가에 “검증”이라는 단어가 있다는 이유로 통과시키지 않는다.

검사 범위. 이 함수는 작은 고정 형식 문서의 계약을 검사한다. 일반적인 한국어 안내문의 의미를 이해하는 함수는 아니다. 표현을 자유롭게 허용하려면 검사 방법도 달라져야 한다. 여기서는 출력 형식을 좁혀서 역할 사이의 전달과 수정 흐름을 결정적으로 확인한다.

fake_model 함수의 첫 줄. 호출을 받으면 역할의 횟수를 먼저 증가시킨다. 따라서 이 값은 성공한 호출만이 아니라 호출 시도 횟수를 뜻한다. 현재 실행에서는 모든 호출이 정상적으로 결과를 반환한다. 조율자의 일반 함수 실행과 파일 저장은 모델 호출 집계에 포함하지 않는다.

작성자의 첫 응답. 수정 의견이 비어 있으면 명세를 이용해 첫 후보를 만든다. 검증을 빠뜨리고 완료 조건을 약하게 쓰는 것은 실험을 위한 고정 동작이다. 모든 작성자가 늘 이런 실수를 한다는 뜻은 아니다. 재현 가능한 실패가 있어야 검토자가 실제로 어떤 수정을 유도했는지 확인할 수 있다.

작성자의 수정 응답. 수정 의견이 있으면 새 문서를 처음부터 임의로 만들지 않고 현재 후보의 행을 바꾼다. NO_CHECK를 받으면 기존 검증 행을 제거한 뒤 완료 행 앞에 올바른 검증 행을 넣는다. WEAK_DONE을 받으면 완료 행을 바꾼다. 이 작성자는 두 문제 코드만 수정할 수 있다. 제목 등의 다른 오류는 상한까지 남을 수 있으며, 그 경우 조율자가 확정을 중단한다.

검토자의 응답. 검토자는 후보를 document_errors에 전달한다. 오류 목록이 비어 있으면 approved가 참이다. 문서를 직접 고치지 않고 판정과 문제 목록만 반환한다. 마지막 예외는 정해지지 않은 역할을 호출했을 때 실행을 중단한다.

require와 run_tests. require는 조건이 거짓이면 검사 실패를 알린다. run_tests는 첫 후보에서 의도한 두 오류를 발견하는지, 최종 후보가 통과하는지 확인한다. 이어서 통과한 후보에서 검증 행을 지우고 다시 실패하는지 검사한다. 정상 결과만 검사하면 검토기가 모든 후보를 통과시키는 잘못을 놓칠 수 있으므로 실패 사례도 만든다.

coordinate의 상태. candidate는 현재 후보, initial은 변경 비교에 사용할 첫 후보, feedback은 다음 작성 호출에 전달할 의견이다. history는 회차별 판정을 기록한다. 작성자는 현재 후보와 의견을 함께 받으므로 무엇을 수정하는지 알 수 있다. 검토자는 작성자의 자기평가 없이 후보를 받는다.

회차 안의 연결. 작성 결과를 candidate에 넣은 직후 그 후보를 검토한다. 통과하면 문제 목록도 비어 있는지 확인하고 해당 후보를 반환한다. 실패하면 문제 목록이 있는지 확인해 다음 회차로 넘긴다. 판정과 문제 목록이 서로 모순되면 그대로 진행하지 않는다. 상한까지 통과하지 못하면 RuntimeError가 발생하며 main은 파일 저장에 도달하지 않는다.

main의 저장 순서. 임시 폴더를 만든 뒤 조율과 테스트를 수행한다. 예상한 두 번의 작성과 두 번의 검토도 확인한다. 이 호출 횟수 검사는 가짜 모델의 고정된 동작을 확인하는 예제용 검사다. 실제 하네스에서는 모든 작업이 반드시 두 번씩 호출되어야 한다고 정하지 않는다.

다시 읽기와 변경 비교. 테스트를 통과한 문자열만 final.txt에 저장한다. 저장한 내용을 다시 읽고 메모리의 최종 후보와 같은지 확인한다. 이어서 첫 후보와 실제 저장 내용을 비교한다. 출력에서 더하기 기호는 추가된 행, 빼기 기호는 삭제된 행이다. 수정 의도와 실제 변화가 맞는지 사람이 살펴볼 수 있다.

확정의 의미. 이 프로그램은 내부 판정과 테스트가 통과한 후보를 확정한다. 실행 결과를 읽는 사람은 변경 비교와 최종 내용을 검토해야 한다. 자동 통과가 사람의 확인을 대신하는 모든 근거가 되지는 않는다. 또한 임시 폴더가 종료 시 삭제되므로 이 예제의 확정 파일은 실행 중에만 존재한다.

실행 결과

먼저 경고를 오류로 취급해 main.py 전체의 구문을 컴파일한 뒤 실행한다. 첫 명령은 성공하면 출력이 없다. 컴파일 명령은 소스 파일을 읽어 메모리에서 검사하며 바이트코드 파일을 만들지 않는다.

python3 -W error -c 'from pathlib import Path; compile(Path("main.py").read_text(encoding="utf-8"), "main.py", "exec")'
python3 main.py

실행 출력은 다음과 같다. 임시 폴더 경로와 현재 시각을 출력하지 않으므로 실행할 때마다 같은 결과가 나온다. 변경 비교에서 삭제와 추가 표시가 없는 행은 두 후보에 공통으로 남아 있는 행이다.

회차 1: 수정 요청 NO_CHECK, WEAK_DONE
회차 2: 검토 통과
테스트: 통과
변경 비교:
--- 초안
+++ 확정안
@@ -1,4 +1,5 @@
 제목: 문서 정리 안내
 목표: 입력 문서를 정리한다.
 작업: 제목 정리, 본문 정리
-완료: 파일 저장
+검증: 제목과 본문이 비어 있지 않은지 검사
+완료: 검증 통과 후 파일 저장
확정 파일: final.txt
확정 내용:
제목: 문서 정리 안내
목표: 입력 문서를 정리한다.
작업: 제목 정리, 본문 정리
검증: 제목과 본문이 비어 있지 않은지 검사
완료: 검증 통과 후 파일 저장
호출 횟수: 작성자=2, 검토자=2, 합계=4

검토자를 추가한 결과 첫 후보의 두 문제가 수정되었고, 총 네 번의 가짜 모델 호출이 발생했다. 이것은 검토가 언제나 두 번이면 충분하다는 뜻이 아니다. 이 입력과 수정 규칙에서 발생한 결과다. 제목·목표·작업 행은 그대로 유지되었고 검증과 완료 행만 바뀌었는지 변경 비교에서 확인한다.

실무에서 자주 틀리는 것

작성자의 성공 보고를 검토 결과로 쓴다

작성자의 자신감이나 완료 보고를 검토 판정으로 바꾸면 별도의 검토 역할을 둔 이유가 약해진다. 다음 잘못된 코드는 필수 항목이 없어도 작성자가 성공이라고 말하면 통과한다. 두 예제는 각각 독립적으로 실행할 수 있는 작은 코드다.

candidate = "제목: 안내\n완료: 파일 저장\n"
writer_report = {"success": True}
approved = writer_report["success"]
print(approved)

고친 코드는 후보에서 요구한 검증 행과 완료 행을 직접 확인한다. 실무에서는 항목별 검사 결과도 함께 남겨 어떤 근거로 판정했는지 살펴볼 수 있게 한다.

candidate = "제목: 안내\n완료: 파일 저장\n"
lines = candidate.splitlines()
approved = (
    lines.count("검증: 제목과 본문이 비어 있지 않은지 검사") == 1
    and lines.count("완료: 검증 통과 후 파일 저장") == 1
)
print(approved)

여러 역할이 같은 파일을 덮어쓴다

다음 코드는 두 역할이 같은 파일을 저장하는 문제를 순차 실행으로 재현한다. 검토자가 반영한 검증 행이 뒤늦게 저장된 작성자의 이전 후보에 덮인다. 실제 동시 실행에서도 이런 저장 순서가 생길 수 있다.

from pathlib import Path
from tempfile import TemporaryDirectory

with TemporaryDirectory() as directory:
    path = Path(directory) / "document.txt"
    writer_old = "작업: 문서 정리\n"
    reviewer_new = writer_old + "검증: 빈 항목 검사\n"
    path.write_text(reviewer_new, encoding="utf-8")
    path.write_text(writer_old, encoding="utf-8")
    print(path.read_text(encoding="utf-8"), end="")

고친 코드에서는 검토자가 의견을 반환하고 조율자가 한 번 저장한다. 별도 후보 파일이 필요하면 역할별로 경로를 나누고, 최종 파일의 쓰기 책임은 하나로 유지한다.

from pathlib import Path
from tempfile import TemporaryDirectory

with TemporaryDirectory() as directory:
    writer_candidate = "작업: 문서 정리\n"
    reviewer_issues = ["검증 행 추가"]
    final = writer_candidate
    if "검증 행 추가" in reviewer_issues:
        final += "검증: 빈 항목 검사\n"
    path = Path(directory) / "document.txt"
    path.write_text(final, encoding="utf-8")
    print(path.read_text(encoding="utf-8"), end="")

먼저 도착한 통과 판정만 채택한다

필수 검토가 여러 개라면 하나의 통과만으로 확정하면 안 된다. 다음 코드는 한 검토가 실패했는데도 다른 검토가 통과했으므로 전체를 통과시킨다. 병렬로 받은 결과를 합칠 때 이런 규칙을 쓰면 실행 순서와 관계없이 필요한 검사를 빠뜨린다.

reviews = {
    "형식": {"approved": True},
    "내용": {"approved": False},
}
approved = any(result["approved"] for result in reviews.values())
print(approved)

고친 코드에서는 필수 검토가 모두 도착했고 모두 통과했는지 확인한다. 빈 결과에 all을 적용하면 참이 나오므로, 결과가 갖춰졌는지 먼저 확인하는 조건이 필요하다.

required = {"형식", "내용"}
reviews = {
    "형식": {"approved": True},
    "내용": {"approved": False},
}
approved = (
    set(reviews) == required
    and all(reviews[name]["approved"] for name in sorted(required))
)
print(approved)

상한에 도달한 후보를 완료로 반환한다

반복 횟수를 제한한 뒤 마지막 후보를 무조건 반환하면 검토 실패와 완료가 섞인다. 다음 코드는 모든 회차에서 검토가 실패했는데도 결과를 확정한 것처럼 출력한다.

candidate = "완료: 파일 저장"
for round_number in range(3):
    approved = False
    if approved:
        break
print("확정:", candidate)

고친 코드에서는 확정 상태를 별도로 두고 통과한 경우에만 설정한다. 상한에 도달했다는 사실은 더 진행하지 않는 이유이며, 결과를 승인할 이유는 아니다. 아래 코드는 실패 상태를 출력하므로 검사 목적에 맞게 정상 종료한다.

candidate = "완료: 파일 저장"
confirmed = None
for round_number in range(3):
    approved = False
    if approved:
        confirmed = candidate
        break
if confirmed is None:
    print("미확정: 수정 상한 도달")
else:
    print("확정:", confirmed)

한눈에 보기

여러 에이전트를 나눠 쓸 때 조율자가 지킬 규칙
주제규칙이 예제의 구현
역할작성과 판정의 책임을 나눈다후보 반환과 검토 결과 반환
검토 기준자기평가보다 완료 조건을 검사한다필수 행과 중복 여부 검사
작업 순서입력 의존 관계가 있으면 순차로 연결한다작성 후 검토, 실패 후 수정
파일 소유권최종 파일의 쓰기 담당자를 하나로 둔다main에서만 저장
확정통과한 후보와 저장한 내용을 연결한다반환 후보 저장 후 다시 읽기
확인실패 사례와 실제 변경을 확인한다검증 행 제거 테스트와 변경 비교
비용호출 횟수와 재작업을 함께 본다작성 2회, 검토 2회 출력
중단상한 도달을 완료로 처리하지 않는다예외 발생 시 저장하지 않음

역할을 나누면 어느 판단이 어디에서 나왔는지 보기 쉬워진다. 그 대신 입력 전달과 결과 합치기, 추가 호출을 관리해야 한다. 작업이 작고 검사 기준이 코드로 충분히 표현된다면 검토 역할을 일반 함수로 두는 선택도 가능하다. 에이전트 수는 작업 구조에 맞춰 정한다.

다음 장에서는 여러 역할이 읽는 입력과 도구 결과에 다른 지시가 섞였을 때 생기는 보안 문제를 다룬다. 이번에 정한 역할의 책임과 파일 소유권은 그 경계를 설명하는 출발점이 된다.

연습 문제

  1. MAX_ROUNDS를 1로 바꾸면 어떤 일이 일어나는가. 작성자와 검토자의 호출 횟수, 최종 파일 저장 여부, 출력 흐름을 설명한다.
  2. 첫 후보와 최종 후보가 “검증”이라는 단어를 포함하는지만 확인하면 왜 부족한가. 오판을 일으키는 후보를 하나 만들고 현재 검토기가 어떻게 처리하는지 설명한다.
  3. 서로 다른 문서 두 편을 병렬로 작성하도록 확장한다고 가정한다. 입력, 후보 이름, 최종 파일 소유자, 결과를 모으는 순서를 정한다. 작성과 검토 중 어느 연결을 순차로 유지해야 하는지도 설명한다.
  4. 형식 검토자와 내용 검토자가 각각 같은 후보를 읽도록 확장한다. 첫 후보는 형식 검토만 통과하고, 수정 후보는 두 검토를 모두 통과한다고 가정한다. 확정 규칙과 역할별 호출 횟수를 구한다.

정답과 해설

  1. 첫 후보를 작성하고 검토하면 두 문제가 반환된다. 다음 작성 회차가 허용되지 않으므로 coordinate가 RuntimeError를 발생시킨다. 작성자와 검토자는 각각 한 번 호출되지만, 현재 main은 coordinate가 반환한 뒤에만 출력하므로 호출 요약을 출력하지 못한다. final.txt는 생성되지 않는다. 오류 추적의 파일 경로나 줄 번호는 환경에 따라 다를 수 있다. 실패 때도 호출 요약이 필요하다면 예외를 받은 조율 계층에서 집계를 출력하도록 확장할 수 있다.

  2. “목표: 검증이라는 단어를 포함한 문서 작성”이 있지만 검증 행이 없는 후보는 단어 포함 검사에 통과할 수 있다. 현재 검토기는 검증 접두어로 시작하는 행을 따로 모으므로 NO_CHECK를 반환한다. 올바른 검증 행이 두 번 들어 있어도 목록이 요구한 행 하나와 같지 않아 실패한다. 단어 존재, 항목 존재, 항목의 값, 항목의 개수는 서로 다른 검사다.

  3. 문서 이름과 입력 자료를 작업별로 고정하고 후보를 이름별 딕셔너리에 모은다. 작성자들은 문자열 후보만 반환하며 최종 파일은 조율자가 저장한다. 결과는 완료 순서가 아니라 문서 이름순이나 명세에 적은 순서로 모은다. 문서별 작성은 겹칠 수 있지만 각 문서의 검토는 해당 후보가 나온 뒤 시작한다. 검토 실패에 따른 수정 작성도 그 검토 결과를 받은 뒤 수행한다. 전체 확정은 필요한 문서가 모두 통과한 뒤 진행한다.

  4. 두 검토가 같은 후보에 대한 결과를 반환했는지 확인하고, 필수 검토 둘 다 통과한 경우에만 확정한다. 실패한 검토의 문제를 모아 작성자에게 전달한다. 첫 후보와 수정 후보를 각각 두 검토자가 검사하므로 작성자는 2회, 형식 검토자는 2회, 내용 검토자는 2회 호출된다. 합계는 6회다. 두 검토를 동시에 실행해도 이 호출 수는 그대로다. 수정 뒤에는 후보가 바뀌었으므로 첫 후보의 형식 통과를 수정 후보에 자동으로 재사용하지 않는다.

댓글 0

아직 댓글이 없습니다. 첫 댓글을 남겨 보세요.

댓글을 남기려면 로그인이 필요합니다.