第10章: LangGraph 工作流编排与智能体

# 安装 LangGraph（它会自动安装 langchain-core 依赖）
pip install langgraph

# 如果使用通义千问模型
pip install dashscope langchain-community

from typing import TypedDict, Annotated, List
import operator

# ========== 方式1：最简单的 State（覆盖模式）==========
class SimpleState(TypedDict):
    """每个字段都是覆盖模式：新值直接替换旧值"""
    user_name: str      # 用户名
    age: int            # 年龄
    score: float        # 分数

# 示例：覆盖模式的行为
# 初始: state = {"user_name": "张三", "age": 25, "score": 80.0}
# 节点返回: {"age": 26}
# 结果: state = {"user_name": "张三", "age": 26, "score": 80.0}
#                                      ↑ 只有 age 被更新


# ========== 方式2：带 Reducer 的 State（累加模式）==========
class ChatState(TypedDict):
    """使用 Annotated 定义字段的合并策略"""
    # operator.add：新值追加到列表末尾（不覆盖）
    messages: Annotated[List[str], operator.add]
    
    # 无 Annotated：普通覆盖模式
    current_speaker: str
    turn_count: int

# 示例：Reducer 的行为
# 初始: state = {
#     "messages": ["用户: 你好"],
#     "current_speaker": "user",
#     "turn_count": 1
# }
# 节点返回: {
#     "messages": ["AI: 你好，我是助手"],
#     "current_speaker": "bot",
#     "turn_count": 2
# }
# 结果: state = {
#     "messages": ["用户: 你好", "AI: 你好，我是助手"],  # ← 追加（operator.add）
#     "current_speaker": "bot",                        # ← 覆盖
#     "turn_count": 2                                  # ← 覆盖
# }


# ========== 方式3：自定义 Reducer 函数 ==========
def merge_metadata(old: dict, new: dict) -> dict:
    """自定义合并逻辑：合并两个字典"""
    return {**old, **new}

class AdvancedState(TypedDict):
    messages: Annotated[List[str], operator.add]
    # 自定义 reducer：合并元数据字典
    metadata: Annotated[dict, merge_metadata]
    status: str

# 示例：自定义 Reducer 的行为
# 初始: state = {
#     "messages": ["消息1"],
#     "metadata": {"source": "web", "user_id": "001"},
#     "status": "processing"
# }
# 节点返回: {
#     "metadata": {"timestamp": "2024-03-21", "version": "1.0"}
# }
# 结果: state = {
#     "messages": ["消息1"],
#     "metadata": {"source": "web", "user_id": "001", "timestamp": "2024-03-21", "version": "1.0"},
#     "status": "processing"
# }

# ========== 模式1：简单处理节点 ==========
def greet_node(state: ChatState) -> dict:
    """读取 state，返回要更新的字段"""
    user_name = state.get("current_speaker", "朋友")
    return {
        "messages": [f"你好 {user_name}，欢迎使用 LangGraph！"],
        "turn_count": state.get("turn_count", 0) + 1
    }


# ========== 模式2：调用 LLM 的节点 ==========
def llm_node(state: ChatState) -> dict:
    """调用大语言模型处理用户消息"""
    from langchain_community.chat_models import ChatTongyi
    
    llm = ChatTongyi(model_name="qwen-turbo")
    user_message = state["messages"][-1]  # 获取最后一条消息
    
    # 调用 LLM
    response = llm.invoke(user_message)
    
    return {
        "messages": [f"AI: {response.content}"],
        "current_speaker": "bot"
    }


# ========== 模式3：条件判断节点 ==========
def check_sentiment_node(state: ChatState) -> dict:
    """分析用户情绪"""
    last_message = state["messages"][-1]
    
    # 简单的情绪判断逻辑
    if "谢谢" in last_message or "感谢" in last_message:
        sentiment = "positive"
    elif "问题" in last_message or "错误" in last_message:
        sentiment = "negative"
    else:
        sentiment = "neutral"
    
    return {
        "metadata": {"sentiment": sentiment},
        "messages": [f"[系统] 检测到情绪: {sentiment}"]
    }


# ========== 模式4：工具调用节点 ==========
def search_database_node(state: ChatState) -> dict:
    """查询数据库的节点"""
    query = state["messages"][-1]
    
    # 模拟数据库查询
    db_results = {
        "LangChain": "LangChain 是一个用于构建 LLM 应用的框架",
        "LangGraph": "LangGraph 是 LangChain 的工作流编排框架"
    }
    
    result = db_results.get(query, "未找到相关信息")
    
    return {
        "messages": [f"数据库查询结果: {result}"],
        "metadata": {"source": "database"}
    }


# ========== 关键点：节点的返回值 ==========
# ✅ 正确：只返回要更新的字段
def good_node(state: ChatState) -> dict:
    return {"messages": ["新消息"]}  # ✅ 只返回需要更新的字段

# ❌ 错误：返回完整的 state（不推荐，但也能工作）
def bad_node(state: ChatState) -> dict:
    state["messages"].append("新消息")  # ❌ 直接修改 state（不推荐）
    return state  # ❌ 返回完整 state（冗余）

# ========== 示例1：普通边（固定流程）==========
from langgraph.graph import StateGraph, START, END

graph = StateGraph(ChatState)

# 添加节点
graph.add_node("greet", greet_node)
graph.add_node("process", process_node)
graph.add_node("farewell", farewell_node)

# 添加普通边：固定顺序执行
graph.add_edge(START, "greet")      # 开始 → greet
graph.add_edge("greet", "process")  # greet → process
graph.add_edge("process", "farewell")  # process → farewell
graph.add_edge("farewell", END)     # farewell → 结束

# 流程图：START → greet → process → farewell → END


# ========== 示例2：条件边（动态路由）==========
# 定义路由函数
def route_by_sentiment(state: ChatState) -> str:
    """根据情绪决定下一个节点"""
    sentiment = state.get("metadata", {}).get("sentiment", "neutral")
    
    if sentiment == "positive":
        return "thank_you_node"  # 积极情绪 → 感谢节点
    elif sentiment == "negative":
        return "help_node"       # 消极情绪 → 帮助节点
    else:
        return "continue_node"   # 中性情绪 → 继续对话节点

graph = StateGraph(ChatState)

# 添加节点
graph.add_node("check_sentiment", check_sentiment_node)
graph.add_node("thank_you_node", thank_you_node)
graph.add_node("help_node", help_node)
graph.add_node("continue_node", continue_node)

# 添加条件边
graph.add_edge(START, "check_sentiment")
graph.add_conditional_edges(
    "check_sentiment",           # 从哪个节点出发
    route_by_sentiment,          # 路由函数（返回下一个节点的名称）
    {
        "thank_you_node": "thank_you_node",  # 映射：函数返回值 → 实际节点名
        "help_node": "help_node",
        "continue_node": "continue_node"
    }
)

# 所有分支最终都到 END
graph.add_edge("thank_you_node", END)
graph.add_edge("help_node", END)
graph.add_edge("continue_node", END)

# 流程图：
#                  ┌→ thank_you_node → END (positive)
# START → check ──┼→ help_node → END (negative)
#                  └→ continue_node → END (neutral)


# ========== 示例3：复杂条件边（多条件判断）==========
def complex_router(state: ChatState) -> str:
    """复杂的路由逻辑"""
    last_message = state["messages"][-1]
    turn_count = state.get("turn_count", 0)
    
    # 多条件判断
    if turn_count > 10:
        return END  # 超过10轮对话，直接结束
    elif "再见" in last_message or "拜拜" in last_message:
        return "farewell_node"  # 用户说再见
    elif "帮助" in last_message or "?" in last_message:
        return "help_node"  # 用户需要帮助
    else:
        return "llm_node"  # 正常对话

graph.add_conditional_edges(
    "process",
    complex_router,
    {
        "farewell_node": "farewell_node",
        "help_node": "help_node",
        "llm_node": "llm_node",
        END: END  # 特殊：直接结束
    }
)

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

# ===== 第1步：定义 State =====
class MyState(TypedDict):
    messages: Annotated[List[str], operator.add]  # 消息会累加
    user_name: str

# ===== 第2步：定义 Node（普通 Python 函数）=====
def greet(state: MyState):
    return {"messages": [f"👋 你好 {state['user_name']}！"]}

def process(state: MyState):
    return {"messages": ["⚙️ 正在处理你的请求..."]}

def farewell(state: MyState):
    return {"messages": ["👋 处理完成，再见！"]}

# ===== 第3步：构建图 =====
graph = StateGraph(MyState)

# 添加节点
graph.add_node("greet", greet)
graph.add_node("process", process)
graph.add_node("farewell", farewell)

# 添加边（定义执行顺序）
graph.add_edge(START, "greet")       # 起点 → greet
graph.add_edge("greet", "process")   # greet → process
graph.add_edge("process", "farewell")# process → farewell
graph.add_edge("farewell", END)      # farewell → 终点

# ===== 第4步：编译并运行 =====
app = graph.compile()

result = app.invoke({
    "messages": [],
    "user_name": "小明"
})

print("最终状态:")
for msg in result["messages"]:
    print(f"  {msg}")

# 输出:
#   👋 你好 小明！
#   ⚙️ 正在处理你的请求...
#   👋 处理完成，再见！

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

# ===== State =====
class CustomerState(TypedDict):
    messages: Annotated[List[str], operator.add]
    user_input: str
    intent: str       # 识别出的意图

# ===== 节点 =====
def classify(state: CustomerState):
    """意图识别节点"""
    text = state["user_input"].lower()
    if "退款" in text or "投诉" in text:
        intent = "complaint"
    elif "查" in text or "订单" in text:
        intent = "query"
    else:
        intent = "chat"
    return {"intent": intent, "messages": [f"🔍 识别意图: {intent}"]}

def handle_query(state: CustomerState):
    return {"messages": ["📋 正在查询您的订单信息..."]}

def handle_complaint(state: CustomerState):
    return {"messages": ["🔧 已记录您的投诉，客服将尽快处理"]}

def handle_chat(state: CustomerState):
    return {"messages": ["💬 感谢您的咨询，还有什么可以帮您？"]}

# ===== 路由函数 =====
# 返回值必须是下一个节点的名字
def route_by_intent(state: CustomerState) -> Literal["handle_query", "handle_complaint", "handle_chat"]:
    return f"handle_{state['intent']}"

# ===== 构建图 =====
graph = StateGraph(CustomerState)

graph.add_node("classify", classify)
graph.add_node("handle_query", handle_query)
graph.add_node("handle_complaint", handle_complaint)
graph.add_node("handle_chat", handle_chat)

graph.add_edge(START, "classify")

# 关键：条件边 —— classify 之后根据 route_by_intent 的返回值决定走哪条路
graph.add_conditional_edges("classify", route_by_intent)

graph.add_edge("handle_query", END)
graph.add_edge("handle_complaint", END)
graph.add_edge("handle_chat", END)

app = graph.compile()

# ===== 测试 =====
for question in ["查一下我的订单", "我要投诉！", "你好呀"]:
    result = app.invoke({"messages": [], "user_input": question, "intent": ""})
    print(f"\n用户: {question}")
    for msg in result["messages"]:
        print(f"  {msg}")

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

class WriterState(TypedDict):
    messages: Annotated[List[str], operator.add]
    draft: str
    attempt: int       # 第几次尝试
    is_approved: bool  # 是否通过审核

def generate(state: WriterState):
    """生成内容（模拟）"""
    attempt = state["attempt"] + 1
    # 模拟：第3次才写出合格内容
    if attempt >= 3:
        draft = "这是一篇高质量的文章，包含详细数据和分析。共500字。"
    else:
        draft = f"这是第{attempt}次草稿，内容较短。"
    return {
        "draft": draft,
        "attempt": attempt,
        "messages": [f"✍️ 第{attempt}次生成: {draft[:30]}..."]
    }

def check_quality(state: WriterState):
    """质量检查"""
    # 这里是“示例用”的质量规则：要能稳定判定第3次之后的草稿为合格。
    # 真实项目中你会把它替换成：更严格的规则（字数/结构）或 LLM-as-a-judge。
    draft = state["draft"]
    is_good = (
        ("高质量" in draft)  # mock 输出的显式标记
        or ("共500字" in draft)  # mock 输出的显式标记
        or (len(draft) >= 20 and "数据" in draft)  # 兜底：有关键要素且长度足够
    )
    return {
        "is_approved": is_good,
        "messages": [f"{'✅ 质量合格！' if is_good else '❌ 质量不达标，需要重写'}"]
    }

def should_continue(state: WriterState):
    """路由：合格就结束，不合格就继续生成"""
    if state["is_approved"]:
        return END
    if state["attempt"] >= 5:  # 安全阀：最多5次
        return END
    return "generate"

# 构建图
graph = StateGraph(WriterState)
graph.add_node("generate", generate)
graph.add_node("check_quality", check_quality)

graph.add_edge(START, "generate")
graph.add_edge("generate", "check_quality")
# 关键：条件边指回 generate，形成循环
graph.add_conditional_edges("check_quality", should_continue)

app = graph.compile()

result = app.invoke({
    "messages": [], "draft": "", "attempt": 0, "is_approved": False
})

print("执行过程:")
for msg in result["messages"]:
    print(f"  {msg}")
print(f"\n最终草稿: {result['draft']}")
print(f"总共尝试: {result['attempt']}次")

# 输出:
#   ✍️ 第1次生成: 这是第1次草稿，内容较短。...
#   ❌ 质量不达标，需要重写
#   ✍️ 第2次生成: 这是第2次草稿，内容较短。...
#   ❌ 质量不达标，需要重写
#   ✍️ 第3次生成: 这是一篇高质量的文章，包含详细数据和分析...
#   ✅ 质量合格！

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

class DemoState(TypedDict):
    messages: Annotated[List[str], operator.add]
    step: int

def node_a(state: DemoState):
    step = state["step"] + 1
    return {
        "step": step,
        "messages": [f"🧩 A 执行完成(step={step})"]
    }

def node_b(state: DemoState):
    step = state["step"] + 1
    return {
        "step": step,
        "messages": [f"🧩 B 执行完成(step={step})"]
    }

# 构建最小可运行 graph
graph = StateGraph(DemoState)
graph.add_node("A", node_a)
graph.add_node("B", node_b)
graph.add_edge(START, "A")
graph.add_edge("A", "B")
graph.add_edge("B", END)
app = graph.compile()

initial_state: DemoState = {"messages": [], "step": 0}

# ===== invoke vs stream =====

# 方式1: invoke — 一次性返回最终结果
final_state = app.invoke(initial_state)
print("invoke 最终状态:", final_state)

# 方式2: stream — 逐节点返回中间状态（增量更新）
print("\nstream 逐节点事件:")
for event in app.stream(initial_state):
    for node_name, updates in event.items():
        print(f"  📍 节点 [{node_name}] 完成 -> updates={updates}")

# 方式3: stream + stream_mode="values" — 返回每一步的完整状态
print("\nstream(values) 完整状态快照:")
for snapshot in app.stream(initial_state, stream_mode="values"):
    print(f"  snapshot={snapshot}")

# 方式4: astream — 异步流式（用于 Web 服务）
async def run_async_stream():
    print("\nastream 异步逐节点事件:")
    async for event in app.astream(initial_state):
        for node_name, updates in event.items():
            print(f"  📍 节点 [{node_name}] 完成 -> updates={updates}")

asyncio.run(run_async_stream())

from typing import TypedDict, Annotated, List
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class ChatState(TypedDict):
    messages: Annotated[List[str], operator.add]
    user_name: str

def chat_node(state: ChatState):
    """简单的对话节点"""
    user_msg = state["messages"][-1] if state["messages"] else ""
    user_name = state.get("user_name", "用户")
    reply = f"你好 {user_name}！你说：'{user_msg}'"
    return {"messages": [reply]}

# 构建图
graph = StateGraph(ChatState)
graph.add_node("chat", chat_node)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)

# 创建检查点存储并编译
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# 配置：thread_id 用于区分不同会话
config = {"configurable": {"thread_id": "conversation_001"}}

# 第一次运行
result1 = app.invoke(
    {"messages": ["你好"], "user_name": "小明"},
    config=config
)
print("第1次对话:", result1["messages"])

# 第二次运行（同一个 thread_id，会自动恢复上次的状态）
result2 = app.invoke(
    {"messages": ["继续之前的对话"]},
    config=config
)
print("第2次对话:", result2["messages"])
# result2 的 messages 会包含之前所有的消息！

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class ApprovalState(TypedDict):
    plan: str
    approved: bool
    result: str

def generate_plan(state: ApprovalState):
    """生成执行方案"""
    plan = "计划：购买服务器，预算 5000 元"
    return {"plan": plan, "approved": False}

def execute_plan(state: ApprovalState):
    """执行方案（需要审批）"""
    if state["approved"]:
        result = f"已执行：{state['plan']}"
    else:
        result = "方案未通过审批，暂不执行"
    return {"result": result}

# 构建图
graph = StateGraph(ApprovalState)
graph.add_node("generate", generate_plan)
graph.add_node("execute", execute_plan)
graph.add_edge(START, "generate")
graph.add_edge("generate", "execute")
graph.add_edge("execute", END)

# 编译时设置检查点和中断点
memory = MemorySaver()
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["execute"]  # 在 execute 节点前暂停
)

config = {"configurable": {"thread_id": "approval_001"}}
initial_state = {"plan": "", "approved": False, "result": ""}

# 第一次运行：会在 execute 节点前暂停
result = app.invoke(initial_state, config=config)
print("工作流已暂停，等待人工审批...")
print(f"当前状态: {result}")

# 模拟人工审批：修改 approved 状态
print("\n人工审批中...")
approval_input = {"approved": True}  # 人工批准

# 恢复执行（传入新的状态更新）
final = app.invoke(approval_input, config=config)
print(f"审批通过，继续执行: {final}")

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class WorkflowState(TypedDict):
    step: int
    data: str

def process_node(state: WorkflowState):
    return {"step": state["step"] + 1, "data": f"处理步骤 {state['step']}"}

# 构建图
graph = StateGraph(WorkflowState)
graph.add_node("process", process_node)
graph.add_edge(START, "process")
graph.add_edge("process", END)

# 使用 SQLite 持久化（数据保存到文件）
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    app = graph.compile(checkpointer=checkpointer)
    
    config = {"configurable": {"thread_id": "workflow_001"}}
    
    # 第一次运行
    result = app.invoke({"step": 0, "data": ""}, config=config)
    print(f"第1次运行: {result}")
    
    # 服务重启后，仍然可以恢复状态
    result2 = app.invoke({"step": 0, "data": ""}, config=config)
    print(f"恢复后运行: {result2}")  # step 会从上次的值继续

from langgraph.checkpoint.postgres import PostgresSaver
from psycopg import Connection

# 连接 PostgreSQL
connection_string = "postgresql://user:password@localhost:5432/langgraph_db"

with Connection.connect(connection_string) as conn:
    # 创建检查点表（首次使用需要）
    PostgresSaver.create_tables(conn)
    
    # 使用 PostgreSQL 持久化
    checkpointer = PostgresSaver(conn)
    app = graph.compile(checkpointer=checkpointer)
    
    config = {"configurable": {"thread_id": "prod_workflow_001"}}
    result = app.invoke({"step": 0, "data": ""}, config=config)

from typing import TypedDict, Annotated, List
import operator
from langgraph.graph import StateGraph, START, END
from langchain_community.chat_models import ChatTongyi
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]

# 初始化 LLM（通义千问）
llm = ChatTongyi(model="qwen-plus", temperature=0)

def call_model(state: AgentState):
    """调用 LLM"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState):
    """判断是否需要继续（是否有工具调用）"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"  # 有工具调用，去执行工具
    return END          # 没有工具调用，结束

# 构建 Agent 图
graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
# graph.add_node("tools", tool_executor)  # 工具执行节点

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
# graph.add_edge("tools", "agent")  # 工具执行完回到 agent

app = graph.compile()

# ===== 运行示例 =====
# 简单对话（无工具调用）
result = app.invoke({
    "messages": [HumanMessage(content="你好，请介绍一下自己")]
})

print("Agent 回复:")
for msg in result["messages"]:
    print(f"- {type(msg).__name__}: {msg.content}")

# 输出示例:
# Agent 回复:
# - HumanMessage: 你好，请介绍一下自己
# - AIMessage: 我是通义千问大模型，可以回答问题、分析文本、协助创作等。

"""
智能订单处理系统 - 基于 LangGraph StateGraph
展示：条件分支 + 循环重试 + 流式输出
"""
from typing import TypedDict, Annotated, List, Optional
import operator
import random
from langgraph.graph import StateGraph, START, END

# ========== 1. 定义状态 ==========
class OrderState(TypedDict):
    messages: Annotated[List[str], operator.add]  # 处理日志
    order_id: str
    amount: float
    status: str              # pending/validated/paid/shipped/failed
    payment_attempts: int    # 支付尝试次数
    tracking_number: Optional[str]

# ========== 2. 定义节点 ==========
def validate(state: OrderState):
    """验证订单"""
    order_id = state["order_id"]
    amount = state["amount"]

    if amount <= 0:
        return {
            "status": "failed",
            "messages": [f"❌ 订单 {order_id} 验证失败：金额无效 (¥{amount})"]
        }
    if amount > 100000:
        return {
            "status": "failed",
            "messages": [f"❌ 订单 {order_id} 验证失败：超出限额 (¥{amount})"]
        }
    return {
        "status": "validated",
        "messages": [f"✅ 订单 {order_id} 验证通过，金额 ¥{amount}"]
    }

def payment(state: OrderState):
    """处理支付（模拟：有概率失败）"""
    attempts = state["payment_attempts"] + 1
    # 模拟支付：前2次50%概率失败，第3次一定成功
    success = attempts >= 3 or random.random() > 0.5

    if success:
        return {
            "status": "paid",
            "payment_attempts": attempts,
            "messages": [f"💳 第{attempts}次支付成功！"]
        }
    return {
        "payment_attempts": attempts,
        "messages": [f"⚠️ 第{attempts}次支付失败，准备重试..."]
    }

def ship(state: OrderState):
    """安排发货"""
    tracking = f"SF{random.randint(100000, 999999)}"
    return {
        "status": "shipped",
        "tracking_number": tracking,
        "messages": [f"📦 已发货，运单号: {tracking}"]
    }

def notify(state: OrderState):
    """通知客户"""
    return {
        "messages": [f"📧 已通知客户，订单 {state['order_id']} 处理完成"]
    }

# ========== 3. 定义路由 ==========
def after_validate(state: OrderState):
    """验证后的路由"""
    if state["status"] == "failed":
        return END
    return "payment"

def after_payment(state: OrderState):
    """支付后的路由"""
    if state["status"] == "paid":
        return "ship"
    if state["payment_attempts"] >= 3:
        return END  # 超过最大重试次数
    return "payment"  # 重试（循环）

# ========== 4. 构建图 ==========
graph = StateGraph(OrderState)

graph.add_node("validate", validate)
graph.add_node("payment", payment)
graph.add_node("ship", ship)
graph.add_node("notify", notify)

graph.add_edge(START, "validate")
graph.add_conditional_edges("validate", after_validate)
graph.add_conditional_edges("payment", after_payment)
graph.add_edge("ship", "notify")
graph.add_edge("notify", END)

app = graph.compile()

# ========== 5. 运行（流式输出）==========
print("=" * 50)
print("🚀 订单处理系统启动")
print("=" * 50)

initial = {
    "messages": [],
    "order_id": "ORD-2024-001",
    "amount": 5999.0,
    "status": "pending",
    "payment_attempts": 0,
    "tracking_number": None
}

# 使用 stream 观察每个节点的执行
for event in app.stream(initial):
    for node_name, updates in event.items():
        print(f"\n📍 节点 [{node_name}]:")
        if "messages" in updates:
            for msg in updates["messages"]:
                print(f"   {msg}")
        if "status" in updates:
            print(f"   状态: {updates['status']}")

print("\n" + "=" * 50)
print("✅ 订单处理完成")