Flow에서 인간 피드백

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

class SimpleReviewFlow(Flow):
    @start()
    @human_feedback(message="이 콘텐츠를 검토해 주세요:")
    def generate_content(self):
        return "검토가 필요한 AI 생성 콘텐츠입니다."

    @listen(generate_content)
    def process_feedback(self, result):
        print(f"콘텐츠: {result.output}")
        print(f"인간의 의견: {result.feedback}")

flow = SimpleReviewFlow()
flow.kickoff()

@start()
@human_feedback(message="이 분석에 대해 어떻게 생각하시나요?")
def analyze_data(self):
    return "분석 결과: 매출 15% 증가, 비용 8% 감소"

@listen(analyze_data)
def handle_feedback(self, result):
    # result는 HumanFeedbackResult입니다
    print(f"분석: {result.output}")
    print(f"피드백: {result.feedback}")

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 "블로그 게시물 초안 내용..."

    @human_feedback(
        message="이 콘텐츠의 출판을 승인하시겠습니까?",
        emit=["approved", "rejected", "needs_revision"],
        llm="gpt-4o-mini",
        default_outcome="needs_revision",
    )
    @listen(or_("generate_content", "needs_revision"))
    def review_content(self):
        return "블로그 게시물 초안 내용..."

    @listen("approved")
    def publish(self, result):
        print(f"출판 중! 사용자 의견: {result.feedback}")

    @listen("rejected")
    def discard(self, result):
        print(f"폐기됨. 이유: {result.feedback}")

from crewai.flow.human_feedback import HumanFeedbackResult

@dataclass
class HumanFeedbackResult:
    output: Any              # 인간에게 표시된 원래 메서드 출력
    feedback: str            # 인간의 원시 피드백 텍스트
    outcome: str | None      # 매핑된 outcome (emit이 지정된 경우)
    timestamp: datetime      # 피드백이 수신된 시간
    method_name: str         # 데코레이터된 메서드의 이름
    metadata: dict           # 데코레이터에 전달된 모든 메타데이터

@listen("approved")
def on_approval(self, result: HumanFeedbackResult):
    print(f"원래 출력: {result.output}")
    print(f"사용자 피드백: {result.feedback}")
    print(f"Outcome: {result.outcome}")  # "approved"
    print(f"수신 시간: {result.timestamp}")

@listen(some_method)
def check_feedback(self):
    if self.last_human_feedback:
        print(f"마지막 피드백: {self.last_human_feedback.feedback}")

@listen(final_step)
def summarize(self):
    print(f"수집된 총 피드백: {len(self.human_feedback_history)}")
    for i, fb in enumerate(self.human_feedback_history):
        print(f"{i+1}. {fb.method_name}: {fb.outcome or '라우팅 없음'}")

# Flow 시작 시 일회성 검토 (self-loop 없음)
@start()
@human_feedback(message="이것을 검토해 주세요:", emit=["approved", "rejected"], llm="gpt-4o-mini")
def my_start_method(self):
    return "content"

# 리스너에서 선형 검토 (self-loop 없음)
@listen(other_method)
@human_feedback(message="이것도 검토해 주세요:", emit=["good", "bad"], llm="gpt-4o-mini")
def my_listener(self, data):
    return f"processed: {data}"

# Self-loop: 수정을 위해 반복할 수 있는 검토
@human_feedback(message="승인 또는 수정 요청?", emit=["approved", "revise"], llm="gpt-4o-mini")
@listen(or_("upstream_method", "revise"))
def review_with_loop(self):
    return "content for review"

@start()
def generate(self):
    return "initial draft"

@human_feedback(
    message="승인하시겠습니까, 아니면 변경을 요청하시겠습니까?",
    emit=["revise", "approved"],
    llm="gpt-4o-mini",
    default_outcome="approved",
)
@listen(or_("generate", "revise"))
def review(self):
    return "content"

@listen("approved")
def publish(self):
    return "published"

@start()
@human_feedback(message="첫 번째 검토:", emit=["approved", "rejected"], llm="gpt-4o-mini")
def draft(self):
    return "draft content"

@listen("approved")
@human_feedback(message="최종 검토:", emit=["publish", "revise"], llm="gpt-4o-mini")
def final_review(self, prev):
    return "final content"

@listen("publish")
def on_publish(self, prev):
    return "published"

# ✅ 좋음 - 명확하고 실행 가능
@human_feedback(message="이 요약이 핵심 포인트를 정확하게 캡처했나요? '예'로 답하거나 무엇이 빠졌는지 설명해 주세요:")

# ❌ 나쁨 - 모호함
@human_feedback(message="이것을 검토해 주세요:")

# ✅ 좋음 - 자연어 outcome
emit=["approved", "rejected", "needs_more_detail"]

# ❌ 나쁨 - 기술적이거나 불명확
emit=["state_1", "state_2", "state_3"]

@human_feedback(
    message="승인하시겠습니까? (수정 요청하려면 Enter 누르세요)",
    emit=["approved", "needs_revision"],
    llm="gpt-4o-mini",
    default_outcome="needs_revision",  # 안전한 기본값
)

@listen(final_step)
def create_audit_log(self):
    log = []
    for fb in self.human_feedback_history:
        log.append({
            "step": fb.method_name,
            "outcome": fb.outcome,
            "feedback": fb.feedback,
            "timestamp": fb.timestamp.isoformat(),
        })
    return log

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

class WebhookProvider(HumanFeedbackProvider):
    """웹훅 콜백을 기다리며 Flow를 일시 중지하는 Provider."""

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

    def request_feedback(self, context: PendingFeedbackContext, flow: Flow) -> str:
        # 외부 시스템에 알림 (예: Slack 메시지 전송, 티켓 생성)
        self.send_notification(context)

        # 실행 일시 중지 - 프레임워크가 자동으로 영속성 처리
        raise HumanFeedbackPending(
            context=context,
            callback_info={"webhook_url": f"{self.webhook_url}/{context.flow_id}"}
        )

class ReviewFlow(Flow):
    @start()
    @human_feedback(
        message="이 콘텐츠를 검토해 주세요:",
        emit=["approved", "rejected"],
        llm="gpt-4o-mini",
        provider=WebhookProvider("https://myapp.com/api"),
    )
    def generate_content(self):
        return "AI가 생성한 콘텐츠..."

    @listen("approved")
    def publish(self, result):
        return "출판됨!"

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

if isinstance(result, HumanFeedbackPending):
    # Flow가 일시 중지됨, 상태가 자동으로 영속화됨
    print(f"피드백 대기 중: {result.callback_info['webhook_url']}")
    print(f"Flow ID: {result.context.flow_id}")
else:
    # 정상 완료
    print(f"Flow 완료: {result}")

# 동기 핸들러:
def handle_feedback_webhook(flow_id: str, feedback: str):
    flow = ReviewFlow.from_pending(flow_id)
    result = flow.resume(feedback)
    return result

# 비동기 핸들러 (FastAPI, aiohttp 등):
async def handle_feedback_webhook(flow_id: str, feedback: str):
    flow = ReviewFlow.from_pending(flow_id)
    result = await flow.resume_async(feedback)
    return result

@dataclass
class PendingFeedbackContext:
    flow_id: str           # 이 Flow 실행의 고유 식별자
    flow_class: str        # 정규화된 클래스 이름
    method_name: str       # 피드백을 트리거한 메서드
    method_output: Any     # 인간에게 표시된 출력
    message: str           # 요청 메시지
    emit: list[str] | None # 라우팅을 위한 가능한 outcome
    default_outcome: str | None
    metadata: dict         # 커스텀 메타데이터
    llm: str | None        # outcome 매핑을 위한 LLM
    requested_at: datetime

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

class SlackNotificationProvider(HumanFeedbackProvider):
    """Slack 알림을 보내고 비동기 피드백을 위해 일시 중지하는 Provider."""

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

    def request_feedback(self, context: PendingFeedbackContext, flow: Flow) -> str:
        # Slack 알림 전송 (직접 구현)
        slack_thread_id = self.post_to_slack(
            channel=self.channel,
            message=f"검토 필요:\n\n{context.method_output}\n\n{context.message}",
        )

        # 실행 일시 중지 - 프레임워크가 자동으로 영속성 처리
        raise HumanFeedbackPending(
            context=context,
            callback_info={
                "slack_channel": self.channel,
                "thread_id": slack_thread_id,
            }
        )

class ContentPipeline(Flow):
    @start()
    @human_feedback(
        message="이 콘텐츠의 출판을 승인하시겠습니까?",
        emit=["approved", "rejected"],
        llm="gpt-4o-mini",
        default_outcome="rejected",
        provider=SlackNotificationProvider("#content-reviews"),
    )
    def generate_content(self):
        return "AI가 생성한 블로그 게시물 콘텐츠..."

    @listen("approved")
    def publish(self, result):
        print(f"출판 중! 검토자 의견: {result.feedback}")
        return {"status": "published"}

    @listen("rejected")
    def archive(self, result):
        print(f"보관됨. 이유: {result.feedback}")
        return {"status": "archived"}


# Flow 시작 (Slack 응답을 기다리며 일시 중지)
def start_content_pipeline():
    flow = ContentPipeline()
    result = flow.kickoff()

    if isinstance(result, HumanFeedbackPending):
        return {"status": "pending", "flow_id": result.context.flow_id}

    return result


# Slack 웹훅이 실행될 때 재개 (동기 핸들러)
def on_slack_feedback(flow_id: str, slack_message: str):
    flow = ContentPipeline.from_pending(flow_id)
    result = flow.resume(slack_message)
    return result


# 핸들러가 비동기인 경우 (FastAPI, aiohttp, Slack Bolt 비동기 등)
async def on_slack_feedback_async(flow_id: str, slack_message: str):
    flow = ContentPipeline.from_pending(flow_id)
    result = await flow.resume_async(slack_message)
    return result

class ArticleReviewFlow(Flow):
    @start()
    def generate_article(self):
        return self.crew.kickoff(inputs={"topic": "AI Safety"}).raw

    @human_feedback(
        message="이 글 초안을 검토해 주세요:",
        emit=["approved", "needs_revision"],
        llm="gpt-4o-mini",
        learn=True,
    )
    @listen(or_("generate_article", "needs_revision"))
    def review_article(self):
        return self.last_human_feedback.output if self.last_human_feedback else "article draft"

    @listen("approved")
    def publish(self):
        print(f"Publishing: {self.last_human_feedback.output}")