LangGraph基础

LangChain带来了很多AI方面的基础模块，并且也提供了一些预构建的Agent。然而在复杂、动态的调度上，还是需要自己去手动组装。LangGraph的出现，图构建的方式解决了这个问题。

主要模块

LangGraph的主要模块是Graph、Node、State、Edge、CheckPoint

核心概念	核心指责	说明
图 (Graph)	定义工作流的整体骨架和流转逻辑	像绘制业务流程图，规定了所有步骤和步骤间的走向
节点 (Node)	承载工作流中的具体任务和业务逻辑	流程图中的每个具体操作步骤，如审批、调用API
状态 (State)	在节点间共享和传递数据与上下文	一个在所有步骤间流转的共享文件夹，每个步骤都能读取或写入信息
边 (Edge)	根据条件决定工作流的执行路径	流程图中的箭头线，可以根据条件决定下一步是A还是B
检查点 (Checkpoint)	实现工作流的持久化、中断恢复和可观测性	执行过程中的存档，异常或者用户交互时支持重入

从一个小demo入门，构建一个最简单的Graph:

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from langgraph.graph import StateGraph, START, END

llm = ChatOpenAI(
    model="DeepSeek-V3-0324",
    base_url="xx",
    api_key="xx",
)

# 定义State
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 定义Node
def chat(state: State) -> State:
    res = llm.invoke(state["messages"])
    return {"messages": [res]}

# 编译Edge、Graph
graph_builder = StateGraph(State)
graph_builder.add_node("chat", chat)
graph_builder.add_edge(START, "chat")
graph_builder.add_edge("chat", END)
graph = graph_builder.compile()


def run(user_input: str):
    return graph.invoke({"messages": [{"role": "user", "content": user_input}]})

run("你好")

这里的State，是用来在整个图里进行数据流转的，所以在LangGraph里是不推荐使用全局变量的。后面的add_messages，是一个reducer，这里理解成LangGraph会自动为每个State的元素，自动执行xxx(old_state, new_state) -> update_state就可以, 比如这里实际上就是进行了拼接:

def add_messages(old_messages, new_messages) -> update_messages
    return [old_messages, new_messages]