Langchain RunnableEach调用JSON错误处理及重试范围咨询

一、避免单个请求失败导致整体流程终止的处理方案 #

1. 为单个请求添加错误捕获与 fallback #

用RunnableLambda包装处理逻辑，捕获单个请求的解析错误，返回标记错误的默认结果，确保批量流程不中断：

from langchain_core.runnables import RunnableLambda

def handle_single_request(input_data):
    try:
        return tagging_chain.invoke(input_data)
    except Exception as e:
        # 返回自定义错误标记，可根据需求调整格式
        return {"flag": None, "answer": f"处理失败：{str(e)}"}

# 替换原RunnableEach绑定逻辑
runnable_each = RunnableEach(bound=RunnableLambda(handle_single_request))

2. 重试+fallback双重保障 #

先给LLM添加重试逻辑处理临时格式错误，再用fallback兜底重试仍失败的请求：

from langchain_core.runnables import RunnableWithFallbacks
from json.decoder import JSONDecodeError

# 针对JSON解析、值验证等异常重试，最多3次
retry_llm = llm.with_retry(
    stop_after_attempt=3,
    retry_on_exception=lambda e: isinstance(e, (JSONDecodeError, ValueError))
)

tagging_chain = prompt | retry_llm

# 定义fallback处理逻辑
def fallback_handler(input_data):
    return {"flag": None, "answer": "重试后仍处理失败"}

# 为chain添加fallback
chain_with_fallback = RunnableWithFallbacks(runnable=tagging_chain, fallbacks=[RunnableLambda(fallback_handler)])

runnable_each = RunnableEach(bound=chain_with_fallback)

3. 优化prompt提升输出稳定性 #

在prompt中明确要求严格的JSON格式，减少LLM输出错误：

请严格按照以下JSON格式返回结果，确保字符串闭合、无语法错误，禁止返回任何额外内容：
{"flag": true/false, "answer": "你的判断理由"}

二、RunnableRetry的作用范围说明 #

你给出的配置：

tagging_chain = prompt | llm.with_retry()
runnable_each = RunnableEach(bound=tagging_chain)

只会重试单个失败的请求，不会重试整个输入序列。

因为RunnableEach会将输入列表的每个元素单独传递给绑定的tagging_chain，每个请求的执行完全独立。当某个请求触发解析错误时，llm.with_retry()仅针对该单个请求重试，其他请求不受影响，整个批量流程不会中断。

内容的提问来源于stack exchange，提问作者A. Colli