用 LangGraph 搭有状态、可中断恢复的生产级 Agent 工作流（检查点 + 人在回路）

pip install langgraph langchain-openai

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    topic: str
    draft: str
    approved: bool

def plan(state: State):
    return {"draft": f"关于 {state['topic']} 的初稿..."}

g = StateGraph(State)
g.add_node("plan", plan)
g.add_edge(START, "plan")
g.add_edge("plan", END)
app = g.compile()

不加检查点时，图跑完即焚；加上 `MemorySaver`，每次状态变更都落盘，崩了能恢复：

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = g.compile(checkpointer=memory)

# 用 thread_id 标识一条「会话」，进度按它存
config = {"configurable": {"thread_id": "task-001"}}
app.invoke({"topic": "AI 客服"}, config)

用 `interrupt_before` 在指定节点前暂停，等人确认后再继续——典型的「人在回路」：

def send_email(state: State):
    # 真正发邮件的逻辑（敏感操作）
    return {"approved": True}

g.add_node("send_email", send_email)
g.add_edge("plan", "send_email")
g.add_edge("send_email", END)

# 在 send_email 前中断，等人审批
app = g.compile(checkpointer=memory, interrupt_before=["send_email"])

app.invoke({"topic": "续费提醒"}, config)
# → 图在 send_email 前停下，等你批准
app.invoke(None, config)   # 你点头，继续往下跑

def route(state: State):
    if state.get("risk") == "high":
        return "human_review"   # 高风险走人工
    return "auto_fix"           # 低风险自动修

g.add_conditional_edges("detect", route)
# detect 节点后，按 route 返回值走不同分支

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-5.6")

def draft_node(state: State):
    msg = llm.invoke(f"写一段关于 {state['topic']} 的简介")
    return {"draft": msg.content}

g.add_node("draft_node", draft_node)
g.add_edge(START, "draft_node")

def fragile_call(state: State):
    # 可能超时的外部调用
    return {"result": call_api()}

# 用 try / 检查点配合：失败时回到上一检查点
# LangGraph 的「更新状态」API 还能人工修正后继续
app.update_state(config, {"draft": "修正后的内容"})

# 导出流程图，确认结构符合预期
app.get_graph().draw_mermaid_png("flow.png")

# 部署：编译后的 app 就是普通 Python 对象
# 套一层 FastAPI，用 thread_id 区分用户会话即可
from fastapi import FastAPI
api = FastAPI()
@api.post("/run")
def run(topic: str):
    return app.invoke({"topic": topic}, {"configurable": {"thread_id": topic}})