ai-agent/05_langgraph/simple_graph.py

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

# ============================================================
# 第 1 步：定义 State —— 描述整个工作流在运行时的数据结构
# ============================================================
class CounterState(TypedDict):
    """工作流的状态：一个简单的计数器"""
    count: int           # 当前计数

# ============================================================
# 第 2 步：写节点函数 —— 每个节点做一件事
# ============================================================
def step_one(state: CounterState) -> dict:
    """第一个处理节点：count + 1"""
    new_count = state["count"] + 1
    return {
        "count": new_count
    }


def step_two(state: CounterState) -> dict:
    """第二个处理节点：count * 2"""
    new_count = state["count"] * 2
    return {
        "count": new_count
    }


# ============================================================
# 第 3 步：创建 StateGraph 对象
# ============================================================
builder = StateGraph(CounterState)

# ============================================================
# 第 4 步：注册节点，连接边
# ============================================================
builder.add_node("step_one", step_one)       # 把函数注册为节点
builder.add_node("step_two", step_two)

builder.add_edge(START, "step_one")          # START → step_one
builder.add_edge("step_one", "step_two")     # step_one → step_two
builder.add_edge("step_two", END)            # step_two → END

# ============================================================
# 第 5 步：编译图（编译后才可执行）
# ============================================================
graph = builder.compile()

# ============================================================
# 第 6 步：执行
# ============================================================
result = graph.invoke({"count": 1, "log": []})
print(result)
# 输出: {'count': 4}

#显示图结构
from IPython.display import Image,display
display(Image(graph.get_graph().draw_mermaid_png()))