目录

核心思想

从 LangChain 的局限说起

LangGraph 的解决方案

四个核心概念

1. State(状态)

2. Node(节点)

3. Edge(边)

4. Graph(图)

实际例子:带循环的检索

LangGraph vs LangChain

运行机制

为什么用图结构

总结


核心思想

LangGraph 是一个用图结构来编排 Agent 执行的框架。

为什么需要图结构?因为传统的 Agent 执行流程是黑盒——模型自己决定先做什么、后做什么。当你需要精确控制流程时(比如"先检索文档,如果不够就再检索一次,然后生成答案"),黑盒就不够用了。

LangGraph 把 Agent 拆分成节点(Node)和边(Edge),让你可以像画流程图一样设计 Agent 的执行逻辑。

从 LangChain 的局限说起

LangChain 的 Agent 工作方式是:

while True:
    # 1. 调用模型
    response = model.invoke(messages)
    
    # 2. 模型决定是否调用工具
    if response.tool_calls:
        tool_result = execute_tool(response.tool_calls[0])
        messages.append(tool_result)
    else:
        # 3. 模型说结束就结束
        break

这个模型的问题在于:流程完全由模型控制

你想实现这样的逻辑:

  • 先检索文档
  • 如果检索结果少于 3 篇,换个关键词再检索一次
  • 然后生成答案
  • 最后检查答案质量,如果不好就重新生成

这在 LangChain 里很难实现,因为你没法在流程中设置明确的检查点和分支逻辑。

LangGraph 的解决方案

LangGraph 让你显式定义流程:

START → 检索节点 → 判断节点 → 生成节点 → END
                     ↑          │
                     └──────────┘

每个方框是一个节点(Node),箭头是边(Edge)。节点执行具体操作,边决定执行顺序。

四个核心概念

1. State(状态)

State 是整个图共享的数据容器。所有节点都能读取和修改 State。

from typing import TypedDict

class State(TypedDict):
    question: str      # 用户问题
    documents: list    # 检索到的文档
    answer: str        # 生成的答案

为什么需要 State?

因为节点之间需要传递数据。检索节点检索到的文档,需要传递给生成节点使用。State 就是这个数据传递的媒介。

设计原则:

  • 只放跨节点共享的数据
  • 字段名要清晰表达含义
  • 不要放临时变量(临时变量应该在节点内部处理)

2. Node(节点)

节点就是一个 Python 函数,接收当前 State,返回 State 的更新。

def retrieve_node(state: State) -> dict:
    # 从 State 读取问题
    question = state["question"]
    
    # 执行检索
    docs = search_database(question)
    
    # 返回更新(只返回变化的字段)
    return {"documents": docs}

关键点:

  • 节点函数接收完整的 State
  • 返回的是 State 的更新(dict),不是完整 State
  • LangGraph 会自动合并多个节点的更新

3. Edge(边)

边决定节点的执行顺序。有两种类型:

普通边: 固定跳转到下一个节点

graph.add_edge("retrieve_node", "generate_node")

条件边: 根据条件选择下一个节点

def should_continue(state: State) -> str:
    # 根据文档数量决定
    if len(state["documents"]) < 3:
        return "retrieve_again"  # 再去检索
    else:
        return "generate"         # 直接生成

graph.add_conditional_edges(
    "retrieve_node",      # 从哪个节点
    should_continue,      # 条件函数
    {
        "retrieve_again": "retrieve_node",
        "generate": "generate_node",
    }
)

4. Graph(图)

图就是节点和边的集合。定义好图之后,编译成可执行的应用。

from langgraph.graph import StateGraph, START, END

# 1. 创建图
graph = StateGraph(State)

# 2. 添加节点
graph.add_node("retrieve", retrieve_node)
graph.add_node("generate", generate_node)

# 3. 设置入口
graph.set_entry_point("retrieve")

# 4. 添加边
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)

# 5. 编译
app = graph.compile()

# 6. 执行
result = app.invoke({"question": "什么是 LangGraph?"})

实际例子:带循环的检索

假设我们要实现:检索文档,如果不够就再检索一次。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    question: str
    documents: list
    attempts: int  # 检索次数


def retrieve(state: State) -> dict:
    """检索文档"""
    docs = search(state["question"])
    return {
        "documents": docs,
        "attempts": state.get("attempts", 0) + 1
    }


def check_docs(state: State) -> str:
    """检查文档是否足够"""
    if len(state["documents"]) < 3 and state["attempts"] < 2:
        return "need_more"
    else:
        return "enough"


def generate(state: State) -> dict:
    """生成答案"""
    answer = llm.generate(state["documents"])
    return {"answer": answer}


# 组装图
graph = StateGraph(State)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)

graph.set_entry_point("retrieve")
graph.add_conditional_edges(
    "retrieve",
    check_docs,
    {
        "need_more": "retrieve",  # 循环
        "enough": "generate",
    }
)
graph.add_edge("generate", END)

app = graph.compile()
result = app.invoke({"question": "什么是 LangGraph?"})

执行流程:

  1. retrieve 节点开始,检索文档
  2. 执行 check_docs,如果文档少于 3 篇且检索次数少于 2 次,回到 retrieve
  3. 否则执行 generate 生成答案
  4. 结束

LangGraph vs LangChain

需求

LangChain

LangGraph

简单问答机器人

适合

不太适合

快速原型验证

适合

配置稍多

复杂的流程控制

困难

原生支持

需要人工审批

需要自己实现

内置支持

多 Agent 协作

困难

轻松表达

状态持久化

需要自己实现

内置支持

选择建议:

如果你的应用只是简单的"用户提问 → Agent 回答",用 LangChain 就够了。

如果你需要:

  • 精确控制执行流程(分支、循环、并行)
  • 在关键节点暂停等待人工审批
  • 多个 Agent 协作完成复杂任务
  • 保存和恢复执行状态

那么 LangGraph 更合适。

运行机制

LangGraph 的执行分为两个阶段:

阶段 1:定义图结构

  • 定义 State(数据结构)
  • 定义 Node(执行逻辑)
  • 定义 Edge(流程控制)

阶段 2:编译执行

  • compile() 把图编译成可执行的应用
  • invoke() 执行应用

编译的时候,LangGraph 会:

  1. 验证图结构是否完整(没有孤立节点)
  2. 优化执行计划
  3. 配置运行时参数(如 Checkpointer)

为什么用图结构

图结构有几个关键优势:

1. 可视化

图可以画出来,方便理解和调试。

START → A → B → C → END
           ↑    │
           └────┘

一看就知道:执行完 B 到 C,C 可能回到 B 形成循环。

2. 并发执行

如果节点之间没有依赖关系,可以并行执行。

       ┌→ A ─┐
START ─┼→ B ─┼→ END
       └→ C ─┘

A、B、C 可以同时执行,最后汇总结果。

3. 状态管理

State 在节点之间传递和更新,天然支持状态追踪和持久化。

4. 流程控制

分支、循环、并行都是图的基本操作,不需要额外实现。

总结

LangGraph 的核心价值:把 Agent 的执行流程从黑盒变成明盒。

  • State:数据容器,节点之间传递信息
  • Node:执行单元,处理具体逻辑
  • Edge:流程控制,决定执行顺序
  • Graph:完整应用,编译后执行
Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐