Human Feedback in Flows

Human Feedback in Flows (플로우 안의 사람 피드백)

AI가 만든 결과를 바로 내보내기 전에 사람의 확인을 거치게 하려면, Flow 실행을 멈추고 피드백을 받아 처리 흐름을 이어가는 게 필요해요. CrewAI의 @human_feedback 데코레이터는 바로 이 휴먼 인 더 루프(HITL) 워크플로를 Flow 안에 직접 넣어 줍니다. 품질 검수, 결정 게이트, 승인/거절/수정 패턴, 반복 개선 같은 시나리오에 적합해요.

출처: 공식문서

본문

@human_feedback 데코레이터는 CrewAI 1.8.0 이상이 필요합니다. 사용 전에 설치 버전을 확인하세요.

@human_feedback은 Flow 실행을 멈추고, 산출물을 사람에게 보여 리뷰하게 한 뒤, 피드백을 수집하고 (선택적으로) 피드백 결과에 따라 다른 리스너로 라우팅합니다.

flowchart LR
    A[Flow Method] --> B[Output Generated]
    B --> C[Human Reviews]
    C --> D{Feedback}
    D -->|emit specified| E[LLM Collapses to Outcome]
    D -->|no emit| F[HumanFeedbackResult]
    E --> G["@listen('approved')"]
    E --> H["@listen('rejected')"]
    F --> I[Next Listener]

빠른 시작

from crewai.flow.flow import Flow, start, listen
from crewai.flow.human_feedback import human_feedback

class SimpleReviewFlow(Flow):
    @start()
    @human_feedback(message="Please review this content:")
    def generate_content(self):
        return "This is AI-generated content that needs review."

    @listen(generate_content)
    def process_feedback(self, result):
        print(f"Content: {result.output}")
        print(f"Human said: {result.feedback}")

flow = SimpleReviewFlow()
flow.kickoff()

이 Flow는 ① generate_content를 실행하고 결과를 반환하고, ② 요청 메시지와 함께 출력을 사용자에게 보여주고, ③ 사용자가 피드백을 입력할 때까지 기다렸다가(Enter로 건너뛰기 가능), ④ HumanFeedbackResult 객체를 process_feedback에 전달합니다.

@human_feedback 데코레이터

파라미터

기본 사용 (라우팅 없음)

emit을 지정하지 않으면 데코레이터는 피드백만 수집하고 다음 리스너에 HumanFeedbackResult를 전달합니다.

@start()
@human_feedback(message="What do you think of this analysis?")
def analyze_data(self):
    return "Analysis results: Revenue up 15%, costs down 8%"

@listen(analyze_data)
def handle_feedback(self, result):
    # result is a HumanFeedbackResult
    print(f"Analysis: {result.output}")
    print(f"Feedback: {result.feedback}")

emit으로 라우팅

emit을 지정하면 데코레이터가 라우터가 됩니다. 사람의 자유 형식 피드백을 LLM이 해석해 지정된 결과 중 하나로 축약합니다.

from crewai.flow.flow import Flow, start, listen, or_
from crewai.flow.human_feedback import human_feedback

class ReviewFlow(Flow):
    @start()
    def generate_content(self):
        return "Draft blog post content here..."

    @human_feedback(
        message="Do you approve this content for publication?",
        emit=["approved", "rejected", "needs_revision"],
        llm="gpt-4o-mini",
        default_outcome="needs_revision",
    )
    @listen(or_("generate_content", "needs_revision"))
    def review_content(self):
        return "Draft blog post content here..."

    @listen("approved")
    def publish(self, result):
        print(f"Publishing! User said: {result.feedback}")

    @listen("rejected")
    def discard(self, result):
        print(f"Discarding. Reason: {result.feedback}")

사람이 "more detail이 필요해"라고 하면 LLM이 이를 "needs_revision"으로 축약해 or_()를 통해 review_content를 다시 실행하는 수정 루프를 만듭니다. 결과가 "approved"나 "rejected"가 될 때까지 계속되죠.

LLM은 사용 가능할 때 구조화 출력(함수 호출)을 사용해 응답이 지정된 결과 중 하나임을 보장합니다. 이렇게 하면 라우팅이 신뢰성 있고 예측 가능해져요.

@start() 메서드는 Flow 시작 시 한 번만 실행됩니다. 수정 루프가 필요하면 시작 메서드와 리뷰 메서드를 분리하고, 리뷰 메서드에 @listen(or_("trigger", "revision_outcome"))을 써서 셀프 루프를 활성화하세요.

HumanFeedbackResult

HumanFeedbackResult dataclass에 사람 피드백 상호작용의 모든 정보가 담깁니다.

from crewai.flow.human_feedback import HumanFeedbackResult

@dataclass
class HumanFeedbackResult:
    output: Any              # The original method output shown to the human
    feedback: str            # The raw feedback text from the human
    outcome: str | None      # The collapsed outcome (if emit was specified)
    timestamp: datetime      # When the feedback was received
    method_name: str         # Name of the decorated method
    metadata: dict           # Any metadata passed to the decorator

emit이 있는 @human_feedback 메서드에 의해 트리거된 리스너는 HumanFeedbackResult를 받습니다.

@listen("approved")
def on_approval(self, result: HumanFeedbackResult):
    print(f"Original output: {result.output}")
    print(f"User feedback: {result.feedback}")
    print(f"Outcome: {result.outcome}")  # "approved"
    print(f"Received at: {result.timestamp}")

피드백 히스토리 접근

Flow 클래스는 피드백에 접근하는 두 가지 속성을 제공합니다.

• last_human_feedback — 가장 최근의 HumanFeedbackResult
• human_feedback_history — Flow 중 수집된 모든 HumanFeedbackResult의 리스트

각 HumanFeedbackResult는 human_feedback_history에 추가되므로 여러 피드백 단계가 서로 덮어쓰지 않습니다. 이 리스트로 Flow 중 수집된 모든 피드백에 접근할 수 있어요.

결합 패턴

@human_feedback은 다른 데코레이터와 함께 쓸 수 있습니다.

# One-shot review at the start of a flow (no self-loop)
@start()
@human_feedback(message="Approve?")
def generate(self):
    ...

# Linear review on a listener (no self-loop)
@listen(some_step)
@human_feedback(message="Approve?")
def review(self):
    ...

# Self-loop: review that can loop back for revisions
@start()
def generate(self):
    ...

@listen(or_("generate", "needs_revision"))
@human_feedback(message="Approve?", emit=["approved", "needs_revision"])
def review(self):
    ...

# Chained routers
@listen("approved")
def next_step(self):
    ...

모범 사례

1. 명확한 요청 메시지를 쓴다.

# ✅ Good - clear and actionable
@human_feedback(message="Approve this draft for publication? Reply 'needs_revision' to request edits.")

# ❌ Bad - vague
@human_feedback(message="Check this.")

2. 의미 있는 결과를 고른다.

# ✅ Good - natural language outcomes
emit=["approved", "rejected", "needs_revision"]

# ❌ Bad - technical or unclear
emit=["pass", "fail"]

3. 항상 기본 결과를 제공한다. 피드백이 없을 때 사용할 결과를 지정하세요 (default_outcome는 emit에 있어야 합니다).
4. 감사 추적에 피드백 히스토리를 사용한다. human_feedback_history로 리뷰 내역을 추적하세요.
5. 라우팅 및 비라우팅 피드백을 모두 처리한다. result.feedback과 result.outcome을 함께 확인하세요.

비동기(논블로킹) 사람 피드백

기본적으로 @human_feedback은 콘솔 입력을 기다리며 실행을 블로킹합니다. 프로덕션에서는 Slack, 이메일, 웹훅, API 같은 외부 시스템과 통합되는 async/논블로킹 피드백이 필요할 수 있어요.

Provider 추상화 — provider 파라미터로 커스텀 피드백 수집 전략을 지정합니다.

from crewai.flow import Flow, start, human_feedback, HumanFeedbackProvider, HumanFeedbackPending, PendingFeedbackContext

class WebhookProvider(HumanFeedbackProvider):
    """Provider that pauses flow and waits for webhook callback."""

    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url

    def request_feedback(self, context: PendingFeedbackContext, flow: Flow) -> str:
        # Notify external system (e.g., send Slack message, create ticket)
        self.send_notification(context)

        # Pause execution - framework handles persistence automatically
        raise HumanFeedbackPending(
            context=context,
            callback_info={"webhook_url": f"{self.webhook_url}/{context.flow_id}"}
        )

class ReviewFlow(Flow):
    @start()
    @human_feedback(
        message="Review this content:",
        emit=["approved", "rejected"],
        llm="gpt-4o-mini",
        provider=WebhookProvider("https://myapp.com/api"),
    )
    def generate_content(self):
        return "AI-generated content..."

    @listen("approved")
    def publish(self, result):
        return "Published!"

HumanFeedbackPending이 발생하면 Flow 프레임워크가 상태를 자동 영속화합니다. 프로바이더는 외부 시스템에 알리고 예외만 발생시키면 되고, 수동 영속 호출은 필요 없어요.

일시정지된 Flow 처리 — async 프로바이더를 쓰면 kickoff()는 예외를 던지는 대신 HumanFeedbackPending 객체를 반환합니다.

flow = ReviewFlow()
result = flow.kickoff()

if isinstance(result, HumanFeedbackPending):
    # Flow is paused, state is automatically persisted
    print(f"Waiting for feedback at: {result.callback_info['webhook_url']}")
    print(f"Flow ID: {result.context.flow_id}")
else:
    # Normal completion
    print(f"Flow completed: {result}")

일시정지된 Flow 재개 — 피드백이 도착하면 Flow를 재개합니다.

# Sync handler:
def handle_feedback_webhook(flow_id: str, feedback: str):
    flow = ReviewFlow.from_pending(flow_id)
    result = flow.resume(feedback)
    return result

# Async handler (FastAPI, aiohttp, etc.):
async def handle_feedback_webhook(flow_id: str, feedback: str):
    flow = ReviewFlow.from_pending(flow_id)
    result = await flow.resume_async(feedback)
    return result

주요 타입:

PendingFeedbackContext에는 재개에 필요한 모든 것이 들어 있습니다: flow_id, flow_class, method_name, method_output, message, emit, default_outcome, metadata, llm, requested_at.

더 알아보기

• CrewAI Flows — @human_feedback가 쓰이는 곳
• Event Listeners — Human In The Loop 이벤트
• Human Feedback in Flows 공식 문서