故障注入、追踪与发布边界：练习答案

参考实现

完整实现：

from langchain_core.runnables import RunnableLambda
counts={"timeout":0,"invalid":0}
def timeout(x):
    counts["timeout"]+=1
    raise TimeoutError("offline")
def invalid(x):
    counts["invalid"]+=1
    raise ValueError("bad input")
for name,fn,typ in [("timeout",timeout,TimeoutError),("invalid",invalid,ValueError)]:
    r=RunnableLambda(fn).with_retry(retry_if_exception_type=(TimeoutError,),stop_after_attempt=2,wait_exponential_jitter=False)
    try:r.invoke(1)
    except typ:pass
    else:raise AssertionError("error swallowed")
assert counts == {"timeout":2,"invalid":1}
print(counts)

限定异常类型使业务非法输入立即失败，避免重复负载。

返回正文