# 扳道员：读行李箱，返回"下一步去哪"的暗号（一个字符串）
def 审校结果如何(state):
    if state["passed"]:               # 审校通过
        return "通过"                  # 返回暗号"通过"
    if state["retries"] >= 3:         # 已重写3次，别死磕
        return "放弃"
    return "回炉"                      # 否则打回重写

# 注意：它没有 return {...} 改 state，只吐一个字符串指方向

from langgraph.graph import END
# 从"审校"站出来，让"审校结果如何"扳道员决定去哪
g.add_conditional_edges(
    "审校",                 # 从这个站出发
    审校结果如何,            # 扳道员函数：返回 "通过"/"回炉"/"放弃"
    {                       # 暗号 → 真实站名 的对照表
        "通过": END,         # 喊"通过" → 到终点
        "回炉": "写稿",       # 喊"回炉" → 绕回写稿站（这就形成了循环！）
        "放弃": END,         # 喊"放弃" → 也结束（兜底，别死循环）
    },
)

from langgraph.checkpoint.memory import MemorySaver   # 最简单的内存存档器

app = g.compile(
    checkpointer=MemorySaver(),        # 有存档才能"停了还能续"（Day37）
    interrupt_before=["发邮件"],        # 在"发邮件"站前落杆暂停
)
cfg = {"configurable": {"thread_id": "u1"}}   # 存档栏位编号，区分不同会话

app.invoke({"draft": "尊敬的客户..."}, cfg)    # 跑到"发邮件"前停住、已存档
# —— 这里把草稿拿给人看，人点了"同意" ——
app.invoke(None, cfg)                          # 传 None：从道口继续，真正发出

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

class State(TypedDict):
    draft: str                      # 当前稿子
    passed: bool                    # 审校过没
    retries: Annotated[int, add]    # 重写次数（追加型：每回炉一次+1）

def 写稿(state):
    n = state.get("retries", 0)
    return {"draft": f"这是第 {n+1} 版稿子，内容更充实了……" * (n+1)}

def 审校(state):
    ok = len(state["draft"]) >= 40           # 简单规则：够长就算过
    return {"passed": ok}

def 扳道员(state):                            # 决定：过了/回炉/放弃
    if state["passed"]:            return "通过"
    if state.get("retries", 0) >= 2: return "放弃"   # 最多再回炉，别死循环
    return "回炉"

g = StateGraph(State)
g.add_node("写稿", 写稿); g.add_node("审校", 审校)
g.add_edge(START, "写稿")
g.add_edge("写稿", "审校")
g.add_conditional_edges("审校", 扳道员,
    {"通过": END, "放弃": END, "回炉": "回炉计数"})   # 回炉先去计数站
g.add_node("回炉计数", lambda s: {"retries": 1})      # +1 次后回写稿
g.add_edge("回炉计数", "写稿")

app = g.compile()
out = app.invoke({"draft": "", "passed": False})
print("最终稿：", out["draft"][:30], "…")
print("重写次数：", out["retries"], "｜通过：", out["passed"])

✋ 动手 10 分钟：给写作 Agent 加一道"人工审批"

在 L06 的图上,把"通过"之后先别直接 END,而是加一个"发布"站,并在发布前落杆等人:

from langgraph.checkpoint.memory import MemorySaver
def 发布(state): return {"draft": state["draft"] + "（已发布）"}
g.add_node("发布", 发布)
# 把"通过"的去向从 END 改成 "发布"，再 发布→END
#   {"通过": "发布", ...} 、 g.add_edge("发布", END)

app = g.compile(checkpointer=MemorySaver(), interrupt_before=["发布"])  # 发布前落杆
cfg = {"configurable": {"thread_id": "t1"}}
app.invoke({"draft": "", "passed": False}, cfg)   # 跑到"发布"前停住
print("暂停，等人审批……当前稿：", app.get_state(cfg).values["draft"][:20])
app.invoke(None, cfg)                              # 人点头 → 继续，真正发布

观察:第一次 invoke 停在发布前;第二次 invoke(None) 才真发布。你就亲手做出了"人在环路"。