You are viewing a single comment's thread from:

RE: LangGraph开发实战

in #starnote5 hours ago

参考

对于任意一个简单或者复杂的图来说,都是基于Graph类来构建和管理图结构的。在Graph类中允许添加节点、边,并定义节点间的动态流转逻辑。LangGraph 将智能体的工作流建模为图。您可以通过三个关键组件来定义智能体的行为:

  • State(状态):一种共享数据结构,代表应用程序的当前快照。它可以是任何数据类型,但通常使用共享状态模式进行定义。
  • Nodes(节点):编码代理逻辑的函数。它们接收当前状态作为输入,执行某些计算或产生副作用,并返回更新后的状态。
  • Eeges(边):根据当前状态决定接下来执行哪个节点的函数。它们可以是条件分支,也可以是固定转换。

如下是Graph类的主要组成部分和功能:

from collections import defaultdict
from typing import Any, Callable, Dict, Optional, Set, Tuple, Union, Awaitable, Hashable

class Graph:
    def __init__(self) -> None:
        self.nodes: Dict[str, Any] = {}  # 一个字典,用于存储图中的所有节点。每个节点可以是一个字符串标识或者是一个可调用对象
        self.edges: Set[Tuple[str, str]] = set()  # 一个集合,用来存储图中所有的边,边由一对节点名称组成,表示从一个节点到另一个节点的直接连接。
        self.branches: defaultdict = defaultdict(dict)  # 一个默认字典,用于存储条件分支,允许从一个节点根据特定条件转移到多个不同的节点。
        self.support_multiple_edges = False  # 一个布尔值,指示图是否支持同一对节点间的多条边。
        self.compiled = False    #  一个布尔值,表示图是否已经被编译。编译是指图的结构已经设置完毕,准备进行执行。

    @property
    def _all_edges(self) -> Set[Tuple[str, str]]:
        """
        获取所有的边的信息。
        """
        return self.edges

    def add_node(self, node: Union[str, Callable], action: Optional[Callable] = None, *, metadata: Optional[Dict[str, Any]] = None) -> 'Graph':
        """
        添加一个新节点到图中。节点可以有附加的元数据,这些元数据存储在节点的字典中。
        """
        pass

    def add_edge(self, start_key: str, end_key: str) -> 'Graph':
        """
        在图中添加一条边,连接两个指定的节点。
        """
        pass

    def add_conditional_edges(self, source: str, path: Callable, path_map: Optional[Dict[Hashable, str]] = None, then: Optional[str] = None) -> 'Graph':
        """
        添加一个条件边,允许在执行时根据某个条件从一个节点动态地转移到一个或多个节点。
        """
        pass

    def set_entry_point(self, key: str) -> 'Graph':
        """
        设置图的入口点,即定义图执行的起始节点。
        """
        pass

    def set_conditional_entry_point(self, path: Callable, path_map: Optional[Dict[Hashable, str]] = None, then: Optional[str] = None) -> 'Graph':
        """
        设置一个条件入口点,允许根据条件动态决定图的起始执行点。
        """
        pass

    def set_finish_point(self, key: str) -> 'Graph':
        """
        设置结束点,定义图执行到此节点时将停止。
        """
        pass

    def validate(self, interrupt: Optional[Set[str]] = None) -> 'Graph':
        """
        验证图的结构是否正确,确保所有节点和边的定义都符合逻辑和图的规则。
        """
        pass

    def compile(self, checkpointer=None, interrupt_before: Optional[Set[str]] = None, interrupt_after: Optional[Set[str]] = None, debug: bool = False) -> 'Graph':
        """
        编译图,确认图的结构合法且可执行后,准备图以供执行。
        """
        pass