[错误] 在背景线程中,放弃部分已消耗的同步流会默默消耗整个上游流

发生了什么事?

dspy.streamify (Program, sync streaming=False)' 返回应用 sync sreaming (sync streamer (.))' (`dspy/sreaming/sreamify.py:238') 。 同步生成器由背景守护进程线程提供,该线程将Aync流泵入无约束的"ueue":

def 生产商( ):
aync def 跑道( ) :
尝试 :
正在同步的项目的同步( G):
队列. put( 项目)
例外为例外:
队列. put ((例外  sentinel, exc))
终于:
队列. put( stop  sentinel)

上下文.run( asyncio.run, run) ()

线程=线程. 线索(目标=生产者,守护进程= True)

没有什么能阻止它 如果消费者离开部分消费的流-断'出循环,gen.close ()',或只是放弃参考,则生产者就一直将`同步-流子 ' 推到最后。 包裹的程序(及其LM调用)在背景中运行到完成:

• 完成部分仍在耗用**——所有剩余标志都从供应商流中取出并开具帐单,因为产出无人阅读;
• 每个块被缓冲入未被限制的队列,一直被保留到进程退出(守护进程线程及其队列每被丢弃的流漏出);
• `Async streamer.aclose ()'从不要求提前离开,因此,Anyio任务组因疲劳被撕毁,从不取消。

这是#10380的同步双(大约是‘aclose ()')在 *async * 生成器的起动上): Async 侧被响亮地撞倒了;同步侧无声地失败了,这可以说是更糟糕的——程序可观察的行为中没有任何东西说流还在运行.

###来再生.

离线, 无网络( LM 流的计数生成器 ) :

[Python] 导入时间 从 dspy. sreaming. sreamify 导入应用 sync sreaming

消耗={"数":0,"已": False]

上游( A) : i 范围(1 000): 消耗的[“计数”]=i+1 输出 f"chunk-{i}" 消耗 ["完成"] = 真实

gen = application sync 分流(上流()) 下一个( gen) # 消耗一个块

走开

时间. sleep( 0.5) 打印( 已消费) # {' 计数' : 1000, '完成' : true}

观察了目前的 " 主要 " (4368715b):在消耗了**一**块并关闭之后,上游已经消耗了**1000/1000**——在`接近()'返回时,生产者已经在整个溪流中奔跑,而`撕裂.假定()'仍然列出生产者的线程。 由于上游速度缓慢(网络速度),效果是一样的,只是分布在溪流的整个期间:`接近()'立即返回,而背景线则不断拉出块,直到提供者完成。

预期行为

关闭(或放弃)部分消耗的同步流会停止背景消耗:生产者任务被取消,Aync生成器被关闭,线程出站,再也没有拉出上游块. 完全消耗和误差传播(# 9142) 行为准确:
. . . . . . .