LangGraph入门：构建智能化工作流的初学者指南

LangGraph作为一种强大的框架，能够编排复杂的工作流程，它允许你定义一系列的操作（节点）以及它们之间的连接方式（边），从而构建一个完整的应用程序。特别是在管理各种组件之间的交互时，例如大型语言模型（LLMs）、外部API和自定义函数，LangGraph表现得尤为出色。本文将带你一步步了解如何使用LangGraph构建智能工作流。

1. LangGraph核心概念：智能体的蓝图

在开始构建之前，我们需要理解构成任何LangGraph应用程序的基本组件。LangGraph的强大之处在于它将复杂任务分解为一系列可管理和可扩展的步骤。它的核心概念就像智能体的蓝图，让开发者能够清晰地定义和控制AI的行为。

图（Graph）: 将其视为AI智能体的总规划。它是应用程序工作流的中心蓝图，是由相互连接的节点和边组成的集合。你可以想象一下流程图，其中每个步骤都是一个节点，箭头表示这些步骤之间的流程。在实际应用中，比如一个客户服务机器人，图可以定义从接收用户查询到提供解决方案的完整路径。
节点（Nodes）: 这些是工作流中的“执行者”或“步骤”。每个节点执行一个特定的操作或代表一个AI智能体。节点非常灵活，可以是：
- LLM调用。例如，调用GPT-4来生成一段文本摘要。
- 工具调用（例如，搜索API、计算器）。比如，使用Tavily Search API来获取最新的新闻信息。
- 自定义Python函数。例如，一个用于数据清洗的函数。
- 另一个LangChain runnable。例如，将多个LangChain链组合成一个更大的工作流。
- 实际案例: 在一个电商推荐系统中，一个节点可能负责从用户历史行为中提取特征，另一个节点则利用这些特征向用户推荐商品。
边（Edges）: 这些是连接节点的路径，定义了工作流的进展方式。在一个节点完成其操作后，边决定下一个要执行的节点。LangGraph支持两种主要类型：
- 条件边（Conditional Edges）: 类似于决策点。基于节点的输出，图可以选择不同的路径（到不同的后续节点）。比如，如果用户询问天气，则跳转到天气查询API节点；如果用户询问产品信息，则跳转到产品数据库查询节点。
- 直接边（Direct Edges）: 这是直接连接；工作流只是从一个节点直接移动到下一个节点。例如，在文本处理流程中，一个节点负责文本清洗，然后通过直接边连接到下一个节点，进行情感分析。
状态（State）: 虽然各个节点独立运行，但实际的AI应用程序需要在整个对话或任务中维护信息。“图状态”是一种在节点之间传递的共享内存。每个节点都可以从该状态读取信息，也可以写入（更新）该状态。我们通常使用Python中的TypedDict来定义此状态。例如，在一个聊天机器人中，状态可以包含用户的对话历史、偏好设置等信息。
- 实际案例：在一个复杂的金融风险评估系统中，状态可能包含客户的财务信息、信用评分和交易历史，这些信息会随着流程的进行而不断更新。
入口点（START）: 此特殊节点标记了图的执行起点。可以将其视为智能体旅程的起点线。
结束点（END）: 此标记表示工作流中特定路径的结论。一旦路径到达END节点，该部分的执行就完成了。

2. 构建你的第一个简单LangGraph图（无头）

让我们从一个基本的图开始，以此来初步了解LangGraph。这将引入“无头”图的概念，其中节点直接接收输入并返回输出，而无需共享的TypedDict状态。这非常适合简单的顺序任务。

示例1：LLM调用和Token计数

假设我们要调用一个LLM，然后计算其响应中的token数量。我们定义两个简单的Python函数：

from langchain_openai import ChatOpenAI

def llm(input):
    model = ChatOpenAI(model='gpt-4o-mini') # 初始化一个LLM
    output = model.invoke(input)
    return output.content # 只返回文本内容

def token_counter(input):
    tokens = str(input).split() # 将文本拆分为单词
    count_no = len(tokens)
    return count_no

现在，我们将使用langgraph.graph.Graph来编排它们：

from langgraph.graph import Graph

workflow = Graph()
workflow.add_node('LLM_Model', llm)      # 添加llm函数作为一个节点
workflow.add_node('Get_Token_Counter', token_counter)         # 添加token_counter作为一个节点
workflow.add_edge('LLM_Model', 'Get_Token_Counter')   # LLM的输出传递给token计数器
workflow.set_entry_point('LLM_Model')    # 从这里开始
workflow.set_finish_point('Get_Token_Counter')  # 在这里结束
app = workflow.compile()                # 将图编译为可运行的应用程序

# 当你调用app时，它会将输入发送到LLM，获取响应，然后计算其token数量。
# for output in app.stream("什么是AI智能体？"):
#     # ... 打印输出 ...

这个简单的设置演示了LangGraph如何将函数连接成一个流程。

本示例的关键要点：

Graph vs. StateGraph：
- Graph()更简单。当一个节点的输出直接成为下一个节点的输入时使用它。没有显式的TypedDict用于所有节点读取和写入的共享状态。
- StateGraph()（在下面的示例中使用）用于更复杂的场景，其中多个节点可能读取/写入共享的、不断发展的状态的不同部分，通常涉及条件逻辑或并行执行。
直接输入/输出流：在Graph()中，workflow.add_edge('LLM_Model', 'Get_Token_Counter')意味着llm函数返回的任何内容都将作为单个位置参数传递给token_counter函数。
set_finish_point()：这明确告诉LangGraph应将哪个节点的输出作为app.invoke()的最终结果返回。

3. 深入探讨：核心组件的详细信息

现在，让我们探索StateGraph以及状态、节点和边如何协同工作以实现更复杂的场景。

状态（State）

定义StateGraph的第一步是定义其状态。此状态模式充当所有节点和边的输入。我们使用TypedDict定义它。

from typing import TypedDict

class State(TypedDict):
    # 在此示例中，我们将拥有一个用于保存字符串状态的单个键
    graph_state: str

每个节点将读取并可能更新此graph_state。

节点（Nodes）

节点表示任何操作或函数。在StateGraph中，节点接收整个当前状态作为其第一个位置参数。它们的返回值用于更新状态。默认情况下，如果节点返回{'key': value}，它将覆盖状态中的现有键。

在这里，我们定义了三个函数，它们将被添加为StateGraph中的节点。

def node_1(state):
    print('--Node1--')
    # 追加到graph_state
    return {'graph_state': state['graph_state'] + "我是"}

def node_2(state):
    print('--Node2--')
    return {'graph_state': state['graph_state'] + " 快乐的"}

def node_3(state):
    print('--Node3--')
    return {'graph_state': state['graph_state'] + " 悲伤的"}

边（Edges）（和条件逻辑！）

边连接节点。虽然普通边只是按顺序移动，但条件边允许你的图根据状态或输出做出决策。

这是一个用于条件边的函数，它随机决定要采用哪个路径：

import random
from typing import Literal

## 用于路由条件边的函数
def mood(state) -> Literal["node_2","node_3"]:
    user_input = state['graph_state']       # 访问当前状态
    if random.random() > 0.5:               # 50% 的时间我们会返回 node2
        return 'call node_2'                # 此字符串与conditional_edges映射中的键匹配
    return 'call node_3'

节点更新状态，而条件边函数根据状态决定接下来要去哪个节点。

使用StateGraph构建图

构建StateGraph涉及几个关键步骤：

初始化StateGraph：将你的状态类传递给它。
添加节点：将你的函数注册为节点。
使用边定义流：使用add_edge连接你的节点以进行直接转换，并使用add_conditional_edges进行决策点。
设置START和END点。
编译你的图。

from langgraph.graph import StateGraph, START, END

## 构建图
workflow = StateGraph(State)           # 使用我们定义的状态初始化
workflow.add_node('node_1',node_1)
workflow.add_node('node_2',node_2)
workflow.add_node('node_3',node_3)

## 定义流程
workflow.add_edge(START,'node_1')      # 从node_1开始
workflow.add_conditional_edges(
    'node_1',                          # 来自node_1
    mood,                              # 使用情绪函数来决定
    {
        'call node_2':'node_2',        # 如果情绪返回'call node_2'，则转到node_2
        'call node_3':'node_3'         # 如果情绪返回'call node_3'，则转到node_3
    })
workflow.add_edge('node_2',END)        # 从node_2，转到END
workflow.add_edge('node_3',END)        # 从node_3，转到END
app = workflow.compile()               # 编译图

StateGraph 工作流

图调用

编译后的图（app）实现了Runnable协议，这意味着你可以使用诸如invoke()之类的方法执行它。

当调用invoke()时，图从START节点开始。它通过定义的节点和边进行。条件边将根据其逻辑路由执行。每个节点的函数接收当前状态并返回对其的更新。执行将继续，直到到达END节点。

# print(app.invoke({"graph_state":"嗨，我的名字是Ashutosh。"}))
# 这将从“嗨，我的名字是Ashutosh。”开始，然后node_1添加“我是”，
# 然后node_2添加“ 快乐的”或node_3添加“ 悲伤的”，具体取决于随机情绪函数。

invoke()同步运行整个图，并在所有节点都执行完毕后返回最终状态。

4. 构建对话链：消息和工具

现在是最令人兴奋的部分：构建更复杂的智能体，可以处理对话并使用外部工具。这结合了四个关键概念：聊天消息、聊天模型、将工具绑定到LLM以及在图中执行工具调用。

消息（Messages）

聊天模型不仅接受字符串；它们使用一系列“消息”来捕获对话中不同的角色。LangChain提供了消息类型，例如：

HumanMessage：用户所说的内容。
AIMessage：AI所说的内容。
SystemMessage：AI的指令。
ToolMessage：工具调用的输出。

每条消息都有内容、可选名称（谁发送的）和response_metadata。

from pprint import pprint
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

messages = [AIMessage(content=f'所以你在研究LangGraph?',name='Model')]
messages.extend([HumanMessage(content=f'是的', name='Ashu')])
messages.extend([AIMessage(content=f'你想学些什么?',name='Model')])
messages.extend([HumanMessage(content=f'我想学习langgraph中的状态图', name='Ashu')])

for msg in messages:
    msg.pretty_print()
    # 我们将获取这些消息的列表，并将其传递给我们的聊天模型。

聊天模型（Chat Model）

ChatModel（例如ChatOpenAI）将这些消息的列表作为输入，并生成AIMessage作为输出。

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model='gpt-4o-mini')
result = llm.invoke(messages) # 'messages'将是HumanMessage/AIMessage的列表

工具调用和工具（Tool Calling & Tools）

每当你想让你的AI智能体与外部世界互动时（例如，搜索网络、访问数据库、执行计算），工具就至关重要。大多数现代LLM都支持“工具调用”，它们输出结构化数据，指示要使用的工具。

以下是工具调用过程中的关键步骤：

工具创建：使用@tool装饰器创建工具。工具是Python函数及其模式之间的关联（该模式描述了其目的和所需的输入）。
工具绑定：工具需要连接到支持工具调用的模型。这使模型了解工具以及工具所需的关联输入模式。
工具调用：当适当的时候，基于用户的输入和对话历史，模型可以决定“调用”一个工具。它确保其响应符合工具的输入模式，并提供必要的参数。
工具执行：然后可以使用模型提供的参数执行实际的工具，并且通常将其输出返回到模型以告知其下一个响应。

from langchain_core.tools import tool

@tool
def multiply(a:int, b:int) ->int:
    "将a和b相乘"
    return a*b

# 将工具与LLM模型绑定
llm_with_tool = llm.bind_tools([multiply])
tool_call = llm_with_tool.invoke('2乘以3是多少')
print(tool_call)
# 我们发现AIMessage中没有内容，但是有工具调用。
# print(tool_call_result.additional_kwargs['tool_calls']) # 显示工具调用结构

使用消息作为状态和Reducer

在对话智能体中，你不希望每个节点完全覆盖消息历史记录。相反，你希望追加新消息。这就是LangGraph的Reducer派上用场的地方。

Reducer函数允许我们指定如何更新状态。如果未指定Reducer函数，则它将更新并覆盖键。

我们使用特殊的注释定义我们的MessageState：

from typing import Annotated
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

class MessageState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages] # 这告诉LangGraph将消息作为列表追加！

add_messages reducer确保当节点返回messages键时，其内容将附加到现有列表中，而不是替换它。

将所有内容放在一起：具有MessageState和工具调用的图

最后，让我们构建一个强大的图，它结合了所有这些概念：

它使用MessageState来管理对话历史记录。
它有一个LLM节点，可以决定调用工具。
它有一个ToolNode，可以执行LLM请求的任何工具。
它使用条件边（tools_condition）来路由到ToolNode（如果LLM调用了工具）。

from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages, AnyMessage
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_openai import ChatOpenAI

# 定义一些工具（例如，Web搜索、自定义函数）
# 我们正在使用Tavily Search获取实时结果
tavily_tool = TavilySearchResults(max_results=3)

@tool
def multiply(a:int,b:int)->int:
    """此工具将进行乘法运算"""
    return a*b

@tool
def add(a:int, b:int) -> int:
    "添加两个整数"
    return a+b

tools_list = [tavily_tool, multiply, add]

# 将工具绑定到你的LLM
llm_with_tool = ChatOpenAI(model='gpt-4o-mini').bind_tools(tools_list)

#始终确保在使用绑定工具时使用良好的推理模型，以确保基于上下文的准确决策、有效的任务编排和可靠的工具调用。

# 我们正在使用ToolNode，这是一个LangGraph预构建的类，它包装了外部工具或函数。当检测到工具调用时，它会被触发，使智能体能够执行适当的函数并将结果返回到工作流。
# 定义LLM聊天机器人的节点
def llm_chatbot(state: State):
    # 使用当前消息历史记录调用LLM
    return {'messages': [llm_with_tool.invoke(state['messages'])]}

# ToolNode将运行最后一个AIMessage请求的工具
# 如果调用了多个工具，它将并行运行
tool_node = ToolNode(tools_list) # 接受工具列表

#我们正在使用tools_condition，这是一个预构建的LangGraph函数，它检查最后一个状态消息是否包含工具调用。如果是，则流将路由到tools节点以进行执行。
#tools_condition的默认映射是{"tools": "tools_node_name"}

# 构建StateGraph
build = StateGraph(State)
build.add_node('LLM', llm_chatbot)
build.add_node('tools', tool_node) # 用于执行工具的节点
build.add_edge(START, 'LLM') # 首先将用户输入发送到LLM

# 添加从“LLM”开始的条件边
build.add_conditional_edges(
    "LLM",
    tools_condition, # 这是一个预构建的LangGraph条件：如果最后一条消息有工具调用，它将路由到“tools”
    # tools_condition的默认映射是 {"tools": "tools_node_name"}
    {"tools": "tools"})
build.add_edge('tools', 'LLM') # 工具运行后，将结果发送回LLM以进行下一轮对话

app = build.compile()

# 示例调用：
# 1. 自然响应（LLM直接回答）
# print(app.invoke({'messages': '你好'}))
# 2. 工具响应（LLM调用工具，然后提供答案）
# msg = app.invoke({'messages': "首先告诉我关于印度航空艾哈迈达巴德飞机坠毁事件。其次计算10*2"})
# for m in msg['messages']:
#     m.pretty_print()
# 预期的输出将显示AI消息、工具调用消息，然后是结合搜索结果和计算的最终AI响应。

这个最终示例展示了LangGraph的巨大威力。你构建了一个AI智能体，它可以：

维护对话（MessageState）。
智能地决定何时使用工具（tools_condition）。
执行这些工具（ToolNode）。
将工具结果合并回其推理中（通过在工具运行后路由回LLM）。

5. 总结

LangGraph直观的基于图的方法，结合其强大的状态管理和工具编排功能，使其成为任何希望构建复杂、适应性强且真正智能的智能体的AI专家的不可或缺的框架。模块化允许轻松调试和扩展，从而将复杂的AI工作流转换为清晰、可管理的结构。立即开始实验，释放AI智能体开发的下一个级别！

LangGraph入门：构建智能化工作流的初学者指南