Agent开发，LangGraph的图设计 / ai #43

https://langchain-ai.github.io/langgraph/

简介

LangGraph, 从这名字大概也能猜到它和LangChain的关系吧。不错，它也是LangChain团队开发的一个专用于Agent的高级库。它设计的特点不同于LangChain的单线设计，而是用图来启用循环的多线设计。

LangGraph 在图这个概念上有很大的侧重，它的出现就是要解决线性序列的局限性问题，而解决的方法就是循环图。在LangGraph框架中，用图来管理代理的生命周期并在其状态内将暂存器作为消息进行跟踪，增加了以循环方式跨各种计算步骤协调多个链或参与者的功能。这就与 LangChain 将代理视为可以附加工具和插入某些提示的对象不同，对于图来说，意味着我们可以从任何可运行的功能或代理或链作为一个程序的起点。

LangGraph通过组合Nodes和Edges去创建复杂的循环工作流程，通过消息传递的方式串联所有的节点形成一个通路。那么维持消息能够及时的更新并向该去的地方传递，则依赖langGraph构建的State概念。 在LangGraph构建的流程中，每次执行都会启动一个状态，图中的节点在处理时会传递和修改该状态。这个状态不仅仅是一组静态数据，而是由每个节点的输出动态更新，然后影响循环内的后续操作。

从LangGraph官方的定义看，该框架是一个用于使用大模型构建有状态、多参与者应用程序的库，可以创建代理和多代理工作流程。而其官方自己总结的LangGraph的优势则是：

• 循环和分支：在应用程序中实现循环和条件。
• 持久性：在图中的每个步骤之后自动保存状态。随时暂停和恢复图形执行，以支持错误恢复、人机交互工作流程等。
• 人机交互：中断图形执行以批准或编辑代理计划的下一个操作。
• 流支持：流输出由每个节点生成（包括令牌流）。
• 与LangChain集成：LangGraph 与LangChain和LangSmith无缝集成。

下载与资源

LangGraph |
LangGraph Github |
LangChain github |
LangChain |

安装

virtualenv lang_env
source lang_env/Scripts/activate   //windows
pip install langgraph -i https://pypi.tuna.tsinghua.edu.cn/simple （清华镜像）
pip install langchain langchain-openai -i https://pypi.tuna.tsinghua.edu.cn/simple 

GraphState

在图中提到了节点、边、状态和路由四个概念。

定义图时要做的第一件事是定义图的State。状态表示会随着图计算的进行而维护和更新的上下文或记忆。它用来确保图中的每个步骤都可以访问先前步骤的相关信息，从而可以根据整个过程中积累的数据进行动态决策。这个过程通过状态图StateGraph类实现，它继承自 Graph 类，这意味着 StateGraph 会使用或扩展基类的属性和方法。

# 构建图
builder = StateGraph(dict) 

Nodes

在 LangGraph 中，节点是一个 python 函数（sync 或async），接收当前State作为输入，执行自定义的计算，并返回更新的State。所以其中第一个位置参数是state 。

def agent_node(state:InputState):
    print("我是一个AI Agent。")
    return 

定义好了节点以后，我们需要使用add_node方法将这些节点添加到图中。在将节点添加到图中的时候，可以自定义节点的名称。而如果不指定名称，则会为自动指定一个与函数名称等效的默认名称。代码如下：

builder.add_node("agent_node", agent_node)
builder.add_node("action_node", action_node)

Edges

Edges（边）用来定义逻辑如何路由以及图何时开始与停止。这是代理工作以及不同节点如何相互通信的重要组成部分。有几种关键的边类型：

• 普通边：直接从一个节点到下一个节点。
• 条件边：调用函数来确定下一个要转到的节点。
• 入口点：当用户输入到达时首先调用哪个节点。
• 条件入口点：调用函数来确定当用户输入到达时首先调用哪个节点。

同样，我们先看普通边。如果直接想从节点A到节点B，可以直接使用add_edge方法。注意：LangGraph有两个特殊的节点：START和END。START表示将用户输入发送到图的节点。使用该节点的主要目的是确定应该首先调用哪些节点。END节点是代表终端节点的特殊节点。当想要指示哪些边完成后没有任何操作时，将使用该节点。因此，一个完整的图就可以使用如下代码进行定义：

from langgraph.graph import START, END

builder.add_edge(START, "agent_node")
builder.add_edge("agent_node", "action_node")
builder.add_edge("action_node", END)
# 最后，通过`compile`编译图。在编译过程中，会对图结构执行一些基本检查（如有没有孤立节点等）。
graph = builder.compile()

graph.invoke({"question":"hello，你好"})

一个完整案例

from langgraph.graph import StateGraph
from typing_extensions import TypedDict
from langgraph.graph import START, END
from dotenv import dotenv_values
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate


env_vars = dotenv_values('.env')
OPENAI_KEY = env_vars['OPENAI_API_KEY'] 
OPENAI_BASE_URL = env_vars['OPENAI_API_BASE'] 


# 定义输入的模式
class InputState(TypedDict):
    question: str


# 定义输出的模式
class OutputState(TypedDict):
    answer: str


# 将 InputState 和 OutputState 这两个 TypedDict 类型合并成一个更全面的字典类型。
class OverallState(InputState, OutputState):
    pass


def agent_node(state: InputState):
    print("I am AI Agent ", InputState, state["question"])
    return {"question": state["question"]}


def action_node(state: InputState):
    print("agent action", InputState, state["question"])
    step = state["question"]
    return {"answer": f"input question {step}，succeed！"}


def llm_node(state: InputState):
    messages = [
        ("user", state["question"])
    ]


    llm = ChatOpenAI(model="gpt-3.5-turbo", api_key=OPENAI_KEY,base_url=OPENAI_BASE_URL)
    # gpt-3.5-turbo o1-mini

    response = llm.invoke(messages) 

    return {"answer": response.content}


# 明确指定它的输入和输出数据的结构或模式
builder = StateGraph(OverallState, input=InputState, output=OutputState)

# 添加节点
builder.add_node("llm_node", llm_node)

# 添加边
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", END)

# 编译图
graph = builder.compile()

final_answer = graph.invoke({"question":"how are you"})
print(final_answer["answer"])

如果没出错，它可以一把跑通的！LangGraph``LangChain安装起来也没什么太大困难，这也是它优势的地方，不像有些Agent是死活装不上。如果成功地迈出了第一步，那么后面的内容也会比较顺利，让我们期待吧。