🤖

본 콘텐츠의 이미지 및 내용은 AI로 생성되었습니다.

⚠️

본 콘텐츠의 이미지 및 내용을 무단으로 복제, 배포, 수정하여 사용할 경우 저작권법에 의해 법적 제재를 받을 수 있습니다.

이미지 로딩 중...

Human-in-the-Loop Workflow 완벽 가이드 - 슬라이드 1/6
A

AI Generated

2025. 12. 27. · 3 Views

Human-in-the-Loop Workflow 완벽 가이드

AI 시스템에서 인간의 판단과 승인을 통합하는 Human-in-the-Loop 워크플로를 알아봅니다. 자동화와 인간 감독의 균형을 맞추는 핵심 패턴을 초급자도 이해할 수 있게 설명합니다.


목차

  1. 인간_승인_대기
  2. 피드백_통합
  3. 재실행_메커니즘
  4. 실습_승인_워크플로
  5. 실습_피드백_루프

1. 인간 승인 대기

어느 날 김개발 씨는 회사의 AI 자동화 시스템을 개발하고 있었습니다. 모든 것이 순조롭게 진행되는 듯했지만, 팀장님이 한 가지 우려를 표했습니다.

"이거 AI가 혼자 결정해서 고객한테 메일 보내도 괜찮은 거야?"

**인간 승인 대기(Human Approval Wait)**는 AI나 자동화 시스템이 중요한 작업을 수행하기 전에 반드시 사람의 확인을 받는 패턴입니다. 마치 은행에서 큰 금액을 이체할 때 한 번 더 확인하는 것과 같습니다.

이 패턴을 적용하면 자동화의 효율성을 유지하면서도 치명적인 실수를 방지할 수 있습니다.

다음 코드를 살펴봅시다.

import asyncio
from enum import Enum
from dataclasses import dataclass

class ApprovalStatus(Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"

@dataclass
class ApprovalRequest:
    task_id: str
    description: str
    payload: dict
    status: ApprovalStatus = ApprovalStatus.PENDING

class ApprovalGate:
    def __init__(self):
        self.pending_requests = {}

    async def request_approval(self, task_id: str, description: str, payload: dict) -> bool:
        # 승인 요청을 생성하고 대기열에 추가합니다
        request = ApprovalRequest(task_id, description, payload)
        self.pending_requests[task_id] = request

        # 인간의 승인을 기다립니다
        while request.status == ApprovalStatus.PENDING:
            await asyncio.sleep(1)  # 폴링 간격

        return request.status == ApprovalStatus.APPROVED

    def approve(self, task_id: str):
        if task_id in self.pending_requests:
            self.pending_requests[task_id].status = ApprovalStatus.APPROVED

    def reject(self, task_id: str):
        if task_id in self.pending_requests:
            self.pending_requests[task_id].status = ApprovalStatus.REJECTED

김개발 씨는 입사 6개월 차 주니어 개발자입니다. 요즘 그가 맡은 프로젝트는 고객 문의에 자동으로 응답하는 AI 시스템 구축이었습니다.

처음에는 모든 것을 완전 자동화하면 된다고 생각했습니다. 그런데 테스트 중에 문제가 발생했습니다.

AI가 고객의 환불 요청에 대해 "죄송합니다만 환불이 불가능합니다"라고 자동 응답을 보낸 것입니다. 알고 보니 해당 고객은 VIP 회원이었고, 회사 정책상 VIP에게는 무조건 환불해주기로 되어 있었습니다.

선배 개발자 박시니어 씨가 상황을 파악하고 조언했습니다. "AI가 아무리 똑똑해도 모든 상황을 완벽하게 판단할 수는 없어요.

중요한 결정에는 사람이 개입해야 해요." 그렇다면 인간 승인 대기 패턴이란 정확히 무엇일까요? 쉽게 비유하자면, 이것은 마치 회사에서 중요한 계약서에 결재를 받는 것과 같습니다.

담당자가 아무리 계약 내용을 잘 작성했어도, 최종 결정은 상사의 서명이 있어야 효력이 발생합니다. 마찬가지로 AI 시스템도 중요한 작업 전에 "이렇게 해도 될까요?"라고 물어보고 기다리는 것입니다.

이 패턴이 없던 시절에는 어땠을까요? 완전 자동화 시스템은 실수를 해도 멈추지 않았습니다.

잘못된 이메일이 수천 명에게 발송되거나, 부적절한 콘텐츠가 게시되거나, 중요한 데이터가 삭제되는 사고가 발생했습니다. 한번 벌어진 일은 되돌리기 어려웠습니다.

바로 이런 문제를 해결하기 위해 인간 승인 대기 패턴이 등장했습니다. AI가 작업을 준비한 뒤, 실행하기 전에 일시 정지 상태로 들어갑니다.

담당자가 내용을 검토하고 승인 버튼을 누르면 그제야 작업이 실행됩니다. 위의 코드를 살펴보겠습니다.

ApprovalGate 클래스가 핵심입니다. request_approval 메서드는 승인 요청을 생성하고, 상태가 PENDING인 동안 계속 대기합니다.

사람이 approve 또는 reject 메서드를 호출하면 그때서야 대기가 풀립니다. 실제 현업에서는 이 패턴을 다양하게 활용합니다.

예를 들어 마케팅 자동화 시스템에서 대량 이메일 발송 전에 마케터의 최종 확인을 받습니다. 금융 시스템에서는 일정 금액 이상의 거래에 관리자 승인을 요구합니다.

콘텐츠 플랫폼에서는 AI가 생성한 글을 에디터가 검토한 후 게시합니다. 하지만 주의할 점도 있습니다.

모든 작업에 승인을 요구하면 자동화의 의미가 없어집니다. 임계값을 설정하는 것이 중요합니다.

위험도가 낮은 작업은 자동으로 처리하고, 위험도가 높은 작업만 승인을 요청해야 합니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.

박시니어 씨의 조언을 듣고 승인 게이트를 구현한 후, 시스템은 안정적으로 운영되기 시작했습니다. AI가 95%의 일상적인 문의는 자동 처리하고, 환불이나 보상 같은 민감한 요청은 담당자에게 넘겼습니다.

인간 승인 대기 패턴을 제대로 이해하면 자동화와 안전성이라는 두 마리 토끼를 잡을 수 있습니다.

실전 팁

💡 - 승인 대기 시간에 타임아웃을 설정하여 작업이 무한정 대기하지 않도록 하세요

  • 승인 요청 시 충분한 컨텍스트 정보를 함께 제공해야 담당자가 빠르게 판단할 수 있습니다

2. 피드백 통합

김개발 씨가 만든 AI 시스템이 어느 정도 안정화되었습니다. 그런데 이번에는 다른 고민이 생겼습니다.

담당자가 AI의 제안을 거절했을 때, "왜 거절했는지" 정보가 사라져버리는 것이었습니다. "이 피드백을 어떻게든 활용할 수 없을까요?"

**피드백 통합(Feedback Integration)**은 인간이 제공한 수정사항이나 의견을 시스템에 반영하는 메커니즘입니다. 마치 원고를 편집자에게 보내면 빨간 펜으로 수정 사항을 받아 고치는 것과 같습니다.

이를 통해 시스템은 점점 더 사람의 기대에 맞게 발전할 수 있습니다.

다음 코드를 살펴봅시다.

from dataclasses import dataclass, field
from typing import Optional, List
from datetime import datetime

@dataclass
class Feedback:
    reviewer: str
    comment: str
    suggestion: Optional[str] = None
    timestamp: datetime = field(default_factory=datetime.now)

@dataclass
class WorkItem:
    content: str
    feedbacks: List[Feedback] = field(default_factory=list)
    revision_count: int = 0

    def add_feedback(self, reviewer: str, comment: str, suggestion: str = None):
        # 피드백을 기록합니다
        feedback = Feedback(reviewer, comment, suggestion)
        self.feedbacks.append(feedback)

    def apply_suggestion(self, feedback_index: int) -> str:
        # 제안된 수정사항을 적용합니다
        if feedback_index < len(self.feedbacks):
            suggestion = self.feedbacks[feedback_index].suggestion
            if suggestion:
                self.content = suggestion
                self.revision_count += 1
                return f"수정 완료: 버전 {self.revision_count}"
        return "적용할 제안이 없습니다"

    def get_feedback_summary(self) -> str:
        # 모든 피드백을 요약합니다
        summary = [f"총 {len(self.feedbacks)}개의 피드백"]
        for i, fb in enumerate(self.feedbacks):
            summary.append(f"  [{i}] {fb.reviewer}: {fb.comment}")
        return "\n".join(summary)

김개발 씨의 AI 시스템은 이제 제법 그럴듯하게 동작했습니다. 하지만 한 가지 문제가 있었습니다.

담당자들이 AI의 제안을 수정할 때마다 그 수정 내용이 그냥 사라져버렸습니다. 어느 날 마케팅팀 이과장 님이 찾아왔습니다.

"저번에 고객 응대 문구를 수정했는데, 똑같은 상황에서 또 이상한 문구가 나왔어요. 제가 고친 건 어디로 간 거예요?" 박시니어 씨가 설명했습니다.

"피드백을 받기만 하고 저장을 안 했군요. 피드백 통합 시스템을 만들어야 해요." 그렇다면 피드백 통합이란 무엇일까요?

비유하자면, 이것은 마치 학교에서 선생님이 채점한 시험지와 같습니다. 단순히 점수만 받는 것이 아니라, 어디가 틀렸고 왜 틀렸는지 코멘트를 받습니다.

학생은 그 코멘트를 보고 같은 실수를 반복하지 않게 됩니다. AI 시스템도 마찬가지로 사람의 피드백을 기록하고 학습해야 합니다.

피드백 통합이 없으면 어떤 문제가 생길까요? 같은 실수가 계속 반복됩니다.

담당자는 같은 수정을 몇 번이고 반복해야 합니다. 시간이 지나도 시스템이 전혀 나아지지 않습니다.

이것은 마치 밑 빠진 독에 물 붓기와 같습니다. 바로 이런 문제를 해결하기 위해 피드백 통합 패턴이 필요합니다.

모든 피드백을 구조화된 형태로 저장합니다. 누가, 언제, 무엇을, 왜 수정했는지 기록합니다.

나중에 이 데이터를 분석하여 시스템을 개선할 수 있습니다. 위의 코드를 살펴보겠습니다.

Feedback 클래스는 리뷰어 이름, 코멘트, 제안 내용, 시간을 저장합니다. WorkItem 클래스의 add_feedback 메서드는 새로운 피드백을 추가하고, apply_suggestion 메서드는 제안된 수정사항을 실제로 적용합니다.

실제 현업에서는 피드백 데이터가 금광과 같습니다. 어떤 유형의 콘텐츠가 자주 수정되는지 분석하면 AI 모델의 약점을 파악할 수 있습니다.

특정 담당자의 피드백 패턴을 학습하면 개인화된 결과물을 생성할 수 있습니다. 피드백 이력을 추적하면 품질이 어떻게 변화하는지 측정할 수 있습니다.

주의할 점도 있습니다. 피드백을 무조건 수용하면 안 됩니다.

때로는 담당자의 수정이 잘못된 경우도 있습니다. 여러 사람의 피드백이 충돌할 수도 있습니다.

따라서 피드백의 신뢰도를 평가하고 우선순위를 정하는 로직이 필요합니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.

피드백 통합 시스템을 구축한 후, 놀라운 변화가 일어났습니다. 한 달간 축적된 피드백 데이터를 분석해보니, AI가 특히 "배송 관련 문의"에서 자주 틀렸습니다.

이 인사이트를 바탕으로 해당 영역의 프롬프트를 개선하자 거절률이 30% 감소했습니다. 피드백 통합을 제대로 구현하면 시스템은 사용할수록 똑똑해집니다.

실전 팁

💡 - 피드백에는 반드시 "이유"를 함께 기록하세요. 무엇을 수정했는지보다 왜 수정했는지가 더 중요합니다

  • 주기적으로 피드백 데이터를 분석하여 시스템 개선에 활용하세요

3. 재실행 메커니즘

피드백 시스템이 잘 동작하게 되었습니다. 그런데 새로운 요구사항이 들어왔습니다.

담당자가 피드백을 주면 AI가 그걸 반영해서 다시 결과물을 만들어야 한다는 것이었습니다. "수동으로 다시 실행하지 않아도 되게 해주세요."

**재실행 메커니즘(Re-execution Mechanism)**은 피드백을 받은 후 이를 반영하여 작업을 다시 수행하는 패턴입니다. 마치 음식점에서 손님이 "덜 짜게 해주세요"라고 요청하면 주방에서 새로 요리하는 것과 같습니다.

이 메커니즘을 통해 사람과 AI 사이의 반복적인 협업이 가능해집니다.

다음 코드를 살펴봅시다.

from typing import Callable, Dict, Any
from dataclasses import dataclass

@dataclass
class ExecutionContext:
    original_input: Dict[str, Any]
    feedback_history: list = None
    retry_count: int = 0
    max_retries: int = 3

    def __post_init__(self):
        if self.feedback_history is None:
            self.feedback_history = []

class ReExecutor:
    def __init__(self, task_fn: Callable, feedback_processor: Callable):
        self.task_fn = task_fn
        self.feedback_processor = feedback_processor

    def execute(self, context: ExecutionContext) -> Dict[str, Any]:
        # 피드백을 반영하여 입력을 수정합니다
        modified_input = self.feedback_processor(
            context.original_input,
            context.feedback_history
        )

        # 수정된 입력으로 작업을 재실행합니다
        result = self.task_fn(modified_input)
        context.retry_count += 1

        return {
            "result": result,
            "retry_count": context.retry_count,
            "can_retry": context.retry_count < context.max_retries
        }

    def add_feedback_and_retry(self, context: ExecutionContext, feedback: str):
        if context.retry_count >= context.max_retries:
            return {"error": "최대 재시도 횟수를 초과했습니다"}

        context.feedback_history.append(feedback)
        return self.execute(context)

김개발 씨는 요즘 행복한 고민에 빠졌습니다. 시스템이 점점 복잡해지면서 새로운 요구사항들이 쏟아져 들어왔기 때문입니다.

그중 하나가 "AI가 피드백을 받으면 알아서 다시 해줬으면 좋겠어요"라는 것이었습니다. 현재 시스템은 이랬습니다.

AI가 결과물을 만들고, 담당자가 피드백을 주고, 개발자가 피드백을 보고 수동으로 다시 실행하고. 이 과정이 하루에도 수십 번 반복되었습니다.

박시니어 씨가 화이트보드에 그림을 그리며 설명했습니다. "재실행 메커니즘이 필요해요.

피드백이 들어오면 자동으로 그걸 반영해서 다시 돌리는 거죠." 그렇다면 재실행 메커니즘이란 무엇일까요? 쉽게 비유하자면, 이것은 마치 미용실에서 머리를 자르는 과정과 같습니다.

미용사가 처음 커트한 후 거울을 보여줍니다. 손님이 "앞머리를 조금 더 짧게요"라고 하면 미용사가 다시 손질합니다.

이 과정이 손님이 만족할 때까지 반복됩니다. AI 시스템도 똑같습니다.

재실행 메커니즘이 없으면 어떤 문제가 생길까요? 매번 사람이 중간에서 작업을 다시 트리거해야 합니다.

피드백이 제대로 반영되었는지 확인하기 어렵습니다. 히스토리 관리가 안 되어 같은 실수가 반복됩니다.

전체 프로세스가 느려지고 비효율적이 됩니다. 바로 이런 문제를 해결하기 위해 재실행 메커니즘이 등장했습니다.

피드백이 들어오면 자동으로 이전 맥락과 함께 작업을 다시 수행합니다. 몇 번째 시도인지 추적하고, 최대 재시도 횟수를 제한하여 무한 루프를 방지합니다.

위의 코드를 살펴보겠습니다. ExecutionContext는 원본 입력, 피드백 히스토리, 재시도 횟수를 관리합니다.

ReExecutor 클래스의 add_feedback_and_retry 메서드가 핵심입니다. 피드백을 히스토리에 추가하고, feedback_processor가 이를 반영한 새로운 입력을 만들어 다시 실행합니다.

실제 현업에서는 이 패턴이 다양하게 활용됩니다. 이미지 생성 AI에서 "더 밝게", "사람을 왼쪽으로"라는 피드백을 받아 자동으로 재생성합니다.

문서 요약 시스템에서 "더 짧게", "핵심 수치 포함"이라는 요청을 반영합니다. 코드 생성 AI에서 "에러 처리 추가", "성능 최적화"라는 피드백으로 코드를 개선합니다.

주의할 점도 있습니다. 반드시 최대 재시도 횟수를 설정해야 합니다.

그렇지 않으면 사람과 AI가 끝없이 핑퐁을 주고받는 상황이 발생할 수 있습니다. 또한 매 시도마다 피드백이 실제로 반영되었는지 확인하는 검증 단계도 필요합니다.

다시 김개발 씨의 이야기로 돌아가 봅시다. 재실행 메커니즘을 구현한 후, 담당자들의 업무 효율이 크게 올랐습니다.

이전에는 피드백 한 번에 5분이 걸렸다면, 이제는 버튼 한 번으로 30초 만에 수정된 결과를 받아볼 수 있게 되었습니다. 재실행 메커니즘을 제대로 구현하면 사람과 AI의 협업이 매끄러워집니다.

실전 팁

💡 - 재시도 횟수 제한을 반드시 설정하세요. 보통 3-5회가 적당합니다

  • 각 재실행 결과를 버전으로 저장하여 비교할 수 있게 하면 유용합니다

4. 실습 승인 워크플로

이론은 충분히 배웠습니다. 이제 김개발 씨와 함께 실제로 승인 워크플로를 구현해보겠습니다.

가상의 시나리오는 이렇습니다. AI가 고객 이메일 초안을 작성하면, 담당자가 검토하고 승인해야만 실제로 발송되는 시스템입니다.

승인 워크플로 실습에서는 앞서 배운 개념들을 조합하여 완전한 워크플로를 구현합니다. 이메일 초안 생성부터 검토, 승인, 발송까지의 전체 흐름을 다룹니다.

실제 프로덕션 환경에서 사용할 수 있는 수준의 코드를 작성해봅니다.

다음 코드를 살펴봅시다.

import asyncio
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime

class WorkflowState(Enum):
    DRAFT = "draft"
    PENDING_REVIEW = "pending_review"
    APPROVED = "approved"
    REJECTED = "rejected"
    SENT = "sent"

@dataclass
class EmailDraft:
    id: str
    recipient: str
    subject: str
    body: str
    state: WorkflowState = WorkflowState.DRAFT
    reviewer: Optional[str] = None
    reviewed_at: Optional[datetime] = None

class EmailApprovalWorkflow:
    def __init__(self):
        self.drafts = {}
        self.on_state_change = None

    def create_draft(self, draft_id: str, recipient: str, subject: str, body: str):
        draft = EmailDraft(draft_id, recipient, subject, body)
        self.drafts[draft_id] = draft
        return draft

    def submit_for_review(self, draft_id: str):
        draft = self.drafts.get(draft_id)
        if draft and draft.state == WorkflowState.DRAFT:
            draft.state = WorkflowState.PENDING_REVIEW
            self._notify_state_change(draft)

    def approve(self, draft_id: str, reviewer: str):
        draft = self.drafts.get(draft_id)
        if draft and draft.state == WorkflowState.PENDING_REVIEW:
            draft.state = WorkflowState.APPROVED
            draft.reviewer = reviewer
            draft.reviewed_at = datetime.now()
            self._notify_state_change(draft)

    def reject(self, draft_id: str, reviewer: str):
        draft = self.drafts.get(draft_id)
        if draft and draft.state == WorkflowState.PENDING_REVIEW:
            draft.state = WorkflowState.REJECTED
            draft.reviewer = reviewer
            draft.reviewed_at = datetime.now()
            self._notify_state_change(draft)

    def _notify_state_change(self, draft: EmailDraft):
        if self.on_state_change:
            self.on_state_change(draft)

김개발 씨는 드디어 실전 프로젝트에 착수했습니다. 고객 서비스팀에서 요청한 "AI 이메일 초안 승인 시스템"을 만들어야 합니다.

팀장님이 말씀하셨습니다. "AI가 만든 이메일이 검토 없이 나가면 큰일 나요.

반드시 승인 단계를 거쳐야 해요." 박시니어 씨와 함께 화이트보드 앞에 섰습니다. "자, 먼저 상태 흐름을 정의해봅시다.

이메일은 어떤 상태들을 거치나요?" 두 사람은 함께 상태를 정의했습니다. DRAFT는 초안이 막 생성된 상태입니다.

PENDING_REVIEW는 검토 대기 중인 상태입니다. APPROVED는 승인된 상태, REJECTED는 거절된 상태입니다.

마지막으로 SENT는 발송 완료 상태입니다. 왜 상태를 명확히 정의해야 할까요?

상태가 모호하면 혼란이 생깁니다. "이 이메일 지금 어떤 상태예요?"라는 질문에 명확하게 답할 수 있어야 합니다.

또한 각 상태에서 할 수 있는 행동과 할 수 없는 행동을 구분해야 합니다. 예를 들어 이미 발송된 이메일을 다시 승인할 수는 없습니다.

위의 코드에서 EmailDraft 클래스를 살펴보겠습니다. 이메일의 모든 정보를 담고 있습니다.

받는 사람, 제목, 본문, 그리고 현재 상태. 누가 언제 검토했는지도 기록합니다.

이 정보들은 나중에 감사 로그로도 활용됩니다. EmailApprovalWorkflow 클래스가 전체 흐름을 관리합니다.

create_draft 메서드로 초안을 생성하면 DRAFT 상태로 시작합니다. submit_for_review를 호출하면 PENDING_REVIEW 상태로 바뀝니다.

이때 담당자에게 알림이 가야겠죠. 담당자가 검토 후 approve를 호출하면 APPROVED 상태가 됩니다.

거절하면 reject로 REJECTED 상태가 됩니다. 각 전환마다 _notify_state_change가 호출되어 외부 시스템에 알림을 보냅니다.

실무에서 이런 워크플로를 구현할 때 중요한 점이 있습니다. 바로 상태 전환 규칙입니다.

코드를 보면 approve 메서드는 현재 상태가 PENDING_REVIEW일 때만 동작합니다. DRAFT 상태의 이메일을 바로 승인하는 것은 불가능합니다.

이런 규칙이 버그와 혼란을 방지합니다. 또 하나 주목할 점은 감사 추적입니다.

누가 언제 승인했는지 기록합니다. 나중에 "이 이메일 누가 승인했어요?"라는 질문에 답할 수 있어야 합니다.

컴플라이언스가 중요한 금융이나 의료 분야에서는 필수입니다. 김개발 씨가 물었습니다.

"만약 담당자가 오랫동안 응답하지 않으면 어떻게 하나요?" 좋은 질문입니다. 실제 시스템에서는 타임아웃에스컬레이션 로직을 추가해야 합니다.

24시간 내 검토가 없으면 상위 관리자에게 알림을 보내는 식입니다. 테스트 코드를 작성해봅시다.

워크플로가 정상적으로 동작하는지, 비정상적인 상태 전환은 거부하는지 확인해야 합니다.

실전 팁

💡 - 상태 전환마다 로그를 남겨 문제 발생 시 추적할 수 있게 하세요

  • 상태별로 허용되는 액션을 문서화하여 팀원들과 공유하세요

5. 실습 피드백 루프

승인 워크플로가 완성되었습니다. 이제 마지막 단계입니다.

담당자가 단순히 승인/거절만 하는 게 아니라, 수정 의견을 주면 AI가 그걸 반영해서 다시 초안을 만드는 시스템을 구현해봅시다. 진정한 Human-in-the-Loop의 완성입니다.

피드백 루프 실습에서는 AI와 인간이 반복적으로 협업하는 완전한 시스템을 구현합니다. 담당자의 피드백을 받아 AI가 자동으로 개선된 버전을 생성하고, 이 과정이 승인될 때까지 반복됩니다.

앞서 배운 모든 개념이 통합됩니다.

다음 코드를 살펴봅시다.

from dataclasses import dataclass, field
from typing import List, Callable, Optional
from datetime import datetime

@dataclass
class FeedbackItem:
    content: str
    author: str
    timestamp: datetime = field(default_factory=datetime.now)

@dataclass
class DraftVersion:
    version: int
    content: str
    created_at: datetime = field(default_factory=datetime.now)

class FeedbackLoop:
    def __init__(self, ai_generator: Callable[[str, List[str]], str]):
        self.ai_generator = ai_generator
        self.versions: List[DraftVersion] = []
        self.feedbacks: List[FeedbackItem] = []
        self.is_approved = False

    def generate_initial(self, prompt: str) -> DraftVersion:
        # 초기 버전을 생성합니다
        content = self.ai_generator(prompt, [])
        version = DraftVersion(version=1, content=content)
        self.versions.append(version)
        return version

    def add_feedback(self, content: str, author: str) -> FeedbackItem:
        # 피드백을 추가합니다
        feedback = FeedbackItem(content, author)
        self.feedbacks.append(feedback)
        return feedback

    def regenerate(self) -> Optional[DraftVersion]:
        # 피드백을 반영하여 새 버전을 생성합니다
        if not self.versions:
            return None

        feedback_texts = [f.content for f in self.feedbacks]
        original_content = self.versions[0].content

        new_content = self.ai_generator(original_content, feedback_texts)
        new_version = DraftVersion(
            version=len(self.versions) + 1,
            content=new_content
        )
        self.versions.append(new_version)
        return new_version

    def approve(self):
        self.is_approved = True

    def get_history(self) -> dict:
        return {
            "total_versions": len(self.versions),
            "total_feedbacks": len(self.feedbacks),
            "is_approved": self.is_approved,
            "latest_version": self.versions[-1] if self.versions else None
        }

김개발 씨의 프로젝트가 대망의 마지막 단계에 접어들었습니다. 시스템은 이제 승인 워크플로까지 갖추었습니다.

하지만 고객 서비스팀에서 한 가지 더 요청이 들어왔습니다. "이메일이 마음에 안 들면 거절만 할 게 아니라, 수정해달라고 피드백을 주고 싶어요.

그러면 AI가 다시 써주면 좋겠어요." 박시니어 씨가 고개를 끄덕였습니다. "진짜 Human-in-the-Loop를 완성하려면 피드백 루프가 필요해요.

사람과 AI가 핑퐁 치듯이 주고받으면서 결과물을 개선하는 거죠." 피드백 루프란 무엇일까요? 비유하자면, 이것은 마치 선생님과 학생의 첨삭 과정과 같습니다.

학생이 에세이를 씁니다. 선생님이 읽고 코멘트를 답니다.

"도입부가 약해요. 더 강렬하게 시작하세요." 학생이 수정합니다.

선생님이 다시 봅니다. 이 과정이 완성될 때까지 반복됩니다.

위의 코드에서 FeedbackLoop 클래스를 살펴보겠습니다. 생성자에서 ai_generator를 받습니다.

이것은 프롬프트와 피드백 목록을 받아 새로운 콘텐츠를 생성하는 함수입니다. 실제로는 LLM API를 호출하는 함수가 될 것입니다.

generate_initial 메서드로 첫 번째 버전을 만듭니다. 아직 피드백이 없으므로 빈 리스트를 전달합니다.

결과물은 DraftVersion 객체로 저장됩니다. 버전 번호와 생성 시간이 함께 기록됩니다.

담당자가 초안을 검토하고 피드백을 줍니다. "너무 딱딱해요.

좀 더 친근한 톤으로 바꿔주세요." add_feedback 메서드로 이 피드백을 저장합니다. 누가 언제 어떤 피드백을 줬는지 모두 기록됩니다.

이제 핵심인 regenerate 메서드입니다. 지금까지 축적된 모든 피드백을 모아서 AI 생성기에 전달합니다.

AI는 원본 콘텐츠와 피드백들을 보고 개선된 버전을 만들어냅니다. 새로운 버전이 목록에 추가되고, 버전 번호가 증가합니다.

이 과정은 담당자가 approve를 호출할 때까지 계속될 수 있습니다. 첫 번째 피드백으로 두 번째 버전이 나왔는데 아직 마음에 안 들면, 다시 피드백을 주고 세 번째 버전을 받으면 됩니다.

get_history 메서드는 전체 히스토리를 요약합니다. 총 몇 번의 버전이 만들어졌는지, 피드백은 몇 개가 있었는지, 최종 승인되었는지를 한눈에 볼 수 있습니다.

이 정보는 품질 관리와 프로세스 개선에 유용합니다. 실무에서 피드백 루프를 구현할 때 고려할 점이 있습니다.

피드백의 맥락을 AI에게 잘 전달해야 합니다. 단순히 "더 친근하게"보다는 "현재 버전이 너무 형식적입니다.

마치 친구에게 말하듯이 편하게 작성해주세요"처럼 구체적인 지시가 좋습니다. 또한 무한 루프를 방지해야 합니다.

버전이 10번, 20번 쌓이면 문제가 있는 것입니다. 최대 반복 횟수를 정하고, 그 안에 해결되지 않으면 사람이 직접 수정하도록 에스컬레이션해야 합니다.

김개발 씨가 모든 기능을 통합한 시스템을 완성했습니다. AI가 초안을 만들고, 담당자가 피드백을 주고, AI가 개선하고, 다시 검토하고, 최종 승인되면 발송됩니다.

진정한 Human-in-the-Loop 워크플로가 완성되었습니다.

실전 팁

💡 - AI에게 피드백을 전달할 때는 이전 버전과 함께 보내 맥락을 유지하세요

  • 버전별 변경 사항을 하이라이트하면 담당자가 빠르게 검토할 수 있습니다

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

#Python#HITL#Workflow#Automation#LLM#LLM,HITL,워크플로

댓글 (0)

댓글을 작성하려면 로그인이 필요합니다.

함께 보면 좋은 카드 뉴스

Context Fundamentals - AI 컨텍스트의 기본 원리

AI 에이전트 개발의 핵심인 컨텍스트 관리를 다룹니다. 시스템 프롬프트 구조부터 Attention Budget, Progressive Disclosure까지 실무에서 바로 적용할 수 있는 컨텍스트 최적화 전략을 배웁니다.

Phase 1 보안 사고방식 구축 완벽 가이드

초급 개발자가 보안 전문가로 성장하기 위한 첫걸음입니다. 해커의 관점에서 시스템을 바라보는 방법부터 OWASP Top 10, 포트 스캐너 구현, 실제 침해사고 분석까지 보안의 기초 체력을 다집니다.

프로덕션 워크플로 배포 완벽 가이드

LLM 기반 애플리케이션을 실제 운영 환경에 배포하기 위한 워크플로 최적화, 캐싱 전략, 비용 관리 방법을 다룹니다. Airflow와 서버리스 아키텍처를 활용한 실습까지 포함하여 초급 개발자도 프로덕션 수준의 배포를 할 수 있도록 안내합니다.

워크플로 모니터링과 디버깅 완벽 가이드

LLM 기반 워크플로의 실행 상태를 추적하고, 문제를 진단하며, 성능을 최적화하는 방법을 다룹니다. LangSmith 통합부터 커스텀 모니터링 시스템 구축까지 실무에서 바로 적용할 수 있는 내용을 담았습니다.

LlamaIndex Workflow 완벽 가이드

LlamaIndex의 워크플로 시스템을 활용하여 복잡한 RAG 파이프라인을 구축하는 방법을 알아봅니다. 이벤트 기반 워크플로부터 멀티 인덱스 쿼리까지 단계별로 학습합니다.