📡 流式输出

# invoke：等到全部结束才返回（用户干等）
result = app.invoke(inputs)
print(result)

# stream：边执行边产出（用户实时看到）
for chunk in app.stream(inputs):
    print(chunk)   # 每个 chunk 是一个中间更新

StreamMode = Literal[
    "values",      # 每步后输出【完整状态】
    "updates",     # 每步后只输出【节点名 + 该步的更新】
    "custom",      # 节点内用 StreamWriter 发【自定义数据】
    "messages",    # LLM 调用【逐 token】输出 + 元数据
    "checkpoints", # 每次存 checkpoint 时输出事件
    "tasks",       # 任务开始/结束时输出事件
    "debug",       # checkpoints + tasks（调试用）
]

# pip install langgraph
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import StreamWriter

class State(TypedDict):
    count: int
    log: list[str]

def step_a(state, writer: StreamWriter):
    writer({"progress": "A 开始处理"})   # custom 流：发自定义数据
    return {"count": state["count"] + 1, "log": state["log"] + ["A"]}

def step_b(state):
    return {"count": state["count"] + 1, "log": state["log"] + ["B"]}

graph = StateGraph(State)
graph.add_node("a", step_a)
graph.add_node("b", step_b)
graph.add_edge(START, "a"); graph.add_edge("a", "b"); graph.add_edge("b", END)
app = graph.compile()

inputs = {"count": 0, "log": []}

# ============ ① updates（最常用）============
print("=== updates ===")
for chunk in app.stream(inputs, stream_mode="updates"):
    print(chunk)
# {'a': {'count': 1, 'log': ['A']}}
# {'b': {'count': 2, 'log': ['A', 'B']}}

# ============ ② values（完整状态）============
print("\n=== values ===")
for chunk in app.stream(inputs, stream_mode="values"):
    print(chunk)
# {'count': 1, 'log': ['A']}
# {'count': 2, 'log': ['A', 'B']}

# ============ ③ custom（节点内自定义流）============
print("\n=== custom ===")
for chunk in app.stream(inputs, stream_mode="custom"):
    print(chunk)
# {'progress': 'A 开始处理'}

# ============ ④ 多模式同时（返回元组）============
print("\n=== updates + custom 同时 ===")
for mode, chunk in app.stream(inputs, stream_mode=["updates", "custom"]):
    print(f"[{mode}] {chunk}")
# [custom] {'progress': 'A 开始处理'}
# [updates] {'a': {...}}
# [updates] {'b': {...}}

# pip install langgraph langchain-openai
from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")

def chatbot(state):
    return {"messages": [model.invoke(state["messages"])]}

app = StateGraph(MessagesState).add_node("chatbot", chatbot).add_edge(START,"chatbot").add_edge("chatbot",END).compile()

# messages 模式：逐 token 输出
for msg, metadata in app.stream(
    {"messages": [("human", "讲个笑话")]},
    stream_mode="messages",
):
    # msg 是 AIMessageChunk（每次一小片 token）
    # metadata 含 langgraph_node, langgraph_step 等
    print(msg.content, end="", flush=True)
print()
# 实时逐字输出："从前有个..."（打字机效果）

StreamWriter = Callable[[Any], None]
"""接受单个参数、写入输出流的 Callable。
当节点把 writer 作为关键字参数请求时，自动注入。
不用 stream_mode="custom" 时，它是空操作（no-op）。"""

# 用法：节点签名声明 writer 参数，自动注入
def long_task(state, writer: StreamWriter) -> dict:
    for i in range(3):
        # 模拟分步处理，每步推送进度
        writer({"step": i, "progress": f"{i*33}% 已完成"})
        time.sleep(1)
    return {"result": "完成"}

# stream_mode="custom" 才能收到 writer 发的数据
for chunk in app.stream(inputs, stream_mode="custom"):
    print(chunk)
# {'step': 0, 'progress': '0% 已完成'}
# {'step': 1, 'progress': '33% 已完成'}
# {'step': 2, 'progress': '66% 已完成'}

async for event in app.astream_events(inputs, version="v2"):
    kind = event["event"]
    if kind == "on_chat_model_stream":
        # LLM 逐 token
        print(event["data"]["chunk"].content, end="")
    elif kind == "on_tool_start":
        print(f"\n[工具开始] {event['name']}")
    elif kind == "on_tool_end":
        print(f"[工具结束] {event['name']}")
    # 还有 on_chain_start/end, on_chat_model_start/end 等