LangGraph 生产级 Agent 搭建 SOP：超时、重试与容错

from typing import Annotated
from typing_extensions import TypedDict
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # reducer: 追加而非覆盖
    query: str
    research: str
    draft: str
    approved: bool

from langgraph.graph import StateGraph, START, END

builder = StateGraph(AgentState)
# 节点稍后添加
builder.add_edge(START, "research")
builder.add_edge("research", "draft")
builder.add_edge("draft", "review")

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

import uuid

config = {"configurable": {"thread_id": str(uuid.uuid4())}}
result = graph.invoke({"query": "对比 LangGraph 和 CrewAI", "messages": []}, config)

async def draft_node(state: AgentState) -> AgentState:
    # 调 LLM 生成回复
    ...

builder.add_node("draft", draft_node, timeout=30)  # 30 秒硬限

from langgraph.types import TimeoutPolicy

builder.add_node("research", research_node, timeout=TimeoutPolicy(
    run_timeout=60,    # wall-clock 硬限：整个节点最多跑 60 秒
    idle_timeout=15,   # idle 超时：15 秒无进度推进就超时，进度信号会重置计时
    refresh_on="auto",  # 自动监听回调事件刷新 idle 计时（默认值）
))

from langgraph.types import RetryPolicy

builder.add_node("research", research_node,
    timeout=TimeoutPolicy(run_timeout=60, idle_timeout=15),
    retry_policy=RetryPolicy(
        max_attempts=3,           # 最多尝试 3 次（含首次）
        initial_interval=1.0,     # 首次重试前等 1 秒
        backoff_factor=2.0,       # 每次重试间隔翻倍：1s → 2s → 4s
        max_interval=128.0,      # 重试间隔上限 128 秒
        jitter=True,              # 加随机抖动，避免重试风暴
    ),
)

from langgraph.errors import NodeTimeoutError

def retry_on_timeout_or_api_error(exc: Exception) -> bool:
    """超时和 API 错误才重试，其他直接崩。"""
    if isinstance(exc, NodeTimeoutError):
        return True
    return False

builder.add_node("research", research_node,
    retry_policy=RetryPolicy(max_attempts=3, retry_on=retry_on_timeout_or_api_error),
)

async def research_fallback(state: AgentState) -> AgentState:
    """research 节点连续失败后的兜底逻辑。"""
    return {
        "research": "外部数据源不可用，使用缓存摘要。",
        "messages": [{"role": "system", "content": "research 降级为缓存模式"}],
    }

builder.add_node("research", research_node,
    timeout=TimeoutPolicy(run_timeout=60, idle_timeout=15),
    retry_policy=RetryPolicy(max_attempts=3),
    error_handler=research_fallback,  # 3 次重试全失败 → 走这里
)

def should_approve(state: AgentState) -> str:
    """审核节点：通过则结束，不通过则回 draft 重写。"""
    return END if state["approved"] else "draft"

builder.add_conditional_edges("review", should_approve)

from langgraph.types import interrupt, Command

def review_node(state: AgentState) -> AgentState:
    # 暂停图执行，把草稿发给人工审核
    answer = interrupt({
        "type": "approval_request",
        "draft": state["draft"],
        "message": "请审核以上回复，回复 yes 批准、no 打回重写。",
    })
    # interrupt 在首次调用时抛 GraphInterrupt，暂停执行
    # 人工用 Command(resume=...) 恢复后，answer 接收人工输入
    return {"approved": answer == "yes"}

# 第一次 invoke 触发 interrupt，图暂停
for chunk in graph.stream({"query": "...", "messages": []}, config):
    print(chunk)
# 输出: {'__interrupt__': (...)}

# 人工审核后恢复
for chunk in graph.stream(Command(resume="yes"), config):
    print(chunk)
# approved=True → 图继续执行到 END

import uuid
import operator
from typing import Annotated
from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command, RetryPolicy, TimeoutPolicy
from langgraph.errors import NodeTimeoutError

# ---- 1. Typed State ----
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    query: str
    research: str
    draft: str
    approved: bool

# ---- 2. Nodes（全部 async，timeout 才生效）----
async def research_node(state: AgentState) -> AgentState:
    """调用搜索 API 获取信息。"""
    # 实际替换为你的搜索工具调用
    result = f"关于「{state['query']}」的搜索结果..."
    return {"research": result, "messages": [{"role": "tool", "content": result}]}

async def draft_node(state: AgentState) -> AgentState:
    """调用 LLM 生成回复草稿。"""
    # 实际替换为你的 LLM 调用
    draft = f"基于「{state['research']}」生成的回复..."
    return {"draft": draft, "messages": [{"role": "assistant", "content": draft}]}

def review_node(state: AgentState) -> AgentState:
    """人工审核断点：暂停等待审批。"""
    answer = interrupt({"draft": state["draft"], "msg": "批准 yes / 打回 no"})
    return {"approved": answer == "yes"}

# ---- 3. Error Handler（fallback 节点）----
async def research_fallback(state: AgentState) -> AgentState:
    return {"research": "数据源不可用，使用缓存。"}

# ---- 4. 自定义重试判断 ----
def retry_on_timeout(exc: Exception) -> bool:
    return isinstance(exc, NodeTimeoutError)

# ---- 5. 构建图 ----
builder = StateGraph(AgentState)

builder.add_node("research", research_node,
    timeout=TimeoutPolicy(run_timeout=60, idle_timeout=15),
    retry_policy=RetryPolicy(max_attempts=3, retry_on=retry_on_timeout),
    error_handler=research_fallback,
)
builder.add_node("draft", draft_node, timeout=30)
builder.add_node("review", review_node)

builder.add_edge(START, "research")
builder.add_edge("research", "draft")
builder.add_edge("draft", "review")

def should_approve(state: AgentState) -> str:
    return END if state["approved"] else "draft"

builder.add_conditional_edges("review", should_approve)

# ---- 6. 编译（挂 checkpointer）----
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# ---- 7. 运行 ----
config = {"configurable": {"thread_id": str(uuid.uuid4())}}

# 第一次调用：跑到 review 节点触发 interrupt，暂停
for chunk in graph.stream(
    {"query": "LangGraph 生产级最佳实践", "messages": []},
    config,
):
    print(chunk)

# 人工审批后恢复
for chunk in graph.stream(Command(resume="yes"), config):
    print(chunk)