流程中的人工反馈

了解如何使用 @human_feedback 装饰器将人工反馈直接集成到你的 CrewAI 流程中

概述

NOTE

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

@human_feedback 装饰器可直接在 CrewAI 流程中启用人在回路（HITL）工作流。它允许你暂停流程执行，将输出呈现给人工审核，收集其反馈，并可选地根据反馈结果路由到不同的监听器。

这对于以下场景尤其有价值：

质量保证：在下游使用 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]

快速开始

以下是将人工反馈添加到流程的最简单方法：

python

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()

当此流程运行时，它将：

执行 generate_content 并返回字符串
使用请求消息向用户显示输出
等待用户输入反馈（或按 Enter 跳过）
将 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_source`	`str`	否	经验存储与召回所用的记忆 `source`。默认 `"hitl"`
`learn_strict`	`bool`	否	预审或经验提炼失败时抛出异常，而不是回退。默认 `False`

基本用法（无路由）

当你未指定 emit 时，该装饰器只收集反馈，并将 HumanFeedbackResult 传递给下一个监听器：

python

@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 解读，并被归并为指定的某个结果：

python

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}")

当人类说类似“需要更多细节”时，LLM 会将其归并为 "needs_revision"，这会通过 or_() 再次触发 review_content——形成一个修订循环。循环会一直持续到结果为 "approved" 或 "rejected"。

TIP

在可用时，LLM 会使用结构化输出（函数调用）来确保响应是你指定的结果之一。这让路由变得可靠且可预测。

WARNING

@start() 方法只在流程开始时运行一次。如果你需要修订循环，请将起始方法与审查方法分开，并在审查方法上使用 @listen(or_("trigger", "revision_outcome")) 来启用自循环。

HumanFeedbackResult

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

python

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：

python

@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：

python

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

human_feedback_history

流程期间收集的所有 HumanFeedbackResult 对象的列表：

python

@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，因此多个反馈步骤不会相互覆盖。使用该列表可访问流程期间收集的全部反馈。

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

下面是一个完整示例，实现了带修订循环的内容审查与审批工作流：

CodeOutput

python

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}")

text

==================================================
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"))——审查方法同时监听初始触发和自身的修订结果，形成一个自循环，重复执行直到人类批准或拒绝。

与其他装饰器组合

@human_feedback 装饰器可与 @start()、@listen() 和 or_() 配合使用。两种装饰器顺序都可行——框架会双向传播属性——但推荐的做法是：

python

# One-shot review at the start of a flow (no self-loop)
@start()
@human_feedback(message="Review this:", emit=["approved", "rejected"], llm="gpt-4o-mini")
def my_start_method(self):
    return "content"

# Linear review on a listener (no self-loop)
@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}"

# Self-loop: review that can loop back for revisions
@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"

自循环模式

要创建修订循环，审查方法必须使用 or_() 同时监听上游触发和自身的修订结果：

python

@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" 时，流程会回到 review（因为它通过 or_() 监听 "revise"）。当结果为 "approved" 时，流程继续前往 publish。这之所以可行，是因为流程引擎将路由豁免于“只触发一次”规则，允许它们在每次循环迭代中重新执行。

链式路由

由某个路由的结果触发的监听器本身也可以是路由：

python

@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 参数是呈现给人类的内容。让它具备可操作性：

python

# ✅ 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 时，选择能自然对应人类响应的结果：

python

# ✅ 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 而未输入内容的情况：

python

@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 来创建审计日志：

python

@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. 同时处理有路由与无路由的反馈

在设计流程时，考虑是否需要路由：

场景	使用
简单审查，只需要反馈文本	不使用 `emit`
需要根据响应分支到不同路径	使用 `emit`
带批准/拒绝/修订的审批关卡	使用 `emit`
仅收集评论用于日志记录	不使用 `emit`

异步人工反馈（非阻塞）

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

provider 抽象

使用 provider 参数指定自定义的反馈收集策略：

python

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

当抛出 HumanFeedbackPending 时，流程框架会自动持久化状态。你的 provider 只需通知外部系统并抛出该异常——无需手动调用持久化。

处理已暂停的流程

使用异步 provider 时，kickoff() 会返回 HumanFeedbackPending 对象，而不是抛出异常：

python

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}")

恢复已暂停的流程

当反馈到达时（例如通过 webhook），恢复流程：

python

# 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 的 Protocol
`PendingFeedbackContext`	包含恢复暂停流程所需的全部信息
`HumanFeedbackPending`	流程因等待反馈而暂停时由 `kickoff()` 返回
`ConsoleProvider`	默认的阻塞式控制台输入 provider

PendingFeedbackContext

该上下文包含恢复所需的全部内容：

python

@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

完整异步流程示例

python

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 异步模式），请使用 await flow.resume_async() 而非 flow.resume()。在正在运行的事件循环中调用 resume() 会抛出 RuntimeError。

异步反馈的最佳实践

检查返回类型：暂停时 kickoff() 返回 HumanFeedbackPending——无需 try/except
使用正确的恢复方法：在同步代码中使用 resume()，在异步代码中使用 await resume_async()
存储回调信息：使用 callback_info 存储 webhook URL、工单 ID 等
实现幂等性：为安全起见，你的恢复处理程序应当是幂等的
自动持久化：抛出 HumanFeedbackPending 时会自动保存状态，默认使用 SQLiteFlowPersistence
自定义持久化：如有需要，向 from_pending() 传入自定义持久化实例

从反馈中学习

learn=True 参数可在人工审查者与记忆系统之间开启反馈循环。启用后，系统会通过从过往的人工纠正中学习，逐步改进其输出。

工作原理

反馈之后：LLM 从输出 + 反馈中提炼可泛化的经验，并将其存储在 learn_source（默认 "hitl"）下的记忆中。无论是运行期间收到回复，还是稍后通过异步 provider恢复暂停的流程，都会执行此操作。如果反馈只是批准（例如“看起来不错”），则不会存储任何内容。
下次审查之前：从记忆中召回过去的 HITL 经验，由 LLM 在人类看到输出之前应用这些经验来改进输出。文本输出会被重写为文本；Pydantic 模型会被重写为同一模型，除非它包含 LLM 看不到的内容：密钥、被排除的字段，或被改离默认值的私有属性。其他输出（例如 dict 或 list）则原样展示。

随着时间推移，人类会看到越来越好的预审输出，因为每次纠正都会为未来的审查提供信息。

NOTE

人类批准的是经过审查的版本，它可能与方法返回的内容不同。在下游，请从 self.last_human_feedback.output（或监听器的参数）读取它。方法在审查之前设置的状态仍然保留原始内容。如果流程因异步反馈而暂停，恢复时输出会从 JSON 还原，因此 Pydantic 模型会以 dict 形式到达。

Example

python

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}")

首次运行：人类看到原始输出并说“事实性陈述始终要包含引用”。该经验被提炼并存入记忆。

第二次运行：系统召回关于引用的经验，在预审中修改输出以添加引用，然后展示改进后的版本。人类的工作从“修复所有问题”转变为“找出系统遗漏的内容”。

配置

参数	默认值	描述
`learn`	`False`	启用 HITL 学习
`learn_source`	`"hitl"`	经验教训所存储与召回的来源
`learn_strict`	`False`	在预审或蒸馏失败时抛出异常，而不是回退

关键设计决策

所有环节共用同一个 LLM：装饰器上的 llm 参数由结果归并、经验蒸馏和预审共享。无需配置多个模型。
结构化输出：当 LLM 支持时，蒸馏和预审都使用带 Pydantic 模型的函数调用，否则回退到文本解析。
非阻塞存储：经验通过 remember_many() 存储，该方法在后台线程中运行——流程会立即继续。
优雅降级：如果 LLM 在蒸馏过程中失败，则不会存储任何内容。如果它在预审过程中失败，则显示原始输出。除非 learn_strict=True（此时会抛出异常），否则这两种失败都不会阻塞流程。
无需 scope/categories：存储经验时只传入 source。编码管线会自动推断 scope、categories 和 importance。

NOTE

learn=True 要求 Flow 具备可用的记忆。Flow 默认会自动获得记忆，但如果你已通过 _skip_auto_memory 将其禁用，HITL 学习将被静默跳过。

流程中的人工反馈

概述

快速开始

@human_feedback 装饰器

参数

基本用法（无路由）

使用 emit 进行路由

HumanFeedbackResult

在监听器中访问

访问反馈历史

last_human_feedback

human_feedback_history

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

与其他装饰器组合

自循环模式

链式路由

限制

最佳实践

1. 编写清晰的请求消息

2. 选择有意义的结果

3. 始终提供默认结果

4. 使用反馈历史作为审计记录

5. 同时处理有路由与无路由的反馈

异步人工反馈（非阻塞）

provider 抽象

处理已暂停的流程

恢复已暂停的流程

关键类型

PendingFeedbackContext

完整异步流程示例

异步反馈的最佳实践

从反馈中学习

工作原理

Example

配置

关键设计决策

相关文档

友情链接