六块积木：把状态图逐个建起来

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

# State：这张图里所有节点都能读、都能写的一份共享数据
class State(TypedDict):
    topic: str
    draft: str

def write_draft(state: State) -> dict:
    return {"draft": f"关于 {state['topic']} 的初稿"}

def polish(state: State) -> dict:
    return {"draft": state["draft"] + "（已润色）"}

# 1) 建图模型，告诉它 state 的形状
builder = StateGraph(State)
# 2) 注册节点：名字 -> 函数
builder.add_node("write", write_draft)
builder.add_node("polish", polish)
# 3) 连边：START 是入口虚拟节点，END 是出口虚拟节点
builder.add_edge(START, "write")
builder.add_edge("write", "polish")
builder.add_edge("polish", END)
# 4) 编译成可执行对象（§1.5 细讲）
graph = builder.compile()

print(graph.invoke({"topic": "向量数据库", "draft": ""}))
# {'topic': '向量数据库', 'draft': '关于 向量数据库 的初稿（已润色）'}

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage

class State(TypedDict):
    # 字段 A：无 reducer -> 默认覆盖。返回 {"user_id": "u2"} 会抹掉旧值。
    user_id: str
    # 字段 B：reducer = operator.add -> 列表拼接。返回 {"events": ["x"]} 被追加。
    events: Annotated[list[str], add]
    # 字段 C：reducer = add_messages -> 按 message.id upsert（新 id 追加、同 id 重写）。
    messages: Annotated[list[AnyMessage], add_messages]

def node_alpha(state: State) -> dict:
    return {
        "user_id": "u_alpha",                                  # 覆盖
        "events": ["alpha"],                                   # 追加
        "messages": [AIMessage(content="from alpha", id="m1")],
    }

def node_beta(state: State) -> dict:
    return {
        "user_id": "u_beta",                                   # 又覆盖
        "events": ["beta"],                                    # 又追加
        "messages": [AIMessage(content="from beta", id="m2")], # 新 id -> 追加
    }

graph = (
    StateGraph(State)
    .add_node("alpha", node_alpha)
    .add_node("beta", node_beta)
    .add_edge(START, "alpha")
    .add_edge("alpha", "beta")
    .add_edge("beta", END)
    .compile()
)

initial = {"user_id": "u0", "events": [], "messages": [HumanMessage("hi", id="h1")]}
final = graph.invoke(initial)

print(final["user_id"])    # u_beta                  <- 默认 reducer 覆盖
print(final["events"])     # ['alpha', 'beta']       <- add 拼接
print([m.content for m in final["messages"]])
# ['hi', 'from alpha', 'from beta']                  <- add_messages 按 id 追加

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

class S(TypedDict):
    counter: int                       # 默认 reducer：覆盖
    events: Annotated[list[str], add]

# 形态 1：同步节点——读 state、返回部分更新 dict
def sync_step(state: S) -> dict:
    return {"counter": state["counter"] + 1, "events": ["sync"]}

# 形态 2：async 节点——契约相同，只是函数本身 async；适合 await LLM / HTTP
async def async_step(state: S) -> dict:
    # result = await some_llm_call(...)
    return {"events": ["async"]}

# 形态 3：返回 Command——同时"更新 state"和"声明下一步去哪"
def step_then_route(state: S) -> Command:
    return Command(
        update={"counter": state["counter"] * 2, "events": ["routed"]},
        goto="finalize",   # 直接指定下一节点，省掉一条 add_edge
    )

def finalize(state: S) -> dict:
    return {"events": ["done"]}

graph = (
    StateGraph(S)
    .add_node("sync_step", sync_step)
    .add_node("async_step", async_step)
    .add_node("step_then_route", step_then_route)
    .add_node("finalize", finalize)
    .add_edge(START, "sync_step")
    .add_edge("sync_step", "async_step")
    .add_edge("async_step", "step_then_route")
    # step_then_route 用 Command.goto 跳到 finalize——无需 add_edge
    .add_edge("finalize", END)
    .compile()
)

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

class State(TypedDict):
    question: str
    attempts: int
    answer: str
    log: Annotated[list[str], add]

def think(state: State) -> dict:
    n = state["attempts"] + 1
    # 假装第 2 次才想出满意答案
    answer = "satisfactory" if n >= 2 else "draft"
    return {"attempts": n, "answer": answer, "log": [f"think#{n}->{answer}"]}

def respond(state: State) -> dict:
    return {"log": ["responded"]}

# 条件函数：读 state，返回下一节点的名字（字符串）
def route_after_think(state: State) -> str:
    if state["answer"] == "satisfactory":
        return "respond"      # 满意 -> 去 respond
    if state["attempts"] >= 3:
        return "respond"      # 兜底：试够 3 次也收手，避免死循环
    return "think"            # 不满意且没超限 -> 回到 think（这就是回边）

graph = (
    StateGraph(State)
    .add_node("think", think)
    .add_node("respond", respond)
    # (1) 普通边：入口固定接 think
    .add_edge(START, "think")
    # (2) 条件边：think 跑完调 route_after_think，按返回值跳转
    #     第三参数是"返回值 -> 节点"映射，供可视化；不收紧运行时校验
    .add_conditional_edges("think", route_after_think,
                           {"think": "think", "respond": "respond"})
    # (3) 普通边：respond 跑完结束
    .add_edge("respond", END)
    .compile()
)

print(graph.invoke({"question": "Q", "attempts": 0, "answer": "", "log": []})["log"])
# ['think#1->draft', 'think#2->satisfactory', 'responded']
# think 跑了两次（第一次回边重来），第二次满意才走 respond

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

class S(TypedDict):
    x: int
    trace: Annotated[list[str], add]

def double(state: S) -> dict:
    return {"x": state["x"] * 2, "trace": ["double"]}

builder = StateGraph(S)
builder.add_node("double", double)
builder.add_edge(START, "double")
builder.add_edge("double", END)

graph = builder.compile()        # <- 冻结 + 拓扑校验 + 包成 Runnable

# 1) invoke：跑一次，拿最终 state
print(graph.invoke({"x": 3, "trace": []}))
# {'x': 6, 'trace': ['double']}

# 2) batch：一批输入并行跑——Runnable 接口自带，图没额外写
print(graph.batch([{"x": 1, "trace": []}, {"x": 10, "trace": []}]))
# [{'x': 2, 'trace': ['double']}, {'x': 20, 'trace': ['double']}]

# 3) stream：按步产出中间过程——同样来自 Runnable 接口
for chunk in graph.stream({"x": 5, "trace": []}):
    print(chunk)
# {'double': {'x': 10, 'trace': ['double']}}   <- 每个节点完成时吐一块

from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver   # v1.0 名字；旧名 MemorySaver

class S(TypedDict):
    count: int
    log: Annotated[list[str], add]

def step(state: S) -> dict:
    return {"count": state["count"] + 1, "log": [f"count={state['count']}"]}

graph = (
    StateGraph(S)
    .add_node("step", step)
    .add_edge(START, "step")
    .add_edge("step", END)
    .compile(checkpointer=InMemorySaver())     # <- 装上 checkpointer，图变有状态
)

cfg_alice = {"configurable": {"thread_id": "alice"}}
cfg_bob   = {"configurable": {"thread_id": "bob"}}

# Alice 连跑三次——同一 thread_id，state 在 checkpoint 上累积
graph.invoke({"count": 0, "log": []}, cfg_alice)   # count -> 1
graph.invoke({"count": 0, "log": []}, cfg_alice)   # 初始 count=0 被忽略！续到 2
graph.invoke({"count": 0, "log": []}, cfg_alice)   # 续到 3

print(graph.get_state(cfg_alice).values["count"])  # 3

# Bob 第一次跑——不同 thread_id，从零开始，与 Alice 互不可见
graph.invoke({"count": 0, "log": []}, cfg_bob)
print(graph.get_state(cfg_bob).values["count"])    # 1

# 时间旅行：倒序列出 Alice 的所有 checkpoint
for snap in graph.get_state_history(cfg_alice):
    print(snap.config["configurable"]["checkpoint_id"], snap.values["count"])