LangGraph简介：基于图结构的 AI 工作流框架

1. 定义与定位

• LangGraph 是由 LangChain 团队开发的开源框架，专为构建有状态、多代理、循环式 AI 工作流设计。它基于 LangChain 生态，通过图结构（节点+边） 建模复杂任务流程，支持动态调整、状态管理和多步骤决策，适用于需要循环交互或复杂任务分解的场景。

2. 核心特性

• 图结构驱动：以 StateGraph 为核心，节点代表计算步骤（如 LLM 调用、工具执行），边定义状态转移路径，支持有向图（含循环） 而非仅限于有向无环图（DAG）。
• 状态管理：通过 State 字典在节点间传递和更新状态（如对话历史、工具调用结果），支持自定义状态类型（如消息列表）。
• 多代理协调：支持多 LLM 代理或工具的协同工作，例如通过 ToolNode 集成搜索、计算等外部工具。
• 人类干预：内置 interrupt() 函数实现“人在回路”（Human-in-the-Loop），允许在工具调用前暂停执行，等待人工审批或反馈。
• 持久化存储：提供内存、文件系统或数据库（如 PostgreSQL + pgvector）存储选项，支持跨会话数据保留和语义搜索。

3. 典型使用场景

• 对话系统：构建多轮对话机器人，支持上下文记忆、工具调用（如搜索、计算）和动态响应生成。
• 任务分解：将复杂任务拆分为子任务，通过图结构管理执行顺序和依赖关系（如自动化报告生成、数据清洗）。
• 多代理系统：协调多个 AI 代理协作完成任务（如客服系统中的问题分类、解决方案推荐）。
• 自动化工作流：在金融、电商等领域实现自动化决策（如订单处理、交易监控），支持高并发和动态调整。

环境配置与安装

为避免不同环境的冲突，可以新建一个环境，再安装相应的库。

conda create -n langgraph_env python=3.10 # 创建新虚拟环境

conda activate langgraph_env # 激活虚拟环境

# 安装相应的库
pip install langgraph 
pip install openai

首先我们需了解langgraph的四个核心组件：

• State: 在图中传递的数据结构

• Node: 执行具体任务的函数

• Edge: 连接节点，决定流程走向

• Graph: 由节点和边组成的完整工作流

怎么理解呢，可以把Graph看作是一张地图，Node就是地图上的城市，Edge就连接不同城市的道路（当然，两个城市之间有没有道路连接可以由自己决定），那么State就是流转于这些城市之间的货车。

一开始，我们会把一个Node当作开始结点，相当于货车从某个城市开始出发。然后沿着规定好的道路行驶，每到一个城市（Node）就会对车上的货物进行处理，就相当于改变State的状态。货车一直行驶下去，到达规定的最后一个城市（Node结束结点）。注意：这个graph应该是有向无环图，开始结点和结束结点都只能有一个。

简单工具流的搭建

建立图

首先定义State数据结构

这一步相当于确定好货车。

from typing import TypedDict, List

class State(TypedDict):
    data: str    # 处理的数据
    current_step: str    # 记录当前到达哪个结点
    node_history: List[str]    # 记录经过的结点

上述代码中，定义了一个State类，其继承于TypedDict，本质上是一个字典，我们定义了3个键值对。data:str 表示键为'data'，而值应该为str类型。

定义开始结点

相当于确定好货车的出发城市

def start_node(state: State) -> State:
    state['current_step'] = 'start_node'   # 记录当前结点 
    print('当前结点：', state['current_step'])
    state['node_history'].append('start_node')    # 记录历史结点
    return state

def start_node(state: State) -> State:

表示声明开始结点函数，state:State表示函数的参数类型是State，->State表示函数的返回值期望是State类型。

定义处理结点

确定好货车经过的城市，在这里可以对数据state做一些处理。

def process_node(state:State) -> State:
    state['current_step'] = 'process_node'
    print('当前结点：', state['current_step'])
    state['node_history'].append('process_node')
    state['data'] = '你好，世界'
    return state

定义结束结点

确定货车的目的地城市

def end_node(state:State) -> State:
    state['current_step'] = 'end_node'
    print('当前结点：', state['current_step'])
    state['node_history'].append('end_node')
    return state

在结点创建好后，我们需要把结点连接起来，形成一个graph地图。

from langgraph.graph import StateGraph, END

# 先定义一个空的地图
builder = StateGraph(State)

# 这一步先把所有结点加入到地图中
builder.add_node('start', start_node)    # 第一个参数是给结点取名，第二个参数是结点
builder.add_node('process', process_node)
builder.add_node('end', end_node)

# 在结点之间添加边
builder.set_entry_point('start') # 确定出发城市
builder.add_edge('start', 'process')
builder.add_edge('process', 'end')
builder.add_edge('end', END) # 注意，这里确定结束结点的方式不大一样

# 最后要对图进行编译
graph = builder.compile()

运行工作流

我们的工作流已经搭建完成了，接下来就是试一下能否正常运行。

首先先初始化一下state：

test_state: State = {
    'data': '',
    'current_step': '',
    'node_history': [],
}

然后通过的invoke方法把state传进去:

result = graph.invoke(test_state)
print(result)

上述代码中，使用result接受工作流的结果，然后打印result。

运行一下看看结果：

可以看到，state经过的每个结点及处理后的最终结果。

下面是完整代码：

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

class State(TypedDict):
    data: str    # 处理的数据
    current_step: str    # 记录当前到达哪个结点
    node_history: List[str]    # 记录经过的结点


def start_node(state: State) -> State:
    state['current_step'] = 'start_node'   # 记录当前结点
    print('当前结点：', state['current_step'])
    state['node_history'].append('start_node')    # 记录历史结点
    return state


def process_node(state:State) -> State:
    state['current_step'] = 'process_node'
    print('当前结点：', state['current_step'])
    state['node_history'].append('process_node')
    state['data'] = '你好，世界'
    return state


def end_node(state:State) -> State:
    state['current_step'] = 'end_node'
    print('当前结点：', state['current_step'])
    state['node_history'].append('end_node')
    return state


# 先定义一个空的地图
builder = StateGraph(State)

# 这一步先把所有结点加入到地图中
builder.add_node('start', start_node)    # 第一个参数是给结点取名，第二个参数是结点
builder.add_node('process', process_node)
builder.add_node('end', end_node)

# 在结点之间添加边
builder.set_entry_point('start') # 确定出发城市
builder.add_edge('start', 'process')
builder.add_edge('process', 'end')
builder.add_edge('end', END) # 注意，这里确定结束结点的方式不大一样

# 最后要对图进行编译
graph = builder.compile()

test_state: State = {
    'data': '',
    'current_step': '',
    'node_history': [],
}

result = graph.invoke(test_state)
print(result)

加入AI对话

在上述工作流搭建的基础上，我们可以把AI加入到我们的工作流当中。

这里我选择使用deepseek-chat模型。

获取api_key

首先我们要到deepseek开发平台上获取api_key，网址https://platform.deepseek.com/

注册并实名认证之后，在左边栏的用量信息这里去充值（只需充一点点小钱就行，我充了5块钱就已经可以使用很久了），也可以查一下左下的产品定价。

              

然后就可以点击左边栏的API Keys， 然后在界面里点击创建

             

输入自己定的api key的名字后，就可以复制自己的api_key了。注意api_key不要泄露出去，也不要与他人共享哦。

那么为了保护好api_key，我们尽量不要直接在代码中显式指定api_key。有两种方法使用api_key：

方法一：把api_key保存到系统环境变量中。

搜索环境变量，点击打开

点击环境变量

在系统变量这里点击新建

输入变量名，在变量值输入刚才的获取api_key

然后一直点击确定即可。

方法二：使用.env

在代码的同目录下新建一个.env文件，注意文件名就叫.env

不会的可以右键打开终端，然后在终端中输入：

New-Item .env

然后右键用记事本打开.env文件，把api_key记录进去（注意等号两边不要有空格）

保存并退出。

创建AI客户端

接下来就可以创建AI客户端了。

先在代码中获取api_key

若你保存api_key时采用方法一：

import os 

api_key = os.getenv('OPENAI_API_KEY')

os.getenv会自动查找系统环境变量里变量名为OPENAI_API_KEY的环境变量，并把变量值赋值给api_key。

若你采用方法二：

from dotenv import load_dotenv
import os

# 加载.env到当前进行的环境变量里
load_dotenv()

api_key = os.getenv("OPENAI_API_KEY")

因为os.getenv只能在系统环境变量里查找，所以我们存在.env里的api_key是找不到的。所以使用load_dotenv()可以把.env加载到当前运行的环境变量里，然后就可以用os.getenv获得api_key了。

下面我使用方法二，连接AI，并向AI提问一个小问题，完整代码如下：

from dotenv import load_dotenv
from openai import OpenAI
import os

# 加载.env到当前进行的环境变量里
load_dotenv()

api_key = os.getenv("OPENAI_API_KEY")

client = OpenAI(
    api_key=api_key,
    base_url = "https://api.deepseek.com",
)

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是一个AI智能助手"},
        {"role": "user", "content": "请问一台计算机主要由哪几部分构成？"}
    ],
    temperature=0.1
)

print(response.choices[0].message.content)

上述代码中：client是创建好的AI客户端。

client.chat.completions.create()是向api发送请求，获得的回应保存在response。

其中的参数：

model 表示指定的模型。

messages 表示发送给AI的提示，{"role": "system", "content": "你是一个AI智能助手"} 表示这是系统提示，告诉AI应该扮演什么角色； {"role": "user", "content": "请问一台计算机主要由哪几部分构成？"} 是用户提示，用于将用户的问题发送给AI。

temperature 控制生成文本的随机性/创造性（范围：0.0 到 2.0） 具体含义： 低温度 (0.0-0.3)：更确定、一致的回答，适合事实性任务。中温度 (0.4-0.7)：平衡创造性和一致性，适合一般对话。高温度 (0.8-2.0)：更有创造性、随机性，适合创意写作

返回的response是一个json对象，包含许多内容，如token用量。使用response.choices[0].message.content将具体的回答内容取出。

运行结果如下：

AI聊天的工作流搭建

接下来，我们先抛掉上面的所有代码，建立一个AI聊天的工作流

下面是所需的库文件

from langgraph.graph import StateGraph, END
from typing import TypedDict, List
import os
from dotenv import load_dotenv
from openai import OpenAI

AI客户端建立:

def create_client():
    # 加载.env到当前进行的环境变量里
    load_dotenv()
    
    api_key = os.getenv("OPENAI_API_KEY")
    
    client = OpenAI(
        api_key=api_key,
        base_url = "https://api.deepseek.com",
    )
    
    return client

为了能够与AI进行交互的state定义并初始化：

class State(TypedDict):
    user_input: str    # 用户输入的数据
    ai_response: str    # ai的回应
    current_step: str    # 记录当前到达哪个结点
    node_history: List[str]    # 记录经过的结点

开始结点、处理结点和结束结点的定义：

def start_node(state: State) -> State:
    state['current_step'] = 'start_node'   # 记录当前结点
    print('当前结点：', state['current_step'])
    state['node_history'].append('start_node')    # 记录历史结点
    return state

# 处理结点
def process_node(state: State, client: OpenAI) -> State:
    state['current_step'] = 'process_node'
    print('当前结点：', state['current_step'])
    state['node_history'].append('process_node')

    user_input = state['user_input']
    # 构建prompt
    message = [
        {"role": "system", "content": "你是一个AI智能助手"},
        {"role": "user", "content": f"{user_input}"}
    ]

    # 获得回应
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=message,
        temperature=0.2,
    )

    # 保存回应结果
    state['ai_response'] = response.choices[0].message.content

    return state

# 结束结点
def end_node(state:State) -> State:
    state['current_step'] = 'end_node'
    print('当前结点：', state['current_step'])
    state['node_history'].append('end_node')
    return state

利用结点创建图：

def create_graph(client: OpenAI) -> StateGraph:
    # 先定义一个空的地图
    builder = StateGraph(State)

    # 这一步先把所有结点加入到地图中
    builder.add_node('start', start_node)    # 第一个参数是给结点取名，第二个参数是结点
    builder.add_node('process', lambda state: process_node(state, client))
    builder.add_node('end', end_node)

    # 在结点之间添加边
    builder.set_entry_point('start') # 确定出发城市
    builder.add_edge('start', 'process')
    builder.add_edge('process', 'end')
    builder.add_edge('end', END) # 注意，这里确定结束结点的方式不大一样

    # 最后要对图进行编译
    graph = builder.compile()

    return graph

定义主函数：

def main():
    # 建立客户端
    client = create_client()
    print('AI客户端建立成功！')
    graph = create_graph(client)
    print('工作流建立成功！')
    user_input = input('请输入：')

    # 初始化状态
    state = {
        'user_input': user_input,
        'ai_response': '',
        'current_step': '',
        'node_history': []
    }
    
    # 传入状态，获得结果
    result_state = graph.invoke(state)

    print('AI回应结果：', result_state['ai_response'])

在以上函数都定义完后，使用：

if __name__ == '__main__':
    main()

来调用主函数。

接下来运行程序，得到如下输出：

若想重复输入，可以在主函数中加入循环：

def main():
    # 建立客户端
    client = create_client()
    print('AI客户端建立成功！')
    graph = create_graph(client)
    print('工作流建立成功！')
    
    while(True):
        user_input = input('请输入：')
    
        # 初始化状态
        state = {
            'user_input': user_input,
            'ai_response': '',
            'current_step': '',
            'node_history': []
        }
    
        # 传入状态，获得结果
        result_state = graph.invoke(state)
    
        print('AI回应结果：', result_state['ai_response'])

输出结果：

以下是完整代码：

from langgraph.graph import StateGraph, END
from typing import TypedDict, List
import os
from dotenv import load_dotenv
from openai import OpenAI


def create_client():
    # 加载.env到当前进行的环境变量里
    load_dotenv()

    api_key = os.getenv("OPENAI_API_KEY")

    client = OpenAI(
        api_key=api_key,
        base_url = "https://api.deepseek.com",
    )

    return client

class State(TypedDict):
    user_input: str    # 用户输入的数据
    ai_response: str    # ai的回应
    current_step: str    # 记录当前到达哪个结点
    node_history: List[str]    # 记录经过的结点

# 开始结点
def start_node(state: State) -> State:
    state['current_step'] = 'start_node'   # 记录当前结点
    print('当前结点：', state['current_step'])
    state['node_history'].append('start_node')    # 记录历史结点
    return state

# 处理结点
def process_node(state: State, client: OpenAI) -> State:
    state['current_step'] = 'process_node'
    print('当前结点：', state['current_step'])
    state['node_history'].append('process_node')

    user_input = state['user_input']
    # 构建prompt
    message = [
        {"role": "system", "content": "你是一个AI智能助手"},
        {"role": "user", "content": f"{user_input}"}
    ]

    # 获得回应
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=message,
        temperature=0.2,
    )

    # 保存回应结果
    state['ai_response'] = response.choices[0].message.content

    return state

# 结束结点
def end_node(state:State):
    state['current_step'] = 'end_node'
    print('当前结点：', state['current_step'])
    state['node_history'].append('end_node')
    return state


def create_graph(client: OpenAI) -> StateGraph:
    # 先定义一个空的地图
    builder = StateGraph(State)

    # 这一步先把所有结点加入到地图中
    builder.add_node('start', lambda state: start_node(state))    # 第一个参数是给结点取名，第二个参数是结点
    builder.add_node('process', lambda state: process_node(state, client))
    builder.add_node('end', lambda state: end_node(state))

    # 在结点之间添加边
    builder.set_entry_point('start') # 确定出发城市
    builder.add_edge('start', 'process')
    builder.add_edge('process', 'end')
    builder.add_edge('end', END) # 注意，这里确定结束结点的方式不大一样

    # 最后要对图进行编译
    graph = builder.compile()

    return graph

def main():
    # 建立客户端
    client = create_client()
    print('AI客户端建立成功！')
    graph = create_graph(client)
    print('工作流建立成功！')

    while(True):
        user_input = input('请输入：')

        # 初始化状态
        state = {
            'user_input': user_input,
            'ai_response': '',
            'current_step': '',
            'node_history': []
        }

        # 传入状态，获得结果
        result_state = graph.invoke(state)

        print('AI回应结果：', result_state['ai_response'])


if __name__ == '__main__':
    main()

以上就是一个基于langgraph的简单AI聊天机器人的搭建了，本人水平有限，如有错误欢迎指出。如果这篇文章对你有帮助，请动动手指点个赞哦，谢谢~