[大模型架构] LangGraph AI 工作流编排（17）
·
一、工作流持久化的核心价值与设计思路
视频首先明确了 “工作流持久化” 的本质是 “将 LangGraph 工作流的执行状态（State）、节点流转记录、工具调用结果等关键数据持久化存储到本地 / 数据库”，核心价值与设计思路如下：
（一）核心价值
- 断点续跑：工作流因程序崩溃、网络中断、手动暂停等原因终止后，可从最后执行的节点恢复，无需重新执行全流程；
- 状态追溯：记录工作流每一步的执行状态、参数、结果，便于问题排查与审计；
- 分布式协作：Electron 主进程 / 渲染进程、多实例间可共享工作流状态，支持协作场景；
- 资源优化：长时工作流（如多文件处理、多轮工具调用）可分段执行，降低内存占用。
（二）核心设计思路
采用 “状态快照 + 节点日志 + 恢复触发器” 的三层设计：
- 状态快照：在关键节点执行完成后，将当前 State 完整序列化并存储；
- 节点日志：记录每一个节点的执行时间、输入参数、输出结果、执行状态（成功 / 失败）；
- 恢复触发器：检测到工作流中断后，读取最新快照与日志，定位中断节点，触发续跑逻辑。
二、LangGraph 工作流状态持久化实现（核心实操）
视频重点讲解了基于本地文件（JSON）+ SQLite 数据库的两种持久化方案（适配不同场景），并实现 LangGraph 与持久化层的无缝集成：
（一）方案 1：本地 JSON 文件持久化（轻量场景）
适合单文件、简单工作流的持久化，核心是实现 State 的序列化 / 反序列化：
-
定义持久化工具类：
python
运行
import json import os from datetime import datetime from langgraph.graph import State # 假设已有之前定义的PDFProcessState class PDFProcessState(State): file_path: str use_ocr: bool = False pdf_text: list = [] workflow_status: str = "running" error_message: str = "" current_node: str = "" # 新增：记录当前执行到的节点 execution_id: str = "" # 新增：工作流唯一标识 class WorkflowPersistence: """工作流持久化工具（JSON文件）""" def __init__(self, base_dir: str = "./workflow_snapshots"): self.base_dir = base_dir os.makedirs(base_dir, exist_ok=True) def _get_snapshot_path(self, execution_id: str) -> str: """生成快照文件路径""" return os.path.join(self.base_dir, f"{execution_id}.json") def save_snapshot(self, state: PDFProcessState) -> bool: """保存状态快照""" try: # 序列化State（转换为字典） state_dict = { "execution_id": state.execution_id, "file_path": state.file_path, "use_ocr": state.use_ocr, "pdf_text": state.pdf_text, "workflow_status": state.workflow_status, "error_message": state.error_message, "current_node": state.current_node, "timestamp": datetime.now().isoformat() # 快照生成时间 } # 写入JSON文件 with open(self._get_snapshot_path(state.execution_id), "w", encoding="utf-8") as f: json.dump(state_dict, f, ensure_ascii=False, indent=2) return True except Exception as e: print(f"保存快照失败：{str(e)}") return False def load_snapshot(self, execution_id: str) -> PDFProcessState | None: """加载最新快照，恢复State""" snapshot_path = self._get_snapshot_path(execution_id) if not os.path.exists(snapshot_path): return None try: with open(snapshot_path, "r", encoding="utf-8") as f: state_dict = json.load(f) # 反序列化为State对象 return PDFProcessState( execution_id=state_dict["execution_id"], file_path=state_dict["file_path"], use_ocr=state_dict["use_ocr"], pdf_text=state_dict["pdf_text"], workflow_status=state_dict["workflow_status"], error_message=state_dict["error_message"], current_node=state_dict["current_node"] ) except Exception as e: print(f"加载快照失败：{str(e)}") return None def delete_snapshot(self, execution_id: str) -> bool: """删除过期快照""" try: os.remove(self._get_snapshot_path(execution_id)) return True except Exception as e: print(f"删除快照失败：{str(e)}") return False -
集成到 LangGraph 节点中：在每个节点执行完成后自动保存快照：
python
运行
# 初始化持久化工具 persistence = WorkflowPersistence() # 改造PDF提取节点，添加快照保存逻辑 def pdf_extract_node(state: PDFProcessState) -> PDFProcessState: # 标记当前执行节点 state.current_node = "pdf_extract" try: tool = PDFTextExtractTool() result = tool.run({"file_path": state.file_path, "use_ocr": state.use_ocr}) if result["success"]: state.pdf_text = result["data"] state.workflow_status = "completed" else: state.error_message = result["message"] state.workflow_status = "failed" # 保存快照（执行完成后） persistence.save_snapshot(state) return state except Exception as e: state.error_message = str(e) state.workflow_status = "failed" # 即使出错也保存快照（便于排查） persistence.save_snapshot(state) return state
（二）方案 2：SQLite 数据库持久化（生产 / 复杂场景）
适合多工作流、需结构化查询的场景，视频中使用sqlite3模块实现：
-
创建数据库表结构：
python
运行
import sqlite3 def init_workflow_db(db_path: str = "./workflow_db.sqlite"): """初始化工作流数据库""" conn = sqlite3.connect(db_path) cursor = conn.cursor() # 1. 工作流状态表（存储快照） cursor.execute(''' CREATE TABLE IF NOT EXISTS workflow_snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, execution_id TEXT UNIQUE NOT NULL, state_data TEXT NOT NULL, # 序列化的State（JSON字符串） current_node TEXT NOT NULL, workflow_status TEXT NOT NULL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) ''') # 2. 节点执行日志表（存储每一步执行记录） cursor.execute(''' CREATE TABLE IF NOT EXISTS node_execution_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, execution_id TEXT NOT NULL, node_name TEXT NOT NULL, input_params TEXT NOT NULL, output_result TEXT, status TEXT NOT NULL, # success/failed/running error_msg TEXT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (execution_id) REFERENCES workflow_snapshots(execution_id) ) ''') conn.commit() conn.close() # 初始化数据库 init_workflow_db() -
实现数据库持久化工具类：
python
运行
class DBWorkflowPersistence: """工作流持久化工具（SQLite）""" def __init__(self, db_path: str = "./workflow_db.sqlite"): self.db_path = db_path init_workflow_db(db_path) def save_snapshot(self, state: PDFProcessState) -> bool: """保存状态快照到数据库""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 序列化State为JSON字符串 state_json = json.dumps({ "execution_id": state.execution_id, "file_path": state.file_path, "use_ocr": state.use_ocr, "pdf_text": state.pdf_text, "error_message": state.error_message }, ensure_ascii=False) # 插入/更新快照 cursor.execute(''' REPLACE INTO workflow_snapshots (execution_id, state_data, current_node, workflow_status, timestamp) VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP) ''', (state.execution_id, state_json, state.current_node, state.workflow_status)) conn.commit() conn.close() return True except Exception as e: print(f"数据库保存快照失败：{str(e)}") return False def save_node_log(self, execution_id: str, node_name: str, input_params: dict, output_result: dict, status: str, error_msg: str = "") -> bool: """保存节点执行日志""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' INSERT INTO node_execution_logs (execution_id, node_name, input_params, output_result, status, error_msg) VALUES (?, ?, ?, ?, ?, ?) ''', ( execution_id, node_name, json.dumps(input_params, ensure_ascii=False), json.dumps(output_result, ensure_ascii=False), status, error_msg )) conn.commit() conn.close() return True except Exception as e: print(f"保存节点日志失败：{str(e)}") return False def load_latest_snapshot(self, execution_id: str) -> PDFProcessState | None: """加载最新的工作流快照""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' SELECT state_data, current_node, workflow_status FROM workflow_snapshots WHERE execution_id = ? ORDER BY timestamp DESC LIMIT 1 ''', (execution_id,)) row = cursor.fetchone() conn.close() if not row: return None # 反序列化State state_dict = json.loads(row[0]) return PDFProcessState( execution_id=execution_id, file_path=state_dict["file_path"], use_ocr=state_dict["use_ocr"], pdf_text=state_dict["pdf_text"], error_message=state_dict["error_message"], current_node=row[1], workflow_status=row[2] ) except Exception as e: print(f"加载数据库快照失败：{str(e)}") return None
三、断点续跑机制实现（核心难点）
视频中实现了 “中断检测→快照加载→节点定位→续跑执行” 的完整断点续跑逻辑，适配主动暂停、程序崩溃两种中断场景：
（一）断点续跑核心逻辑
python
运行
import uuid
from langgraph import Graph
# 初始化数据库持久化工具
db_persistence = DBWorkflowPersistence()
def run_workflow(execution_id: str = None, resume: bool = False) -> PDFProcessState:
"""
运行/续跑工作流
:param execution_id: 工作流ID（续跑时必填）
:param resume: 是否续跑
:return: 最终状态
"""
# 1. 处理续跑逻辑
if resume and execution_id:
# 加载最新快照
state = db_persistence.load_latest_snapshot(execution_id)
if not state:
raise ValueError(f"未找到执行ID为{execution_id}的工作流快照")
# 检查工作流状态（已完成则无需续跑）
if state.workflow_status == "completed":
print("工作流已完成，无需续跑")
return state
else:
# 新建工作流，生成唯一ID
execution_id = str(uuid.uuid4())
state = PDFProcessState(
execution_id=execution_id,
file_path="C:/docs/contract.pdf",
use_ocr=True,
current_node="start",
workflow_status="running"
)
# 保存初始快照
db_persistence.save_snapshot(state)
# 2. 构建工作流（复用之前的节点）
graph = Graph(PDFProcessState)
graph.add_node("pdf_extract", pdf_extract_node)
graph.add_node("text_clean", text_clean_node)
graph.add_edge("pdf_extract", "text_clean")
graph.set_entry_point("pdf_extract")
graph.set_finish_point("text_clean")
app = graph.compile()
# 3. 续跑：定位中断节点，从该节点开始执行
if resume:
# 获取中断节点
start_node = state.current_node
if start_node == "pdf_extract":
# 从pdf_extract节点续跑
final_state = app.invoke(state, start_node=start_node)
elif start_node == "text_clean":
# 从text_clean节点续跑
final_state = app.invoke(state, start_node=start_node)
else:
# 未知节点，从头执行
final_state = app.invoke(state)
else:
# 全新执行：从头开始
final_state = app.invoke(state)
# 4. 保存最终快照
db_persistence.save_snapshot(final_state)
return final_state
# 示例1：全新执行工作流
new_exec_id = str(uuid.uuid4())
run_workflow(execution_id=new_exec_id, resume=False)
# 示例2：续跑中断的工作流（假设程序崩溃后）
run_workflow(execution_id=new_exec_id, resume=True)
（二）主动暂停与恢复（Electron 前端控制）
实现前端手动暂停工作流，后续可恢复执行：
python
运行
# 新增：工作流暂停标记（存储在State中）
class PDFProcessState(State):
# 原有字段...
is_paused: bool = False # 新增：暂停标记
# 改造节点逻辑，检测暂停标记
def pdf_extract_node(state: PDFProcessState) -> PDFProcessState:
state.current_node = "pdf_extract"
# 检测是否被暂停
if state.is_paused:
state.workflow_status = "paused"
db_persistence.save_snapshot(state) # 保存暂停状态
return state
# 原有执行逻辑...
# 前端触发暂停的IPC处理（Electron主进程）
# main.js
ipcMain.handle('pause_workflow', async (_, execution_id) => {
// 更新State中的暂停标记
const state = db_persistence.load_latest_snapshot(execution_id);
if (state) {
state.is_paused = true;
db_persistence.save_snapshot(state);
return { success: true, message: "工作流已暂停" };
}
return { success: false, message: "工作流不存在" };
});
# 前端触发恢复的IPC处理
ipcMain.handle('resume_workflow', async (_, execution_id) => {
const state = db_persistence.load_latest_snapshot(execution_id);
if (state) {
state.is_paused = false;
// 调用续跑逻辑
const result = run_workflow(execution_id, resume=True);
return { success: true, data: result };
}
return { success: false, message: "工作流不存在" };
});
四、Electron 端状态管理与可视化
视频最后实现了前端对工作流状态的可视化展示、中断恢复操作，让用户可直观管理工作流：
（一）前端工作流状态面板（React 示例）
tsx
import { useState, useEffect } from 'react';
import { Table, Button, Tag, Modal } from 'antd';
import { PlayCircleOutlined, PauseCircleOutlined, ReloadOutlined } from '@ant-design/icons';
const WorkflowStatusPanel = () => {
const [workflowList, setWorkflowList] = useState([]);
const [loading, setLoading] = useState(false);
// 加载所有工作流状态
useEffect(() => {
fetchWorkflowList();
}, []);
const fetchWorkflowList = async () => {
setLoading(true);
// 调用IPC获取所有工作流快照
const list = await window.electronAPI.getWorkflowList();
setWorkflowList(list);
setLoading(false);
};
// 暂停工作流
const handlePause = async (execution_id) => {
await window.electronAPI.pauseWorkflow(execution_id);
fetchWorkflowList(); // 刷新列表
};
// 恢复工作流
const handleResume = async (execution_id) => {
await window.electronAPI.resumeWorkflow(execution_id);
fetchWorkflowList(); // 刷新列表
};
// 表格列配置
const columns = [
{
title: '工作流ID',
dataIndex: 'execution_id',
key: 'execution_id',
ellipsis: true,
},
{
title: '文件路径',
dataIndex: 'file_path',
key: 'file_path',
ellipsis: true,
},
{
title: '当前节点',
dataIndex: 'current_node',
key: 'current_node',
},
{
title: '状态',
dataIndex: 'workflow_status',
key: 'workflow_status',
render: (status) => {
const color = status === 'running' ? 'blue' : status === 'completed' ? 'green' : status === 'failed' ? 'red' : 'orange';
return <Tag color={color}>{status}</Tag>;
},
},
{
title: '操作',
key: 'action',
render: (_, record) => (
<>
{record.workflow_status === 'running' && (
<Button icon={<PauseCircleOutlined />} onClick={() => handlePause(record.execution_id)}>
暂停
</Button>
)}
{(record.workflow_status === 'paused' || record.workflow_status === 'failed') && (
<Button icon={<ReloadOutlined />} onClick={() => handleResume(record.execution_id)}>
恢复
</Button>
)}
</>
),
},
];
return (
<Table
columns={columns}
dataSource={workflowList}
loading={loading}
rowKey="execution_id"
pagination={{ pageSize: 10 }}
/>
);
};
五、常见问题与优化策略
| 问题现象 | 核心原因 | 解决方案 |
|---|---|---|
| 快照保存频繁导致性能下降 | 每个节点都保存快照，IO 操作过多 | 1. 仅在关键节点（如工具调用完成、状态变更）保存快照；2. 批量写入，减少 IO 次数；3. 使用内存缓存 + 定时刷盘 |
| 续跑时节点执行重复 | 中断节点已执行部分逻辑，但快照未记录 | 1. 节点执行逻辑设计为幂等性（重复执行结果一致）；2. 快照中记录节点 “执行进度”（如 PDF 提取到第 3 页） |
| 序列化失败（State 含不可序列化对象） | State 中包含函数、文件句柄等不可序列化数据 | 1. 仅存储可序列化数据（字符串、数字、列表、字典）；2. 敏感 / 不可序列化数据通过 ID 关联存储 |
| 数据库锁导致快照保存失败 | 多进程同时读写数据库 | 1. 使用数据库连接池；2. 加锁机制（如文件锁）；3. 采用异步写入（队列） |
| 大文件快照加载缓慢 | State 中存储大文本 / 二进制数据 | 1. 大数据单独存储（如文件路径），快照中仅存引用；2. 快照压缩存储（gzip） |
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
19
0- 0
已为社区贡献23条内容
所有评论(0)