什么是LangGraph

Langgraph是一个用于构建循环、有状态的多智能体（Multi-agent）应用的框架，它的特点是：

• 持久执行：构建能够在失败后持续运行、长时间运行并从中断处恢复的代理。

• 人机交互：通过在任何时候检查和修改代理状态，融入人工监督。

• 综合记忆： 创建具有短期工作记忆以进行持续推理和跨会话长时记忆的有状态代理。

• 使用LangSmith调试：通过可视化工具深入了解复杂代理行为，追踪执行路径、捕获状态转移，并提供详细的运行时指标。

• 简易部署：借助可扩展的基础架构，自信地部署复杂的代理系统，该基础架构旨在应对有状态、长时间运行的工作流所面临的独特挑战。

Langgraph与langchain区别

首先，LangGraph和LangChain是两个相关但不同的工具，都来自LangChain生态系统

LangChain

LangChain是一个用于构建大语言模型应用程序的框架

• 线性工作流：主要支持顺序执行的链式操作

• 组件库：提供丰富的预构建组件，如提示模板、向量存储、检索器等

• 简单集成：易于快速原型开发和简单的LLM应用

• 抽象层：为不同的LLM提供统一接口

• 中间件：提供十多种预构建中间件和自定义中间件去进行Agent执行中的拦截

LangGraph

LangGraph是LangChain团队开发的更高级工具，专门用于构建复杂的多智能体系统：

• 图状工作流：支持复杂的分支、循环和条件逻辑

• 状态管理：内置强大的状态管理机制

• 多智能体协作：原生支持多个AI智能体之间的交互

• 复杂决策流：可以根据条件动态选择执行路径

• 持久化：支持长时间运行的工作流和状态持久化

总而言之，Langchain的工作流程很简单，就是用户Query——向量检索——模型生成——格式化输出，对于那些流程明确的工作（比如文本翻译、问答），就适合langchain框架。

而langgraph更适合解决复杂的任务，langgraph原生支持持久化不需要像langchain一样专门存储历史记录，langgraph支持循环结构，允许模型在犯错的时候从上一步重新开始，同时langgraph支持多智能体协作，在一些重要决策前加入人工判断的步骤。

langgraph的基础知识

langgraph安装

 pip install -U "langgraph"

核心基础

State（状态）	整个工作流中被传输和存储的数据的数据结构
Node（结点）	工作流中执行的一个步骤（调用函数、工具）
Edge（边）	控制流程下一步应该去哪个结点
Graph（流程图）	将所有边和结点组织成一张流程图

LangGraph 的核心是将代理工作流程建模为图表，它有三个核心组件：

• State：通常是一个 TypedDict 或 Pydantic的BaseModel，表示当前流程共享的数据结构

• Node：接收当前值state作为输入，执行计算并返回一个新的state

• Edge：根据当前条件决定下一步执行哪个Node

通过组合Node和Edge，您可以创建复杂的循环工作流，使其state随时间推移而演化。然而，真正的强大之处在于 LangGraph 对 的管理发送state

状态图Stategraph

Stategraph是主要使用的python图形类，这个类提供了很多方法：

• 向图中添加执行单元（节点）

• 在节点之间连线

• 添加条件分支

• 指定图的起点

• 验证图是否正确(是否有孤立结点)

为什么叫“状态图”而不是“流程图”

LangGraph 不只是流程控制，还强调：

• 每个节点执行前、执行后都可以访问和修改状态（state）

• 状态在节点之间流动

• 节点的跳转可以依据状态来判断

from typing import TypedDict
from langgraph.graph import StateGraph

# 定义状态结构
class MyState(TypedDict):
    question: str
    answer: str

# 定义节点函数
def search_node(state):
    return {"answer": "这是答案"}

# 创建状态图
builder = StateGraph(state_schema=MyState)
# 添加一个节点
builder.add_node("search", search_node)
# 第一个要调用的节点
builder.set_entry_point("search")
# 要构建图，首先要定义状态，然后添加节点和边，最后进行编译，会进行基本的检查
graph = builder.compile()

# 执行图
result = graph.invoke({"question": "什么是状态图？"})
print(result["answer"])  # 输出：这是答案

状态State

在使用 LangGraph 构建流程图之前，第一件事就是定义图的状态 State。这是整个图运行中用于共享和传递信息的核心机制。

LangGraph 中的 State 是图中所有节点（Node）之间传递数据的模式结构，可以类比为一个共享的上下文字典，它包含输入、输出、中间变量等

定义 State 时，需要包含两个部分：

Schema（模式）：指定 State 的字段结构（一般用TypedDict或Pydanic）

# langgraph推荐使用TypedDict
"""
1. TypedDict 是标准库的一部分（来自 typing 模块），零依赖，零性能开销而 Pydantic 会在每一步创建模型实例，会增加运行时负担
2. LangGraph 中的 State 实质就是一个字典（dict），而 TypedDict 就是“有类型注解的 dict”，与 LangGraph 的执行机制无缝对接，而 Pydantic 是类结构，需要 .dict() 转换，略显多余
"""
from typing import TypedDict


class State1(TypedDict):
    user_input: str

# 使用 pydantic 可以进行参数校验和提供默认值
from pydantic import BaseModel


class State2(BaseModel):
    question: str
    result: str = ""

多个模式（Multiple Schemas）：在大多数情况下，LangGraph 使用一个统一的 State 模式。但你也可以设置“输入模式”和“输出模式”分开

1. 输入模式：接收用户输入的字段（如question）

2. 输出模式：只保留最终输出的字段（如final_answer）
    

   from typing import TypedDict
   from langgraph.graph import StateGraph
   
   
   # 1. 定义输入、输出、图内部的状态结构
   
   # 输入字段：用户的问题
   class InputState(TypedDict):
       question: str
   
   
   # 中间状态：包括中间结果
   class InternalState(TypedDict):
       question: str
       search_result: str
       final_answer: str
   
   
   # 输出字段：只想返回最终答案
   class OutputState(TypedDict):
       final_answer: str
   
   
   # 2. 定义节点函数（中间节点用中间字段）
   def search_node(state: InternalState) -> dict:
       return {"search_result": f"搜索了：{state['question']}"}
   
   
   def answer_node(state: InternalState) -> dict:
       return {"final_answer": f"根据搜索结果：{state['search_result']}，这是答案"}
   
   
   # 3. 创建 StateGraph，显式指定输入/输出 Schema
   builder = StateGraph(state_schema=InternalState, 
                        input_schema=InputState, 
                        output_schema=OutputState)
   
   # 4. 添加节点
   builder.add_node("search", search_node)
   builder.add_node("answer", answer_node)
   
   # 5. 配置流程
   builder.set_entry_point("search")
   builder.add_edge("search", "answer")
   
   # 6. 编译并执行图
   app = builder.compile()
   
   result = app.invoke({"question": "什么是LangGraph？"})
   
   print(result)  # {'final_answer': '根据搜索结果：搜索了：什么是LangGraph？，这是答案'}

   -> dict：返回值注解，表示该函数应该返回一个 dict 类型的对象

3. Reducer（归并函数）：在 LangGraph 中，所有节点返回的都是“局部更新结果”，Reducer 是用于合并多个节点输出更新的机制。 将每个节点返回的“局部状态更新”统一合并进全局的 State

   在TypeDict 中，我们通过Annotated来给某个字段指定 Reducer 函数

Annotated[类型, Reducer函数] ：这里的 add 表示：每次节点返回这个字段时，都把新列表拼接到旧列表后面

1. from typing import Annotated 
   from typing_extensions import TypedDict 
   from operator import add 
   
   class State(TypedDict):     
       foo: int     
       bar: Annotated[list[str], add] # 每条消息是 {role, content}，会自动追加到列表末尾

   在长任务中，state会随着任务轮次的增加变得臃肿，容易导致解析变慢或者上下文溢出，为了解决这种现象，需要引入reducer机制

在图表中使用消息

在许多情况下，将之前的对话历史记录以消息列表的形式存储在图状态中会很有帮助。为此，我们可以向图状态添加一个键（通道），该键存储Message对象列表，并使用 Reducer 函数对其进行注释。Reducer 函数对于指示图如何Message在每次状态更新（例如，当节点发送更新时）时更新状态中的对象列表至关重要。如果您未指定 Reducer，则每次状态更新都会用最新提供的值覆盖消息列表。如果您只想将消息附加到现有列表中，可以使用operator。

from typing import TypedDict
from langgraph.graph import StateGraph
from typing import Annotated
import operator


# 定义状态结构   如果定义的是list[dict]，会覆盖之前的数据
class ChatState(TypedDict):
    messages: Annotated[list, operator.add]  # 每条消息是 {role, content}，会自动追加到列表末尾


# 节点函数：添加用户问题
def user_input_node(state: ChatState) -> dict:
    user_msg = {"role": "user", "content": "什么是LangGraph？"}
    return {"messages": [user_msg]}


# 节点函数：添加助手回复
def assistant_node(state: ChatState) -> dict:
    reply = {"role": "assistant", "content": "LangGraph 是一个有状态的图编排框架。"}
    return {"messages": [reply]}


# 构建状态图
builder = StateGraph(state_schema=ChatState)
builder.add_node("user_input", user_input_node)
builder.add_node("assistant_reply", assistant_node)
builder.set_entry_point("user_input")
builder.add_edge("user_input", "assistant_reply")

graph = builder.compile()

result = graph.invoke({"messages": []})
print(result["messages"])

MessagesState

MessagesState是langgraph中自带的状态模板，它能够自动存储对话历史并自动追加消息，不需要写typedict和reducer

from langgraph.graph import MessagesState

# 和上述代码不同会在State类中自动维护一个messages 字段，不需要显示创建
class State(MessagesState):
    documents: list[str]

节点Nodes

节点（Nodes）是图中执行逻辑的基本单位。每个节点表示一个函数步骤、处理阶段或子逻辑流程，多个节点通过边连接成有向图，组成一个完整的有状态计算流程。

LangGraph 中的节点就是你定义的一个函数（或 Runnable 对象），用于接收状态、执行逻辑，并返回更新后的状态

def my_node(state: dict) -> dict:
    # 处理输入状态，并返回更新字段
    return {"new_key": "new_value"}
    
# LangGraph 会自动用 reducer 把这些更新合并进全局状态。

START节点

Node START是一个特殊节点，表示将用户输入发送到图的节点。引用此节点的主要目的是确定应首先调用哪些节点。

from langgraph.graph import START

graph.add_edge(START, "node_a")

add_edge负责把两个节点连接起来：workflow.add_edge(上一个节点, 下一个节点)

End节点

Node End是一个特殊节点，表示终端节点。当需要指示哪些边在完成后没有操作时，可以引用此节点。

from langgraph.graph import END

graph.add_edge("node_a", END)

并行运行节点

并行运行节点指的是同时启动多个不互相依赖的节点，而不是按顺序一个接一个地执行。

在langgraph中实现并行只需要将同一个起点的边指向多个不同的节点即可

# 当 start_node 运行结束时，A 和 B 会同时启动
workflow.add_edge("start_node", "node_A")
workflow.add_edge("start_node", "node_B")

边Edge

Edge（边）是连接节点的通道，表示图中节点之间的执行跳转关系。你可以把它理解为「节点执行完之后，下一步去哪，是构成 LangGraph 流程图的核心

普通边

graph.add_edge("节点A", "节点B")

条件边

条件边会根据state里的内容，判断下一步执行哪一个逻辑函数

def route_condition(state: MyState):
    """条件函数：只负责路由决策"""
    if state["type"] == "a":
        return "a"
    elif state["type"] == "b":
        return "b"
    else:
        return "default"

def node_a(state):
    return {"result": "走了 A 分支"}


def node_b(state):
    return {"result": "走了 B 分支"}


def node_default(state):
    return {"result": "走了默认分支"}

graph.add_conditional_edges("judge_node", route_condition, {
    "a": "a",
    "b": "b",
    "default": "default"
})

条件边的格式

workflow.add_conditional_edges(
    "起点节点", 
    routing_function,      # 判断逻辑函数
    {"path_a": "节点A", "path_b": "节点B"} # 路由映射表
)

条件入口点

根据初始输入的数据，动态决定首先运行哪个节点

from langgraph.graph import StateGraph, START, END

# 1. 定义路由函数
def route_start(state: State):
    if state["question"].startswith("查资料"):
        return "search_node"
    else:
        return "chat_node"

# 2. 构建图
builder = StateGraph(State)
builder.add_node("search_node", search_func)
builder.add_node("chat_node", chat_func)

# 3. 设置条件入口点：从 START 开始进行条件跳转
builder.add_conditional_edges(
    START,           # 起点是内置的 START
    route_start,     # 判断函数
    {
        "search_node": "search_node", 
        "chat_node": "chat_node"
    }
)

发送Send

默认情况下，Nodes和Edges是提前定义的，并在相同的共享状态下运行。但是，在某些情况下，确切的边无法提前知道，并且可能希望同时存在State的不同版本。

也就是说，普通的边只能沿着图预设的路线走；而send可以强制跳到任意节点并传输数据，支持运行时动态地分发任务

• 条件路由：根据某些条件将消息发送到不同的节点（可以直接用条件边实现，没必要用send）

• 并行处理：同时向多个节点发送消息 （核心）

• 动态工作流：根据运行时状态决定消息的发送目标

Map-Reduce

Map-Reduce就是send的工作原理，它是一种经典的并行计算模式，特别适合处理大规模数据

Map-Reduce 将复杂的数据处理任务分解为两个阶段：

• Map（分发）：Send将主状态中的一个列表拆开，并行启动多个目标节点。每个节点只拿到属于它的那部分数据。

• Process（处理）：这些节点同时运行，互不干扰。

• Reduce（聚合）：当所有并行的节点都跑完后，将所有小任务的结果合并成最终结果

标准结构是  Send(目标节点名称, 要传递的状态数据)

Send("node_name", {"key": value})

"""
LangGraph Map-Reduce 简单案例：数字求和
把一堆数字分给多个worker算平方，然后把结果加起来
"""
from typing import Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.constants import Send
from typing import TypedDict, List

# 状态定义
class State(TypedDict):
    numbers: List[int]        # 输入的数字
    results: Annotated[list[int], operator.add]        # worker的结果
    final_sum: int           # 最终求和

# 1. Map阶段：分发数字
def split_numbers(state: State):
    """把数字分发给不同的worker"""
    numbers = state["numbers"]
    print(f"分发数字: {numbers}")

    # 每个数字发给一个worker
    return [Send("worker", {"number": num}) for num in numbers]

# 2. Worker阶段：计算平方
def calculate_square(state: State):
    """每个worker计算一个数字的平方"""
    number = state["number"]
    square = number * number
    print(f"Worker: {number}² = {square}")
    return {"results": [square]}

# 3. Reduce阶段：求和
def sum_results(state: State):
    """把所有结果加起来"""
    results = state.get("results", [])
    total = sum(results)
    print(f"求和: {results} = {total}")
    return {"final_sum": total}

# 构建图
def create_simple_graph():
    graph = StateGraph(State)

    # 添加节点
    graph.add_node("splitter", lambda s: s)  # 分发器
    graph.add_node("worker", calculate_square)  # 工作节点
    graph.add_node("summer", sum_results)      # 求和器

    # 连接节点
    graph.add_edge(START, "splitter")
    graph.add_conditional_edges("splitter", split_numbers, ["worker"])  # Map阶段
    graph.add_edge("worker", "summer")  # Worker完成后求和
    graph.add_edge("summer", END)

注意这里，graph.add_node("worker",calculate_square)

第一个字符串参数是节点在图中的标识符

第二个参数是具体要执行的函数

这里解释两个点

• 分发器：graph.add_node("splitter", lambda s: s)；这种节点经常用作数据分流（路由）的起点，lambda是python中的一个匿名函数，lambda s: s说明这个函数传入什么就输出什么；这个看起来什么也不干的“splitter”节点就是负责接收数据然后分发给每个worker节点。

• Map分发过程：graph.add_conditional_edges("splitter", split_numbers, ["worker"])，这里Send已经指定了要去的节点因此不需要路由映射表；split_numbers函数返回了n个Send，LangGraph 看到这些 Send，就知道要并行启动n个worker节点。

命令Command

Command 是 LangGraph 中用于控制图执行流程、更新图状态，并支持人机交互、工具调用的标准化对象。

核心作用：

  第一，更新图的运行状态；

  第二，控制图的执行流向（指定下一个或多个执行节点）；

  第三，衔接中断恢复、工具调用、人机交互等场景。

command参数

update：应用状态更新（类似于从节点返回更新）。

goto：导航到特定节点（类似于条件边，以后无需再stategraph里连线）。

graph：在从子图导航时定位到父图。

resume：在中断后提供一个值以继续执行。

在节点函数中返回command时，必须添加返回类型注释，其中包含节点路由到的节点名称列表，例如command[Literal ["my_other_node"]]。这对于图形渲染是必需的，它告诉 LangGraph 当前节点可以导航到my_other_node。

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.types import Command, Send
from typing import Literal


class MyState(TypedDict):
    type: str
    text: str
    result: str


def judge_node(state: MyState) -> Command[Literal["a", "b", "default"]]:
    """条件函数：使用Command进行路由和状态更新"""
    if state["type"] == "a":
        return Command(update={"text": "走了 A 分支"}, goto="a")
    elif state["type"] == "b":
        return Command(update={"text": "走了 B 分支"}, goto="b")
    else:
        return Command(update={"text": "走了默认分支"}, goto="default")


def node_a(state):
    return {"result": f"A节点处理: {state['text']}"}


def node_b(state):
    return {"result": f"B节点处理: {state['text']}"}


def node_default(state):
    return {"result": f"默认节点处理: {state['text']}"}


# 构建图
graph = StateGraph(state_schema=MyState)
graph.add_node("judge_node", judge_node)
graph.add_node("a", node_a)
graph.add_node("b", node_b)
graph.add_node("default", node_default)

graph.set_entry_point("judge_node")

# 添加结束边
graph.add_edge("a", END)
graph.add_edge("b", END)
graph.add_edge("default", END)

app = graph.compile()

# 测试
print("测试 A:", app.invoke({"type": "a", "text": "", "result": ""}))
print("测试 B:", app.invoke({"type": "b", "text": "", "result": ""}))
print("测试其他:", app.invoke({"type": "default", "text": "", "result": ""}))

在这里，我们使用了Command对象过后就不需要写builder.add_node和builder.add_conditional_edges函数了

什么时候应该使用Command而不是条件边

如果你想在节点内部既要改状态state，又要实现跳转，那么就必须使用Command，在条件边中跳转和状态修改分散在两个函数里面，而command可以把这两个逻辑聚在一起。