Haystack Pipeline.stream 放弃消费时怎样取消后台任务

Haystack 的 Pipeline.stream() 返回一个可异步迭代的 PipelineStreamHandle。如果消费方决定提前停止，最稳妥的做法是在 try/finally 中显式执行 await handle.aclose()：它会取消底层 Pipeline 任务，并等待有界的清理时间。不要把“停止读取几个 chunk”误认为“后台任务已经结束”。

官方地址：https://docs.haystack.deepset.ai/reference/pipeline-api

先判断：你手里的是流句柄还是最终结果

Pipeline.stream() 返回的 PipelineStreamHandle 实现了异步迭代协议。消费方通过 async for 逐个取得流式组件产出的 StreamingChunk；当迭代自然结束后，handle.result 才可能提供完整的 Pipeline 输出。

这意味着客户端停止渲染并不自动等于服务端工作已经停止。流句柄背后仍有一个异步任务：它可能正在等待组件、向下游传递输入或整理最终状态。需要主动放弃时，要把“停止消费”和“取消任务”作为两个动作处理。

提前停止时用 aclose 取消后台任务

最小处理方式是保存句柄，并在决定不再消费时调用 await handle.aclose()。例如只想展示前几个流块，或浏览器连接已经断开，都可以沿用下面的收尾结构：

async def consume_until_enough(pipeline, data, should_stop):
    # stream 返回句柄，底层 Pipeline 任务会在后台运行。
    handle = pipeline.stream(data)
    try:
        async for chunk in handle:
            # 这里只处理流块；是否停止由业务条件决定。
            render_chunk(chunk)
            if should_stop(chunk):
                break
    finally:
        # 无论 break 还是异常退出，都显式取消并等待清理。
        await handle.aclose()

    # 句柄被取消后不要读取 handle.result 作为完整答案。
    return None

aclose() 的职责是取消底层任务，并在有限时间内等待取消稳定下来。它不是“把已收到的 chunk 拼成最终结果”的接口，也不会把一次未完成的 Pipeline 变成成功运行。需要把已展示内容保存下来时，应该在业务层单独记录已经消费的 chunk。

为什么要把 aclose 放进 finally

提前 break、客户端断开、上游抛出异常，都可能让消费循环离开。虽然 Haystack 的异步迭代器在默认配置下会在放弃迭代时触发清理，但把调用写在业务自己的 finally 中更容易表达资源所有权，也不会把关键收尾寄托在对象何时被回收。

async def stream_to_client(pipeline, data, send):
    # 句柄的生命周期覆盖整个发送过程。
    handle = pipeline.stream(data)
    try:
        async for chunk in handle:
            # 发送失败会跳出循环，进入下面的清理分支。
            await send(chunk)
    except ConnectionError:
        # 客户端断开时不再等待剩余流内容，交给 finally 取消任务。
        raise
    finally:
        # 正常结束时 aclose 是幂等式的收尾调用；未结束时会取消任务。
        await handle.aclose()

正常读完整个流时，底层任务通常已经结束，aclose() 不需要再次取消；提前离开时，它才承担取消职责。生产代码还应把断开原因、已发送 chunk 数和取消耗时写入自己的日志，但不要把取消误记成模型或组件失败。

取消后 result 为什么不能读取

官方 API 将 result 定义为成功且完整运行后的最终 Pipeline 输出。如果任务尚未完成，读取它会得到“Pipeline 尚未完成”的运行时错误；如果任务已经被取消，则会得到“Pipeline 已取消、没有结果”的运行时错误。这个设计可以避免把半截流误当成完整输出。

排查清单

1. 先确认调用方是否只停止了读取，却没有保存句柄并调用 aclose()。
2. 确认 aclose() 位于 finally，这样发送失败和客户端断开也能进入清理分支。
3. 不要在取消后读取 handle.result，已发送的部分内容与完整 Pipeline 输出是两种数据。
4. 如果下游组件内部把同步工作放进线程，取消异步任务不一定能立刻中断已经运行的线程；日志中应区分任务取消和外部副作用是否已停止。
5. 只有确实读完整个流并成功结束，才把 result 写入完成状态；其它路径记录为客户端主动中止或传输异常。

常见问题

直接对 async for 使用 break，Haystack 会自动取消吗？

默认迭代器设计包含放弃迭代时的清理逻辑，但业务层仍建议在自己的 finally 中显式调用 aclose()。这样提前停止的意图和句柄生命周期都清楚，也不依赖对象回收时机。

aclose 会返回已经收到的内容吗？

不会。它负责取消底层 Pipeline 任务并等待清理；已经收到的 chunk 由调用方自己保存，取消后的句柄不能当成完整答案。

什么时候可以读取 handle.result？

只有异步迭代自然结束、底层任务成功完成时才读取。任务仍在运行或已经取消时，应该根据运行时错误处理，而不是继续重试读取。