批量结果串项
LangChain 批量重试后失败项拿到邻项答案,检查最后一轮成功与失败混合。逐项保留结果归属,沿用重试配置,回查已保存记录。
三个输入经过 with_retry().batch():第一项重试后成功,第二项一直失败,第三项首次就成功。返回列表却把第一项的答案填进第二项,连 return_exceptions=False 也可能没有抛出错误。此时不能把“列表长度正确、调用没报错”当作验收。
这条条件有 langchain-core 1.6.9 的反馈,已核对的正式 1.6.10 中,同步 _batch 与异步 _abatch 仍保留相同的结果拼装路径。它与配置错了异常过滤、导致不该重试的错误反复执行,是不同问题。
错配出现在最后一轮结果拼装
每轮只重试尚未成功的输入,成功值按原始索引写进 results_map。最后一轮可能同时包含刚成功的 A 与仍失败的 B。
最终拼装时,已有成功记录的 A 从映射中取值;没有成功记录的 B 却从最后一轮完整结果列表 pop(0),取到的仍是 A。于是永久失败项看起来也成功了,而真正异常没有留在对应位置。
保留每项稳定标识、原始索引、尝试次数与实际结果归属,先判断是否符合“部分恢复、部分耗尽”的组合。只有所有项都成功,或最后一轮全部失败,可能看不出这条路径。
暂时绕开批量重试拼装
对依赖这一版本且需要可靠失败归属的路径,可以逐项调用已经配置好的 retryable Runnable,让每项走自己的 invoke 重试。下面只定义串行处理函数,不执行链:
def invoke_each(retryable, inputs, configs, **kwargs):
if len(inputs) != len(configs):
raise ValueError("one config required per input")
outputs = []
for index, (item, config) in enumerate(zip(inputs, configs, strict=True)):
try:
value = retryable.invoke(item, config=config, **kwargs)
except Exception as error:
value = error
outputs.append((index, value))
return outputs
这里选择了显式保留每项异常的契约,返回的是“原始索引、结果或异常”,不是原来纯结果列表的直接替代品。上层若要求任何失败立即传播,应改为相应的严格处理,不能把异常对象又转成成功字符串。
retryable 沿用原来的异常筛选、退避和尝试预算,configs 也应沿应用流程按项准备,保留 tracing、metadata 与其他约定。不能因为改成逐项处理,就把一套所有错误都重试的默认配置加到整条 Agent 流程上。
这会改变批量吞吐与调度方式。需要并发时,可在经过核验的应用调度层分别执行 ainvoke,按原索引收集并限制并发;不能把同一个 retryable 的 abatch 当成绕过方式,正式实现也有相同拼装问题。
对照失败项,而不是只看成功项
获准的无副作用验收应保留三种输入:首次成功、重试后成功、一直失败。逐项检查最后一种仍属于自己的异常,严格模式确实失败,再覆盖同步、异步与空输入。
升级后仍需对照实际版本的拼装实现和这组混合结果,而不是只运行全成功样例。
已经保存的批量结果也要按业务标识回查。不要为了确认归属重跑付款、写入或创建类操作;重试本身不证明远端没有完成副作用。无法确认的记录应保留为待核实,不能沿错配的成功值继续提交。


