https://blog.langchain.com/langgraph/

一、介绍

在我们 LangChain v0.1 的发布公告中，我们重点介绍了一款新库：LangGraph。
LangGraph 构建在 LangChain 之上，并且与整个 LangChain 生态系统完全互操作。
它的主要价值在于引入了一种简单的方法来创建 循环图（cyclical graphs）。这在构建 agent 运行时（agent runtimes） 时常常非常有用。

二、设计动机

LangChain 的一个重要价值主张是能够轻松创建 自定义链（custom chains）。我们已经在 LangChain 表达式语言（LangChain Expression Language, LCEL） 上投入了大量功能开发。
但到目前为止，我们仍然缺乏一种方法，能够轻松地将 循环（cycles） 引入到这些链中。实际上，这些链都是 有向无环图（DAGs, Directed Acyclic Graphs） —— 这也是大多数数据编排框架的形态。

当人们创建更复杂的 LLM 应用 时，我们常见的一种模式就是在运行时引入循环。这些循环通常利用 LLM 来推理 下一步应该做什么。
LLM 的一大突破性价值就是能够胜任这种推理任务。
换句话说，这可以被视为在 for 循环 中运行一个 LLM。这类系统通常被称为 agents（智能体）。

为什么这种 agent 行为 会如此强大？

可以通过典型的 RAG（检索增强生成，Retrieval Augmented Generation） 应用来说明。
在一个典型的 RAG 应用中，我们会调用一个检索器，它返回一些文档，然后将这些文档传递给 LLM 来生成最终答案。
虽然这种方式通常有效，但在某些情况下会失效 —— 比如 首次检索步骤未能返回有用结果。
在这种情况下，理想的做法是让 LLM 能够推理出：检索器返回的结果质量很差，接着再发起一次（更优化的）检索请求，并使用新的结果。
换句话说，将 LLM 运行在一个循环中，可以帮助创建更灵活的应用，从而能够处理一些模糊的、非预定义的用例。

这些类型的应用通常被称为 agents。最简单 —— 但同时也最具野心 —— 的形式是一个基本只有两步的循环：

  1. 调用 LLM 来决定：（a）要采取什么动作，或者（b）要给用户什么回复
  2. 执行上述动作，并回到步骤 1

这个循环会不断重复，直到最终生成一个响应。它正是我们核心 AgentExecutor 的底层逻辑，也是 AutoGPT 等项目流行的原因。

  • 它之所以“简单”，是因为循环结构相对直观；
  • 它之所以“雄心勃勃”，则是因为它几乎将 所有的决策与推理能力 都完全交给了 LLM。

在与社区和企业合作，将 agent 投入生产的过程中，我们发现：实际应用中往往需要 更多的控制。例如：

  1. 你可能希望强制 agent 总是先调用某个特定工具；
  2. 你可能希望对 工具调用方式 拥有更多控制；
  3. 你可能希望根据 agent 的 不同状态，为它设计不同的提示词（prompt）。

当我们讨论这些 更受控的流程 时，内部通常称之为 “状态机（state machines）”。下面的图（摘自我们关于认知架构的博客）展示了这种思路：
在这里插入图片描述
这些 状态机（state machines） 具备循环的能力 —— 这使得它们能够处理比简单链更模糊的输入。
然而，在循环的构建方式上，仍然需要一定的人为引导。

LangGraph 提供了一种方式，可以通过将它们定义为 图（graphs） 来创建这些状态机。

三、核心功能

1. StateGraph

LangGraph 的核心接口很简单：StateGraph 表示一个图。它依赖你传入的 state 类型定义，这个状态对象将在执行过程中不断被节点更新——节点返回一个 key-value 结构来更改状态。

状态字段更新方式支持两种模式：
• 覆盖（override）：节点返回新值直接覆盖旧值；
• 累加（add）：对列表等可累积内容追加。

示例代码：

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

class State(TypedDict):
    input: str
    all_actions: Annotated[List[str], operator.add]

graph = StateGraph(State)

2. 节点（Nodes）

使用 graph.add_node("model", model_fn) 添加节点，每个节点是一个功能步骤，比如执行 LLM 或工具。节点函数接受 state 并返回状态更新。

特殊节点

from langgraph.graph import START,END

3. 边（Edges）

The Starting Edge

graph.set_entry_point("node01")

#或者 用START 指向 入口节点node01
from langgraph.graph import START
graph_builder.add_edge(START, "node01")

Normal Edges

graph.add_edge("node01", "node02") # node01 -> node02

Conditional Edges

graph.add_conditional_edges(
    source: str,
    path: Callable[[State], str],   # 决定下一个节点的函数
    path_map: Dict[str, str],       # {条件值 -> 节点}
    default: Optional[str] = None   # (可选) fallback 节点
)

编译&运行

app = graph.compile()

完整案例

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, List, Annotated
import operator

MAX_RECURSION_LIMIT = 10  # 最大递归次数
global_num: int = 0

# 定义状态
class State(TypedDict):
    input: str
    all_actions: Annotated[List[str], operator.add]  # 自动累积所有动作


# 定义节点
def node_01_start(state: State) -> State:
    global global_num
    global_num += 1
    result = {"all_actions": [f"node_01:{global_num}"]}
    return result

def node_021(state: State) -> State:
    global global_num
    global_num += 1
    result = {"all_actions": [f"node_021:{global_num}"]}
    return result

def node_022(state: State) -> State:
    global global_num
    global_num += 1
    result = {"all_actions": [f"node_022:{global_num}"]}
    return result

def node_03_end(state: State) -> State:
    global global_num
    global_num += 1
    result = {"all_actions": [f"node_03:{global_num}"]}
    return result

def node_fallback(state: State) -> dict:
    global global_num
    global_num += 1
    return {"all_actions": [f"node_fallback:{global_num}"]}

# 构建 graph
graph_builder = StateGraph(State)

graph_builder.add_node("node_01", node_01_start)
graph_builder.add_node("node_021", node_021)
graph_builder.add_node("node_022", node_022)
graph_builder.add_node("node_03", node_03_end)
graph_builder.add_node("node_fallback", node_fallback)

# 两个工具都执行，然后进入 model
graph_builder.add_edge("node_01", "node_021")
graph_builder.add_edge("node_01", "node_022")
graph_builder.add_edge("node_021", "node_03")
graph_builder.add_edge("node_022", "node_03")

def should_continue(state: State) -> str:
    if "hello" in state["input"]:
        return "end"
    else:
        # 当执行到一定次数时, 跳转至: node_fallback
        if len(state['all_actions']) == MAX_RECURSION_LIMIT-2:
            return "node_fallback" 
        return "node_01"

graph_builder.add_conditional_edges(source="node_03", path=should_continue,path_map={
    "end": END,
    "node_01": "node_01",
    "node_fallback": "node_fallback"
})

# 定义从哪个节点开始开始
# 以下两种方式等价
#graph_builder.set_entry_point("node_01")  
graph_builder.add_edge(START, "node_01")

graph = graph_builder.compile()

打印流程图: display(Image(graph.get_graph().draw_mermaid_png()))
在这里插入图片描述

执行1

# 运行1
global_num: int = 0
result = graph.invoke({"input": "hello", "all_actions": []}, config={"recursion_limit": MAX_RECURSION_LIMIT}) # 增加递归限制为5

print("\n=== Final State ===")
print(result)

输出

=== Final State ===
{'input': 'hello', 'all_actions': ['node_01:1', 'node_021:2', 'node_022:3', 'node_03:4']}

执行2: recursion_limit

global_num: int = 0
result = graph.invoke({"input": "hi", "all_actions": []}, config={"recursion_limit": MAX_RECURSION_LIMIT}) # 增加递归限制为5

print("\n=== Final State ===")
print(result)

输出

=== Final State ===
{'input': 'hi', 'all_actions': ['node_01:1', 'node_021:2', 'node_022:3', 'node_03:4', 'node_01:5', 'node_021:7', 'node_022:6', 'node_03:8', 'node_fallback:9']}

Agent Executor

我们使用 LangGraph 重建了经典的 LangChain AgentExecutor。这使您可以继续使用现有的 LangChain 代理，但同时更轻松地修改 AgentExecutor 的内部实现。
默认情况下，此图的状态包含了您在使用 LangChain 代理时应该熟悉的概念：input、chat_history、intermediate_steps，以及用于表示最近一次代理结果的 agent_outcome。

from typing import TypedDict, Annotated, List, Union
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.messages import BaseMessage
import operator


class AgentState(TypedDict):
   input: str
   chat_history: list[BaseMessage]
   agent_outcome: Union[AgentAction, AgentFinish, None]
   intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]
  • input: 表示传给代理的输入文本，例如用户的问题或指令。
  • chat_history 保存代理的对话历史.每条消息是 BaseMessage 类型（通常包括消息内容、发送者等信息）。
  • agent_outcome 表示代理的最新动作或结果。
    • AgentAction：表示Function call
    • AgentFinish：代理完成整个任务的结果。
    • None：尚未产生动作或结果。
    • 使用 Union 表示可能是三者之一。
  • intermediate_steps 保存代理的中间步骤历史。类型是一个列表，列表中每个元素是一个 (AgentAction, str) 元组
    • AgentAction：代理采取的动作。
    • str：动作的输出文本或反馈。

案例:agent_v6.py

使用langraph重写案例： agent_v3

Chat Agent Executor

我们观察到的一个普遍趋势是：越来越多的模型是 “聊天（chat）”模型，它们以消息列表作为输入进行运作。
这些模型通常配备了诸如 函数调用（function calling） 之类的功能，使得类似代理（agent）的体验更加可行。

我们创建了一个能基于这种状态运行的代理运行时（agent runtime）。
它的输入是一个消息列表，而各个节点只需随着时间推移不断往这个消息列表中添加内容即可。

from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage


class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

Modifications（修改与定制）

LangGraph 允许你对 agent 运行时进行多种定制，以适应不同的业务需求或工作流。以下是几个常见的修改示例：

1. 强制先调用某个工具（Force Calling a Tool First）

有时你希望 agent 在执行其他操作之前 总是先调用某个特定工具。
在传统 LangChain 中，这种控制需要额外的逻辑实现，而在 LangGraph 中，你可以通过 图的结构和条件边（conditional edges）轻松实现这一点。

2. 控制工具调用方式（Controlling How Tools Are Called）

你可能希望对 工具的调用方式 有更多控制，例如：
• 调用顺序
• 调用参数
• 调用次数

在 LangGraph 中，每个节点代表一个步骤，你可以通过边的条件和节点的逻辑来灵活控制工具调用。

3. 根据状态使用不同提示词（Different Prompts Based on State）

在运行过程中，agent 的 状态可能不同，你可能希望：
• 根据状态使用不同的提示词（prompt）
• 调整模型的输出策略

LangGraph 可以将这些状态表示为节点的状态或图的属性，从而动态选择不同提示。

4. 动态直接返回工具输出（Dynamically Returning Tool Output）

有时你希望工具输出直接返回，而不是经过 agent 决策。
LangGraph 支持这种灵活性，可以让 LLM 根据上下文决定是否直接返回输出（仅适用于 Chat Agent Executor）。

Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

AIGC降重工具排行榜背后的评测标准解读

深度解读AIGC降重工具排行榜的评测标准体系，揭示排名背后的关键指标，帮助用户理性选择适合自己的降AI工具。

avatar 2048 AI社区
cover

14k stars！用AI自动赚钱？这个开源项目让你的副业全程“无人值守“！

avatar 2048 AI社区

不同论文降重降AI方案成本对比分析：花最少的钱办最大的事

全面分析不同论文降重降AI方案的成本差异，包含免费方案、付费工具、人工代改等多种选择，帮你找到最省钱又有效的方案。

avatar 2048 AI社区