LangGraph 入门

用一个极简的图引擎演示 LangGraph 的核心抽象：节点（Node）、边（Edge）、 入口/出口。每个节点是一个接收状态并返回状态更新的函数，图按边定义的顺序执行。 本实现为纯 Python，便于离线测试；概念与 LangGraph 的 StateGraph 一一对应。 对应教程：第8卷《LangGraph》第1章 概览与第2章 图工作流。

文件路径: examples/volume-08-langgraph\hello_graph.py

---

源代码

"""LangGraph Hello World 示例。

用一个极简的图引擎演示 LangGraph 的核心抽象：**节点（Node）**、**边（Edge）**、
**入口/出口**。每个节点是一个接收状态并返回状态更新的函数，图按边定义的顺序执行。

本实现为纯 Python，便于离线测试；概念与 LangGraph 的 ``StateGraph`` 一一对应。

对应教程：第8卷《LangGraph》第1章 概览与第2章 图工作流。
"""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any

# 结束哨兵：与 LangGraph 的 END 类似
END = "__END__"
State = dict[str, Any]
NodeFunc = Callable[[State], State]


@dataclass
class Graph:
    """极简有向图。

    Attributes:
        entry: 入口节点名。
    """

    entry: str | None = None
    _nodes: dict[str, NodeFunc] = field(default_factory=dict)
    _edges: dict[str, str] = field(default_factory=dict)

    def add_node(self, name: str, func: NodeFunc) -> Graph:
        """添加节点。

        Args:
            name: 节点名。
            func: 节点函数，接收状态返回状态更新。

        Returns:
            ``self``。
        """
        self._nodes[name] = func
        return self

    def add_edge(self, from_node: str, to_node: str) -> Graph:
        """添加固定边。

        Args:
            from_node: 起点节点名。
            to_node: 终点节点名，可为 ``END``。

        Returns:
            ``self``。
        """
        self._edges[from_node] = to_node
        return self

    def set_entry_point(self, name: str) -> Graph:
        """设置入口节点。"""
        self.entry = name
        return self

    def invoke(self, initial_state: State, max_steps: int = 20) -> State:
        """从入口执行图。

        Args:
            initial_state: 初始状态。
            max_steps: 最大步数，防止环导致死循环。

        Returns:
            最终状态。

        Raises:
            ValueError: 未设置入口或节点缺失时。
            RuntimeError: 超过最大步数。
        """
        if self.entry is None:
            raise ValueError("未设置入口节点")
        state = dict(initial_state)
        current = self.entry
        for _ in range(max_steps):
            if current == END:
                break
            if current not in self._nodes:
                raise ValueError(f"未知节点: {current}")
            update = self._nodes[current](state)
            if update:
                state.update(update)
            current = self._edges.get(current, END)
        else:
            raise RuntimeError(f"图执行超过最大步数 {max_steps}")
        return state


def build_hello_graph() -> Graph:
    """构建一个 Hello World 图：greet -> farewell。

    Returns:
        已编译的 ``Graph``。
    """
    g = Graph()
    g.add_node("greet", lambda s: {"message": f"你好，{s.get('user', '朋友')}！"})
    g.add_node("farewell", lambda s: {"message": s["message"] + " 祝你一切顺利。"})
    g.add_edge("greet", "farewell")
    g.add_edge("farewell", END)
    g.set_entry_point("greet")
    return g


def main() -> None:
    """演示 Hello World 图。"""
    graph = build_hello_graph()
    result = graph.invoke({"user": "AI 学习者"})
    print(result["message"])


if __name__ == "__main__":
    main()