하네스·루프·그래프 엔지니어링 차이: AI 에이전트 설계 실전 가이드

하네스 = 어디서 무엇을 가지고 일할 것인가
루프   = 언제 다시 시도하고 언제 멈출 것인가
그래프 = 어떤 조건에서 무엇을 다음에 실행할 것인가

project_state.md
├── 최종 목표
├── 완료된 작업
├── 아직 남은 작업
├── 확인된 사실과 출처
├── 현재 막힌 문제
└── 다음 실행에서 시작할 위치

이 프로젝트에서 AI 에이전트가 안전하게 장기 작업을 수행할 수 있는
최소 하네스를 설계해줘.

반드시 포함할 것:
1. 사용할 수 있는 도구와 각 도구의 목적
2. 읽기·쓰기·삭제·외부 발송 권한의 구분
3. 작업 상태를 저장할 파일과 데이터 구조
4. 중단 후 재개할 때 읽어야 할 정보
5. 최대 실행 시간과 비용 제한
6. 사람이 승인해야 하는 행동
7. 실행 로그에 남길 항목
8. 실패했을 때 복구하거나 사람에게 넘기는 조건

도구를 많이 추가하는 방향보다
이 업무에 꼭 필요한 최소 구성으로 설계해줘.

사람이 요청한다
→ AI가 답한다
→ 사람이 문제를 발견한다
→ 사람이 다시 요청한다
→ AI가 수정한다

목표 설정
→ 작업 수행
→ 결과 검사
→ 기준 미달이면 수정
→ 다시 검사
→ 통과하거나 한도에 도달하면 종료

초안 작성
→ 사실·논리·누락 검사
→ 실패 항목 수정
→ 재검사
→ 통과 또는 최대 2회 후 사람에게 전달

매일 오전 9시
→ 새로운 AI 뉴스 수집
→ 중복 제거
→ 중요도 평가
→ 요약 작성
→ 결과 발송

실패 기록 수집
→ 공통 원인 분류
→ 지침 또는 도구 수정
→ 이전 실패 사례로 재평가
→ 개선됐을 때 새 버전 적용

목표:
[만들어야 할 최종 결과물을 적습니다.]

성공 기준:
1. [반드시 포함할 항목]
2. [숫자·출처·형식의 검사 기준]
3. [통과해야 하는 테스트 또는 검토 기준]

작업 방식:
1. 먼저 결과물을 만든다.
2. 결과물을 성공 기준과 하나씩 대조한다.
3. 통과하지 못한 항목만 수정한다.
4. 수정 후 같은 기준으로 다시 검사한다.
5. 모든 기준을 통과하면 종료한다.

종료 규칙:
- 최대 재시도는 2회다.
- 같은 오류가 반복되면 더 이상 추측하지 않는다.
- 확인할 증거가 없으면 해당 내용을 단정하지 않는다.
- 2회 후에도 실패하면 남은 문제와 필요한 사람의 판단을 보고하고 멈춘다.

마지막에는 아래 형식으로 보고한다.
- 최종 결과물
- 통과한 기준
- 남은 문제
- 재시도 횟수

이 글을 처음 보는 검사자라고 생각하세요.

잘한 점은 쓰지 마세요.
“전반적으로 좋습니다” 같은 평가는 금지합니다.
글을 대신 고쳐 쓰지 말고 틀린 곳만 찾으세요.

다음 항목만 검사하세요.
1. 근거 없이 단정한 내용
2. 앞뒤가 어긋나는 내용
3. 숫자·날짜·이름의 오류
4. 원래 요청에서 빠진 항목
5. 독자가 오해할 수 있는 표현

출력 형식:
판정: 통과 | 수정 필요

지적:
1. 문제가 있는 문장
2. 문제가 되는 이유
3. 무엇을 확인하거나 수정해야 하는지

지적은 최대 5개로 제한하세요.
문제가 없다면 억지로 만들지 말고 통과라고 답하세요.

루프
작업 → 검사 → 수정 → 재검사
              ↑        ↓
              └────────┘

그래프
          → 조사자 A →
기획자 → 조사자 B → 통합자 → 검사자 → 결과
          → 조사자 C →              ↓
                       ← 문제 노드로 되돌림

[기획자]
비교 대상과 기준 확정
    ↓
[조사자 A] ─┐
[조사자 B] ─┤
[조사자 C] ─┼→ [통합 작성자] → [검사자] → 통과 → [최종 보고서]
[조사자 D] ─┤                     ↓
[조사자 E] ─┘              문제가 있는 노드만 재실행

너는 기획 노드다.

요청을 읽고 다음 항목만 결정한다.
1. 조사할 대상
2. 모든 대상에 공통으로 적용할 비교 기준
3. 반드시 확인해야 하는 숫자와 날짜
4. 최종 결과물의 형식
5. 조사 완료를 판단할 기준

직접 조사하거나 결론을 내리지 마라.
출력은 작업 명세서 형식으로 작성하라.

너는 조사만 하는 노드다.
글을 쓰거나 최종 결론을 내리지 마라.

맡은 대상:
[경쟁사 이름]

확인할 기준:
[기획 노드가 정한 공통 기준]

다음 형식으로만 반환한다.

## 확인된 사실
- 한 줄 사실 — 출처: URL 또는 문서명

## 확인하지 못한 것
- 확인되지 않은 주장 — 확인하지 못한 이유

규칙:
- 사실 한 줄마다 출처 하나를 붙인다.
- 출처가 없으면 사실로 포함하지 않는다.
- 숫자, 날짜, 가격, 고유명사는 원문 그대로 옮긴다.
- 추측으로 빈칸을 채우지 않는다.
- 원문 전체나 링크 목록을 덤프하지 않는다.
- 결과는 15줄 이내로 제한한다.

너는 조사 노트를 통합하는 노드다.

규칙:
1. 중복된 사실은 하나로 합친다.
2. 출처가 없는 문장은 버린다.
3. 서로 충돌하는 숫자는 임의로 선택하지 않는다.
4. 충돌한 정보는 각각의 출처와 함께 표시한다.
5. 확인하지 못한 내용은 사실과 분리한다.
6. 새로운 사실을 추가하거나 직접 조사하지 않는다.

최종 작성자에게 전달할 정리 노트만 출력한다.

너는 작성만 하는 노드다.
직접 조사하지 마라.

원래 요청과 전달받은 정리 노트만 보고 결과물을 작성한다.
노트에 없는 사실은 사용하지 않는다.
부족한 내용은 “자료가 부족해 작성하지 못한 부분”으로 따로 표시한다.
확인하지 못한 내용은 단정하지 말고 “확인 필요”라고 표시한다.
숫자, 날짜, 가격, 이름은 노트의 표기를 그대로 사용한다.
본문만 출력하고 작업 과정은 설명하지 않는다.

너는 이 결과물을 처음 보는 검사 노드다.

잘한 점은 쓰지 마라.
“전반적으로 좋습니다” 같은 평가는 금지한다.
결과물을 대신 고쳐 쓰지 마라.
어디가 왜 틀렸는지만 지적한다.

검사할 것:
1. 노트에 없는 주장이 추가됐는가?
2. 숫자, 날짜, 가격, 이름이 바뀌었는가?
3. 앞뒤 논리가 어긋나는가?
4. 원래 요청에서 빠진 항목이 있는가?
5. 확인하지 못한 내용을 사실처럼 단정했는가?

출력:
판정: 통과 | 수정 필요

## 지적
1. 위치 — 문제인 이유 — 확인하거나 수정할 것

지적은 최대 5개로 제한한다.
문제가 없다면 억지로 만들지 말고 통과라고 답한다.

- 통과하면 최종 결과물을 출력한다.
- 수정 필요라면 지적 목록만 작성 노드에 전달한다.
- 작성 노드는 지적받은 부분만 수정한다.
- 수정한 결과를 새로운 검사 노드에 보낸다.
- 최대 두 번까지만 되돌린다.
- 두 번 후에도 실패하면 남은 문제를 보여주고 사람에게 넘긴다.

from concurrent.futures import ThreadPoolExecutor, as_completed

MAX_REVISIONS = 2


def call_model(role: str, payload: dict) -> dict:
    """실제 환경에서는 여기에서 AI API를 호출합니다."""
    raise NotImplementedError


def plan(request: str) -> dict:
    return call_model("planner", {"request": request})


def research(company: str, criteria: list[str]) -> dict:
    return call_model(
        "researcher",
        {
            "company": company,
            "criteria": criteria,
            "rules": [
                "Every fact must have one source.",
                "Do not guess missing values.",
                "Preserve numbers, dates, prices and names.",
            ],
        },
    )


def research_in_parallel(companies: list[str], criteria: list[str]) -> list[dict]:
    if not companies:
        return []

    notes = []
    worker_count = min(5, len(companies))

    with ThreadPoolExecutor(max_workers=worker_count) as executor:
        jobs = {
            executor.submit(research, company, criteria): company
            for company in companies
        }

        for job in as_completed(jobs):
            company = jobs[job]

            try:
                notes.append(job.result())
            except Exception as error:
                notes.append({
                    "company": company,
                    "status": "failed",
                    "error": str(error),
                })

    return notes


def synthesize(request: str, notes: list[dict]) -> str:
    return call_model(
        "writer",
        {
            "request": request,
            "notes": notes,
            "rule": "Use only facts contained in the notes.",
        },
    )


def verify(request: str, notes: list[dict], draft: str) -> dict:
    return call_model(
        "verifier",
        {"request": request, "notes": notes, "draft": draft},
    )


def revise(draft: str, findings: list[dict]) -> str:
    return call_model(
        "reviser",
        {
            "draft": draft,
            "findings": findings,
            "rule": "Revise only the sections identified by the verifier.",
        },
    )


def run_competitor_report(request: str) -> dict:
    specification = plan(request)
    notes = research_in_parallel(
        specification["companies"],
        specification["criteria"],
    )
    draft = synthesize(request, notes)

    for revision_count in range(MAX_REVISIONS + 1):
        review = verify(request, notes, draft)

        if review["decision"] == "pass":
            return {
                "status": "complete",
                "report": draft,
                "revision_count": revision_count,
                "unresolved": [],
            }

        if revision_count == MAX_REVISIONS:
            return {
                "status": "needs_human_review",
                "report": draft,
                "revision_count": revision_count,
                "unresolved": review["findings"],
            }

        draft = revise(draft, review["findings"])

    raise RuntimeError("Unreachable state")

다음 업무를 순차 실행이 아니라 제어 가능한 워크플로 그래프로 설계해줘.

업무:
[구체적인 업무를 적습니다.]

먼저 이 일이 그래프를 쓸 만큼 복잡한지 판단해줘.

다음 네 가지 중 하나도 해당하지 않으면 그래프를 만들지 말고,
단일 에이전트와 명확한 완료 기준만 제안해줘.

1. 단계마다 서로 다른 전문성이 필요한가?
2. 서로 독립적인 작업을 병렬로 실행할 수 있는가?
3. 단계마다 필요한 도구나 권한이 다른가?
4. 하나의 검사자가 여러 기준을 계속 놓치고 있는가?

그래프가 필요하다면 다음을 설계해줘.

1. 각 노드의 이름과 단 하나의 책임
2. 각 노드가 받을 입력과 반환할 출력
3. 순차 실행되는 구간
4. 병렬 실행되는 구간
5. 병렬 결과가 합쳐지는 지점
6. 결과에 따른 분기 조건
7. 실패했을 때 돌아갈 노드
8. 전체가 아니라 실패한 노드만 재실행하는 방법
9. 각 노드가 사용할 수 있는 도구와 권한
10. 최대 재시도 횟수와 사람에게 넘기는 조건
11. 저장해야 할 상태와 체크포인트
12. 각 실행의 비용과 결과를 기록하는 방법

마지막에는 아래를 제공해줘.

- 전체 흐름도
- 노드별 프롬프트
- 상태 데이터 구조
- 실행 가능한 코드
- 테스트 시나리오
- 실패와 복구 시나리오

[하네스]
파일 시스템 · 웹 도구 · API 권한 · 상태 저장 · 로그 · 승인 규칙
    └── [그래프]
        기획 → 병렬 조사 → 통합 → 작성 → 검사 → 승인
                                └── [루프]
                                    검사 실패 → 수정 → 재검사

이 작업은 다음 조건을 모두 충족하면 끝난다.
1. 필수 항목 다섯 개가 포함된다.
2. 모든 숫자에 출처가 있다.
3. 깨진 링크가 없다.
4. 검사 결과가 통과다.
5. 최대 두 번까지만 수정한다.