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

# 1️⃣ 创建状态：定义在节点之间流转的数据结构
class MyState(TypedDict):
    question: str
    answer: str

# 2️⃣ 创建节点：每个节点接收 state，返回更新后的 state
def think_node(state: MyState):
    state["answer"] = "思考结果"
    return state

# 3️⃣ 把节点串起来：add_edge 定义执行顺序
def build_graph():
    gb = StateGraph(MyState)
    gb.add_node("think", think_node)          # 注册节点
    gb.add_edge(START, "think")               # 起点 → think
    gb.add_edge("think", END)                 # think → 终点
    return gb.compile()                       # 编译成可运行图

from langgraph.types import RunnableConfig

def step1(state: State, config: RunnableConfig):   # 👈 第二个参数
    prefix = config.get("configurable", {}).get("prefix", "")
    uname  = config.get("configurable", {}).get("uname", "")
    state["step1"] = f"{prefix}{state['input']}-{uname}"
    return state

# 运行时注入
config = RunnableConfig(configurable={
    "prefix": "[前缀] ",
    "uname":  "wangliangwu",
    "renshe": "好人卡"
})
graph.invoke({"input": "你好"}, config=config)

config = RunnableConfig(configurable={
    "thread_id": "12345"      # 👈 线程唯一标识
})
graph.invoke({"input": "你好"}, config=config)

from langgraph.checkpoint.memory import InMemorySaver

def build_graph():
    gb = StateGraph(State)
    gb.add_node("add", add_node)
    gb.add_edge(START, "add")
    gb.add_edge("add", END)
    memory = InMemorySaver()              # 👈 创建内存记忆器
    return gb.compile(memory)             # 👈 编译时挂载
    # 等价写法：gb.compile(checkpointer=memory)

graph = build_graph()

# 同一个 thread_id 多次调用 → state 会累加保留
config = RunnableConfig(configurable={"thread_id": "12345"})
graph.invoke({"new_value": 5},  config=config)   # total = 5
graph.invoke({"new_value": 7},  config=config)   # total = 5+7 = 12

# 换一个 thread_id → 全新开始
config2 = RunnableConfig(configurable={"thread_id": "99999"})
graph.invoke({"new_value": 10}, config=config2)  # total = 10

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

def human_approval(state: State):
    # 👇 触发中断：图会暂停在这里，等待外部 resume
    approval = interrupt("请输入审批结果（批准 / 不批准）：")
    state["approval"] = approval
    return state

# 必须挂载 memory，否则中断后无法恢复
memory = InMemorySaver()
graph = gb.compile(checkpointer=memory)

config = RunnableConfig(configurable={"thread_id": "12345"})

# === 第一次调用：执行到 interrupt 处自动暂停 ===
state = graph.invoke({}, config=config)
# 此时 state 包含中断信息，图暂停在 human_approval 节点

# === 第二次调用：传入人工输入，恢复执行 ===
user_decision = input("是否批准？")          # 模拟人工输入
state = graph.invoke(Command(resume=user_decision), config=config)
# 👆 用 Command(resume=...) 把人工结果回灌进图，继续往后跑

invoke → ... → interrupt() ⛔ 暂停
                        ↓ 状态存入 memory
                        ↓ 等待人工
invoke(Command(resume=...)) → 继续 → END

def build_graph():
    gb = StateGraph(MyState)
    gb.add_node("start", start_cooking)
    gb.add_node("make_vegetables", make_vegetables)   # 炒菜（10秒）
    gb.add_node("heat_rice", heat_rice)               # 热饭（2秒）
    gb.add_node("serve", serve_dish)

    # 👇 fan-out：start 之后同时进入两个节点（并行）
    gb.add_edge("start", "make_vegetables")
    gb.add_edge("start", "heat_rice")

    # 👇 fan-in：两个节点都完成后才进入 serve
    gb.add_edge(["make_vegetables", "heat_rice"], "serve")

    gb.add_edge("serve", END)
    return gb.compile()

        ┌→ make_vegetables ┐
start ──┤                   ├→ serve → END
        └→ heat_rice ───────┘

from typing import Annotated
import operator

def my_merge(a, b):
    """自定义合并逻辑：这里直接返回 b（也可改成 a+b 等）"""
    return b

class MyState(TypedDict):
    menu: Annotated[list, my_merge]   # 👈 指定合并函数
    all_ready: bool

def build_subgraph():
    g = StateGraph(SubState)            # 👈 子图有自己的 State
    g.add_node("hello", say_hello)
    g.add_node("bye", say_bye)
    g.add_edge(START, "hello")
    g.add_edge("hello", "bye")
    g.add_edge("bye", END)
    return g.compile()

def build_main_graph():
    subgraph = build_subgraph()         # 👈 先创建子图实例
    g = StateGraph(MainState)
    g.add_node("start", start_main)
    g.add_node("sub", subgraph)         # 👈 直接把子图当节点加入
    g.add_node("finish", finish_main)
    g.add_edge(START, "start")
    g.add_edge("start", "sub")
    g.add_edge("sub", "finish")
    g.add_edge("finish", END)
    return g.compile()

def run_subgraph(state: MainState):
    subgraph = build_subgraph()
    # 👇 手动构造子图输入（字段名/数据可任意转换）
    sub_state = {"name": "小帅"}
    result = subgraph.invoke(sub_state)
    # 👇 手动把子图输出合并回主图 state
    state["sub_output"] = result
    return state

def build_main_graph():
    g = StateGraph(MainState)
    g.add_node("start", start_main)
    g.add_node("sub", run_subgraph)     # 👈 包装函数，不是子图本身
    ...

# invoke：阻塞到全部完成
result = graph.invoke({"question": "..."})

# stream：每个节点完成时立刻返回一次
for chunk in graph.stream({"question": "为什么天空是蓝色的？"}):
    print(chunk)
# 输出示例：
# {"思考节点": {"thought": "是怎么染色的呢？"}}
# {"回答节点": {"answer": "是用彩笔染色的。"}}

import asyncio

async def main(state: MyState):
    graph = build_graph()
    result = await graph.ainvoke(state)   # 👈 异步版本
    return result

# 启动异步事件循环
asyncio.run(main({"question": "..."}))

async def run():
    task1 = asyncio.create_task(main({"question": "..."}))   # LangGraph 任务
    task2 = asyncio.create_task(fastprint())                 # 其他异步任务
    await task1
    await task2
asyncio.run(run())

async def main(state: MyState):
    graph = build_graph()
    async for event in graph.astream(state):   # 👈 异步流式
        print(event)

asyncio.run(main({"question": "..."}))

from langgraph.graph import StateGraph, START, END
from langgraph.types import RunnableConfig, interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
from typing import TypedDict, Annotated

# 基础状态
class State(TypedDict):
    field: str

# 带合并函数的状态（并行场景必备）
class State(TypedDict):
    menu: Annotated[list, operator.add]   # 多节点写同一字段时自动合并

# 最简签名
def node(state): ...

# 带配置（需要读 config 时）
def node(state, config: RunnableConfig): ...

# 异步节点
async def node(state): ...

01 config          → 学会往节点传运行时参数
   ↓
02 thread_id       → 理解会话隔离的概念
   ↓
03 memory          → 掌握多轮记忆（thread_id + memory 配合）
   ↓
04 human_approval  → 学会中断恢复（依赖 memory）
   ↓
05 并行            → 学会 fan-out / fan-in
   ↓
06 并行 list       → 解决并行写冲突（依赖 05）
   ↓
07/08 子图         → 学会复用子流程
   ↓
09 stream          → 同步流式
   ↓
10 ainvoke         → 异步执行
   ↓
11 astream         → 异步流式（生产推荐）