第 16 章 容错：重试、超时与兜底

from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.pregel import RetryPolicy

class ScreenState(TypedDict):
    resume_text: str
    score: Annotated[int, operator.add]
    log: Annotated[list[str], operator.add]

# 模拟一个不稳定的模型调用：前两次失败，第三次成功
_attempts = {"count": 0}

def flaky_evaluate(state: ScreenState) -> dict:
    _attempts["count"] += 1
    if _attempts["count"] < 3:
        raise ConnectionError(f"模型服务暂时不可用（第 {_attempts['count']} 次）")
    return {"score": 75, "log": ["evaluate 成功（第 3 次尝试）"]}

def report(state: ScreenState) -> dict:
    return {"log": [f"最终得分 {state['score']}"]}

builder = StateGraph(ScreenState)
builder.add_node(
    "evaluate",
    flaky_evaluate,
    retry=RetryPolicy(initial_interval=1.0,    # 首次重试等 1 秒
                      backoff_factor=2.0,      # 指数退避：1s → 2s
                      max_attempts=3),         # 最多尝试 3 次
)
builder.add_node("report", report)
builder.add_edge(START, "evaluate")
builder.add_edge("evaluate", "report")
builder.add_edge("report", END)
graph = builder.compile()

result = graph.invoke({"resume_text": "李思远 ..."})
print(result["log"])
# ['evaluate 成功（第 3 次尝试）', '最终得分 75']

import asyncio
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o-mini",
    timeout=30,          # LangChain 模型自带的请求超时
    max_retries=2,       # SDK 层自动重试
)

async def evaluate(state) -> dict:
    # 节点内部再包一层总闸：整个节点 45 秒拿不下就放弃
    try:
        result = await asyncio.wait_for(
            llm.ainvoke(build_prompt(state)),
            timeout=45,
        )
        return {"score": parse_score(result)}
    except asyncio.TimeoutError:
        raise RuntimeError("evaluate 节点超时")

from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class ScreenState(TypedDict):
    resume_text: str
    bg_report: str
    score: Annotated[int, operator.add]
    log: Annotated[list[str], operator.add]

SERVICE_OK = False        # 改成 True 模拟下游恢复

def background_check(state: ScreenState) -> Command:
    try:
        if not SERVICE_OK:
            # 模拟不稳定的下游：真实场景这里是外部调用
            raise ConnectionError("背景核查服务不可用")
        return Command(
            update={"bg_report": "无风险记录", "score": 5, "log": ["background_check 成功"]},
            goto="evaluate",               # 成功：正常流转
        )
    except ConnectionError as e:
        # 失败信息写入状态，流向兜底节点
        return Command(
            update={"log": [f"background_check 失败：{e}"]},
            goto="bg_fallback",            # 异常分流：去兜底节点
        )

def bg_fallback(state: ScreenState) -> dict:
    # 降级策略：标记"未核查"，流程继续
    return {
        "bg_report": "背景核查暂不可用（已降级跳过）",
        "score": 0,                        # 不加分也不扣分
        "log": ["bg_fallback：降级处理，流程继续"],
    }

def evaluate(state: ScreenState) -> dict:
    return {"score": 80, "log": ["evaluate 完成"]}

builder = StateGraph(ScreenState)
builder.add_node("background_check", background_check)   # 返回 Command：不挂出边
builder.add_node("bg_fallback", bg_fallback)
builder.add_node("evaluate", evaluate)
builder.add_edge(START, "background_check")
builder.add_edge("bg_fallback", "evaluate")
builder.add_edge("evaluate", END)
graph = builder.compile()

result = graph.invoke({"resume_text": "李思远 ..."})
print(result["bg_report"])   # 背景核查暂不可用（已降级跳过）
print(result["score"])       # 80（背景分 0 + 评估 80）
print(result["log"])

# 失败发生后：
config = {"configurable": {"thread_id": "batch-042"}}
snap = graph.get_state(config)
print(snap.next)        # ('evaluate_17',) —— 失败点：第 17 份简历的评估节点

# 修好外部依赖后，从失败点重跑（输入 None = 用检查点原状态）
for chunk in graph.stream(None, config, stream_mode="updates"):
    print(chunk)
# 从 evaluate_17 继续，前 16 份的结果直接从检查点恢复，不重算