LangGraph--框架核心思想

        谈不上理解多深，langgraph他是一个框架，定义什么样的输入输出数据，交给了我们自己，框架不会去做，当然也有框架的一些参数，这些参数我们直接使用即可，因此该框架的核心就是你是需要做什么事？把这些事拆开成一步步的工作节点这就是节点，工作节点之间肯定有数据交互和选择，这就是边了，也就是说整体工作步骤你是设计好的，只是在调用工具时大模型自主决策，我们不用关系，但是整个系统的工作节点和边的选择是我们需要定义的，因此深入理解节点和边就很重要了，对于数据流怎么参数，其实就是State，我们自己定义的参数数据，这个参数将贯穿整个agent的节点，都是通过定义的参数结构体进行交互的，但是还需要理解langgarph他是怎么执行的，他是按照层或者回合进行执行，具体如下：

先定义一个数据流结构体State：

import operator
from typing import Annotated,Any
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START,ENDclass State(TypedDict):aggregate:Annotated[list, operator.add] # 只添加不覆盖

该数据流就是一个字典，只有一个键aggregate，值是一个list，该list只添加不覆盖数据

定义节点

# 定义多个节点
def a(state:State):print(f'把 "A" 添加到{state["aggregate"]}')return {"aggregate":["A"]}
def b(state:State):print(f'把 "B" 添加到{state["aggregate"]}')return {"aggregate":["B"]}
def c(state:State):print(f'把 "C" 添加到{state["aggregate"]}')return {"aggregate":["C"]}
def d(state:State):print(f'把 "D" 添加到{state["aggregate"]}')return {"aggregate":["D"]}builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_node(d)
builder.add_edge(START,"a")
builder.add_edge("a","b")
builder.add_edge("a","c")
builder.add_edge("b","d")
builder.add_edge("c","d")
builder.add_edge("d",END)
graph = builder.compile()from IPython.display import Image,display
display(Image(graph.get_graph().draw_mermaid_png()))

分别定义了a\b\c\d 四个节点，每个几点把自己的数据添加总体数据字典中的列表了，然后把节点连接起来，图如下：

这里我们解释一些，langgraph是怎么执行a节点是一层，b和c是一层 ，d是一层，那么添加数据顺序一个是先添加a，在添加b和c，最后添加d：

result  = graph.invoke({"aggregate":[]}, {"configurable":{"thread_id":"foo"}})
result

把 "A" 添加到[]
把 "B" 添加到['A']
把 "C" 添加到['A']
把 "D" 添加到['A', 'B', 'C']
{'aggregate': ['A', 'B', 'C', 'D']}

从结果可以看到，和我们理解的是一样的，A是一层，b和c是一层，他是等到A添加进去以后（该开始是空列表），在然后把b和c添加进去。加入我们的图是怎样的：

 这样的图，langgraph 是怎么运行的呢？

还是按照层来考虑，a是一层，b个c是一层，b_2和d是单独一层（这里的d是来源c下面），d是单独一层（这里的d是来源b_2下面的）langgraph执行不管这些，还是一层一层的执行了，所以执行结果应该是：

A；A\B\C;A\B\C\B_2\D（这个D是从C这里来的）;A\B\C\B_2\D\D (这个d就是最后一个节点了，即从B_2过来的)

代码：

def b_2(state:State):print(f'把 "B_2" 添加到{state["aggregate"]}')return {"aggregate":["B_2"]}builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(b_2)
builder.add_node(c)
builder.add_node(d)
builder.add_edge(START,"a")
builder.add_edge("a","b")
builder.add_edge("a","c")
builder.add_edge("b","b_2")
builder.add_edge("b_2","d")
builder.add_edge("c","d")
builder.add_edge("d",END)
graph = builder.compile()from IPython.display import Image,display
display(Image(graph.get_graph().draw_mermaid_png()))
result  = graph.invoke({"aggregate":[]}, {"configurable":{"thread_id":"foo"}})
result

 结果如下：

把 "A" 添加到[]
把 "B" 添加到['A']
把 "C" 添加到['A']
把 "B_2" 添加到['A', 'B', 'C']
把 "D" 添加到['A', 'B', 'C']
把 "D" 添加到['A', 'B', 'C', 'B_2', 'D']
{'aggregate': ['A', 'B', 'C', 'B_2', 'D', 'D']}

从上面可以发现，和上面分析是一样的，但是我们最终希望要的是 ['A', 'B', 'C', 'B_2', 'D']就可以了，能否实现呢？当然可以，我们可以想想怎么实现，如果让你写langgraph框架代码你会怎么实现呢？其实很简单，我们只需要把c和b_2作为同级即可，怎么实现呢？使用列表，在边那边同时连即可，具体如下：

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(b_2)
builder.add_node(c)
builder.add_node(d)
builder.add_edge(START,"a")
builder.add_edge("a","b")
builder.add_edge("a","c")
builder.add_edge("b","b_2")
builder.add_edge(["b_2", "c"],"d")
# builder.add_edge("b_2","d")
# builder.add_edge("c","d")
builder.add_edge("d",END)
graph = builder.compile()from IPython.display import Image,display
display(Image(graph.get_graph().draw_mermaid_png()))
result  = graph.invoke({"aggregate":[]}, {"configurable":{"thread_id":"foo"}})
result

把 "A" 添加到[]
把 "B" 添加到['A']
把 "C" 添加到['A']
把 "B_2" 添加到['A', 'B', 'C']
把 "D" 添加到['A', 'B', 'C', 'B_2']
{'aggregate': ['A', 'B', 'C', 'B_2', 'D']}

从上面可以看出输出的图都是一样的，只是层级关系改变了，因为我们通过边把b_2和c强制为同一节点了，这样就可以按照我们的预期工作了。

同时我们可以发现整个过程我们只处理了我们定义的参数流，然后定义节点和边即可，每个节点你都可以定位很复杂的功能或者智能体这样就可以完成很复杂的功能的智能体了