Flow 中的人类反馈

概览

Note

@human_feedback 装饰器要求 CrewAI 1.8.0 或更高版本。在使用此功能前，请确保已更新安装。

@human_feedback 装饰器可以让 human-in-the-loop（HITL）工作流直接运行在 CrewAI Flows 中。它允许你暂停 flow 执行，将输出展示给人类审核，收集反馈，并可根据反馈结果将流程路由到不同的监听器。

它尤其适用于：

• 质量保证：在 AI 生成内容进入下游之前先进行审核
• 决策关口：让人类在自动化工作流中做出关键决策
• 审批工作流：实现批准 / 拒绝 / 修改 模式
• 交互式迭代：收集反馈以逐步改进输出

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]

快速开始

下面是向 flow 添加人类反馈的最简单方式：

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 运行时，它会：

1. 执行 generate_content 并返回字符串
2. 将输出和请求消息一起展示给用户
3. 等待用户输入反馈（或直接按 Enter 跳过）
4. 将 HumanFeedbackResult 对象传递给 process_feedback

@human_feedback 装饰器

参数

参数	类型	必需	描述
message	str	是	与方法输出一起显示给人的消息
emit	Sequence[str]	否	可能结果的列表。反馈会被折叠成其中一个结果，并触发 @listen 装饰器
llm	str | BaseLLM	指定 emit 时	用于解释反馈并映射到结果的 LLM
default_outcome	str	否	在没有提供反馈时使用的结果。必须包含在 emit 中
metadata	dict	否	面向企业集成的附加数据
provider	HumanFeedbackProvider	否	用于异步/非阻塞反馈的自定义 provider。参见 异步人类反馈
learn	bool	否	启用 HITL 学习：从反馈中提炼经验，并在未来输出前进行预审。默认 False。参见 从反馈中学习
learn_limit	int	否	预审时最多回溯的历史经验条数。默认 5

基础用法（无路由）

当你不指定 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}")

当人类说出类似 “needs more detail” 的话时，LLM 会将其折叠为 "needs_revision"，然后通过 or_() 重新触发 review_content，从而形成一个修改循环。这个循环会一直持续，直到结果变为 "approved" 或 "rejected"。

Tip

在可用时，LLM 会使用结构化输出（函数调用）来保证响应一定属于你指定的结果之一。这使路由更可靠、更可预测。

Warning

@start() 方法只会在 flow 开始时运行一次。如果你需要修改循环，请把 start 方法和 review 方法分开，并在 review 方法上使用 @listen(or_("trigger", "revision_outcome")) 来启用自循环。

HumanFeedbackResult

HumanFeedbackResult 数据类包含一次人类反馈交互的全部信息：

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：

@listen(some_method)
def check_feedback(self):
    if self.last_human_feedback:
        print(f"Last feedback: {self.last_human_feedback.feedback}")

human_feedback_history

flow 期间收集到的所有 HumanFeedbackResult 对象列表：

@listen(final_step)
def summarize(self):
    print(f"Total feedback collected: {len(self.human_feedback_history)}")
    for i, fb in enumerate(self.human_feedback_history):
        print(f"{i+1}. {fb.method_name}: {fb.outcome or 'no routing'}")

Warning

每个 HumanFeedbackResult 都会追加到 human_feedback_history，因此多次反馈步骤不会互相覆盖。使用这个列表可以访问 flow 中收集到的所有反馈。

完整示例：内容审批工作流

下面是一个完整示例，展示带有修改循环的内容审核与审批工作流：

Code

from crewai.flow.flow import Flow, start, listen, or_
from crewai.flow.human_feedback import human_feedback, HumanFeedbackResult
from pydantic import BaseModel


class ContentState(BaseModel):
    draft: str = ""
    revision_count: int = 0
    status: str = "pending"


class ContentApprovalFlow(Flow[ContentState]):
    """A flow that generates content and loops until the human approves."""

    @start()
    def generate_draft(self):
        self.state.draft = "# AI Safety\n\nThis is a draft about AI Safety..."
        return self.state.draft

    @human_feedback(
        message="Please review this draft. Approve, reject, or describe what needs changing:",
        emit=["approved", "rejected", "needs_revision"],
        llm="gpt-4o-mini",
        default_outcome="needs_revision",
    )
    @listen(or_("generate_draft", "needs_revision"))
    def review_draft(self):
        self.state.revision_count += 1
        return f"{self.state.draft} (v{self.state.revision_count})"

    @listen("approved")
    def publish_content(self, result: HumanFeedbackResult):
        self.state.status = "published"
        print(f"Content approved and published! Reviewer said: {result.feedback}")
        return "published"

    @listen("rejected")
    def handle_rejection(self, result: HumanFeedbackResult):
        self.state.status = "rejected"
        print(f"Content rejected. Reason: {result.feedback}")
        return "rejected"


flow = ContentApprovalFlow()
result = flow.kickoff()
print(f"\nFlow completed. Status: {flow.state.status}, Reviews: {flow.state.revision_count}")

==================================================
OUTPUT FOR REVIEW:
==================================================
# AI Safety

This is a draft about AI Safety... (v1)
==================================================

Please review this draft. Approve, reject, or describe what needs changing:
(Press Enter to skip, or type your feedback)

Your feedback: Needs more detail on alignment research

==================================================
OUTPUT FOR REVIEW:
==================================================
# AI Safety

This is a draft about AI Safety... (v2)
==================================================

Please review this draft. Approve, reject, or describe what needs changing:
(Press Enter to skip, or type your feedback)

Your feedback: Looks good, approved!

Content approved and published! Reviewer said: Looks good, approved!

Flow completed. Status: published, Reviews: 2

关键模式是 @listen(or_("generate_draft", "needs_revision")) - review 方法同时监听初始触发和它自己的修改结果，从而形成一个自循环，直到人类批准或拒绝。

与其他装饰器结合使用

@human_feedback 装饰器可与 @start()、@listen() 和 or_() 配合使用。两种装饰器顺序都可以工作 - 框架会双向传播属性 - 但推荐模式如下：

# 在 flow 开头进行一次性审核（无自循环）
@start()
@human_feedback(message="Review this:", emit=["approved", "rejected"], llm="gpt-4o-mini")
def my_start_method(self):
    return "content"

# 在线性监听器上进行审核（无自循环）
@listen(other_method)
@human_feedback(message="Review this too:", emit=["good", "bad"], llm="gpt-4o-mini")
def my_listener(self, data):
    return f"processed: {data}"

# 自循环：可回到自身进行修改的审核
@human_feedback(message="Approve or revise?", emit=["approved", "revise"], llm="gpt-4o-mini")
@listen(or_("upstream_method", "revise"))
def review_with_loop(self):
    return "content for review"

自循环模式

要创建修改循环，review 方法必须通过 or_() 同时监听一个上游触发器和它自己的修改结果：

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

@human_feedback(
    message="Approve or request changes?",
    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"

当结果是 "revise" 时，flow 会回到 review（因为它通过 or_() 监听了 "revise"）。当结果是 "approved" 时，flow 会继续执行 publish。这能正常工作，是因为 flow 引擎会把路由器排除在“只触发一次”规则之外，从而允许它们在每次循环迭代中重新执行。

级联路由器

由某个路由器结果触发的监听器，也可以本身成为一个路由器：

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

@human_feedback(message="First review:", emit=["approved", "rejected"], llm="gpt-4o-mini")
@listen("generate")
def first_review(self):
    return "draft content"

@human_feedback(message="Final review:", emit=["publish", "hold"], llm="gpt-4o-mini")
@listen("approved")
def final_review(self, prev):
    return "final content"

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

@listen("hold")
def on_hold(self, prev):
    return "held for later"

限制

• @start() 方法只运行一次：@start() 方法不能自循环。如果你需要修改循环，请使用单独的 @start() 作为入口，并把 @human_feedback 放在 @listen() 方法上。
• 同一个方法不能同时使用 @start() 和 @listen()：这是 Flow 框架的限制。一个方法只能是起点或监听器，不能同时兼任两者。

最佳实践

1. 编写清晰的请求消息

message 参数是人类会看到的内容。让它具备可执行性：

# ✅ Good - clear and actionable
@human_feedback(message="Does this summary accurately capture the key points? Reply 'yes' or explain what's missing:")

# ❌ Bad - vague
@human_feedback(message="Review this:")

2. 选择有意义的结果

当使用 emit 时，选择能自然映射到人类反馈的结果：

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

# ❌ Bad - technical or unclear
emit=["state_1", "state_2", "state_3"]

3. 始终提供默认结果

使用 default_outcome 处理用户直接按 Enter 的情况：

@human_feedback(
    message="Approve? (press Enter to request revision)",
    emit=["approved", "needs_revision"],
    llm="gpt-4o-mini",
    default_outcome="needs_revision",  # Safe default
)

4. 使用反馈历史做审计追踪

访问 human_feedback_history 来创建审计日志：

@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

5. 同时处理带路由和不带路由的反馈

设计 flow 时，考虑是否需要路由：

场景	使用方式
只需要反馈文本的简单审核	不使用 emit
需要根据反馈分支到不同路径	使用 emit
带批准 / 拒绝 / 修改 的审批关卡	使用 emit
只收集评论用于记录	不使用 emit

异步人类反馈（非阻塞）

默认情况下，@human_feedback 会阻塞执行并等待控制台输入。对于生产应用，你可能需要异步 / 非阻塞的反馈，与 Slack、Email、Webhook 或 API 等外部系统集成。

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!"

Tip

flow 框架在抛出 HumanFeedbackPending 时会自动持久化状态。你的 provider 只需要通知外部系统并抛出这个异常，不需要手动调用持久化。

处理暂停的 Flow

当使用异步 provider 时，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

当收到反馈时（例如通过 webhook），恢复 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

关键类型

类型	描述
HumanFeedbackProvider	自定义反馈 provider 的协议
PendingFeedbackContext	恢复已暂停 flow 所需的全部信息
HumanFeedbackPending	flow 暂停等待反馈时由 kickoff() 返回的对象
ConsoleProvider	默认的阻塞式控制台输入 provider

PendingFeedbackContext

这个上下文包含恢复所需的一切：

@dataclass
class PendingFeedbackContext:
    flow_id: str           # Unique identifier for this flow execution
    flow_class: str        # Fully qualified class name
    method_name: str       # Method that triggered feedback
    method_output: Any     # Output shown to the human
    message: str           # The request message
    emit: list[str] | None # Possible outcomes for routing
    default_outcome: str | None
    metadata: dict         # Custom metadata
    llm: str | None        # LLM for outcome collapsing
    requested_at: datetime

完整异步 Flow 示例

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

class SlackNotificationProvider(HumanFeedbackProvider):
    """Provider that sends Slack notifications and pauses for async feedback."""

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

    def request_feedback(self, context: PendingFeedbackContext, flow: Flow) -> str:
        # Send Slack notification (implement your own)
        slack_thread_id = self.post_to_slack(
            channel=self.channel,
            message=f"Review needed:\n\n{context.method_output}\n\n{context.message}",
        )

        # Pause execution - framework handles persistence automatically
        raise HumanFeedbackPending(
            context=context,
            callback_info={
                "slack_channel": self.channel,
                "thread_id": slack_thread_id,
            }
        )

class ContentPipeline(Flow):
    @start()
    @human_feedback(
        message="Approve this content for publication?",
        emit=["approved", "rejected"],
        llm="gpt-4o-mini",
        default_outcome="rejected",
        provider=SlackNotificationProvider("#content-reviews"),
    )
    def generate_content(self):
        return "AI-generated blog post content..."

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

    @listen("rejected")
    def archive(self, result):
        print(f"Archived. Reason: {result.feedback}")
        return {"status": "archived"}


# Starting the flow (will pause and wait for Slack response)
def start_content_pipeline():
    flow = ContentPipeline()
    result = flow.kickoff()

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

    return result


# Resuming when Slack webhook fires (sync handler)
def on_slack_feedback(flow_id: str, slack_message: str):
    flow = ContentPipeline.from_pending(flow_id)
    result = flow.resume(slack_message)
    return result


# If your handler is async (FastAPI, aiohttp, Slack Bolt async, etc.)
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

Warning

如果你使用的是异步 Web 框架（FastAPI、aiohttp、Slack Bolt async 模式），请使用 await flow.resume_async()，而不是 flow.resume()。在运行中的事件循环里调用 resume() 会抛出 RuntimeError。

异步反馈最佳实践

1. 检查返回类型：kickoff() 在 flow 暂停时会返回 HumanFeedbackPending，不需要 try/except
2. 使用正确的恢复方法：同步代码里用 resume()，异步代码里用 await resume_async()
3. 保存回调信息：使用 callback_info 保存 webhook URL、ticket ID 等
4. 实现幂等性：你的恢复处理器应该具备幂等性以保证安全
5. 自动持久化：在抛出 HumanFeedbackPending 时状态会自动保存，默认使用 SQLiteFlowPersistence
6. 自定义持久化：如有需要，可将自定义持久化实例传给 from_pending()

从反馈中学习

learn=True 参数会在人类审核者和记忆系统之间建立反馈闭环。启用后，系统会通过学习过去的人类修正不断改进输出。

工作原理

1. 反馈之后：LLM 会从输出 + 反馈中提取可泛化的经验，并以 source="hitl" 的形式存入记忆。如果反馈只是批准（例如 “looks good”），则不会存储任何内容。
2. 下一次审核之前：会从记忆中回忆过去的 HITL 经验，并由 LLM 在人类看到之前应用这些经验来改进输出。

随着时间推移，人类会看到越来越好的预审输出，因为每次修正都会影响未来的审核。

示例

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

    @human_feedback(
        message="Review this article draft:",
        emit=["approved", "needs_revision"],
        llm="gpt-4o-mini",
        learn=True,  # enable HITL learning
    )
    @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}")

第一次运行：人类看到原始输出并说 “Always include citations for factual claims.” 这条经验会被提炼并存入记忆。

第二次运行：系统会回忆起引文经验，先对输出进行预审并补充引用，然后展示改进后的版本。人类的工作从“修复一切”变成“捕捉系统遗漏的部分”。

配置

参数	默认值	描述
learn	False	启用 HITL 学习
learn_limit	5	预审时最多回忆的历史经验数

关键设计决策

• 所有事情共用同一个 LLM：装饰器上的 llm 参数同时用于结果折叠、经验提炼和预审，无需配置多个模型。
• 结构化输出：当 LLM 支持时，经验提炼和预审都会使用带 Pydantic 模型的函数调用，否则回退到文本解析。
• 非阻塞存储：经验通过 remember_many() 存储，并在后台线程中运行 - flow 会立即继续。
• 优雅降级：如果 LLM 在提炼阶段失败，不会存储任何内容；如果在预审阶段失败，会直接展示原始输出。两种失败都不会阻塞 flow。
• 无需作用域/类别：存储经验时只传入 source。编码管道会自动推断作用域、类别和重要性。

Note

learn=True 需要 Flow 可用记忆。Flow 默认会自动获得记忆，但如果你通过 _skip_auto_memory 禁用了它，HITL 学习会被静默跳过。

相关文档

• Flows 概览 - 了解 CrewAI Flows
• Flow 状态管理 - 管理 flow 中的状态
• Flow 持久化 - 持久化 flow 状态
• 使用 @router 进行路由 - 更多关于条件路由的内容
• 执行时的人类输入 - 任务级人类输入
• 记忆 - HITL 学习使用的统一记忆系统