[大模型架构] LangGraph AI 工作流编排（18）

一、并发控制的核心价值与设计原则

视频首先明确了 “工作流并发控制” 的本质是 “在多线程 / 进程 / 分布式环境下，安全、高效地并行执行 LangGraph 节点，同时保证状态一致性与资源不超限”，核心价值与设计原则如下：

（一）核心价值

（二）核心设计原则

二、LangGraph 单进程并发执行实现（核心实操）

视频首先实现了单进程内的多线程并发，基于concurrent.futures.ThreadPoolExecutor管控并发节点执行：

（一）线程池并发工作流封装

python

运行

import uuid
from concurrent.futures import ThreadPoolExecutor, as_completed
from langgraph import Graph
from typing import List, Dict, Any
# 复用之前定义的PDFProcessState和DBWorkflowPersistence
from persistence import PDFProcessState, DBWorkflowPersistence

# 初始化数据库持久化工具
db_persistence = DBWorkflowPersistence()

class ConcurrentWorkflowManager:
"""并发工作流管理器"""
def __init__(self, max_workers: int = 4):
"""
:param max_workers: 最大并发数（建议等于CPU核心数）
"""
self.max_workers = max_workers
self.executor = ThreadPoolExecutor(max_workers=max_workers)
# 存储正在执行的任务（task_id: future）
self.running_tasks: Dict[str, Any] = {}

def build_workflow(self) -> Any:
"""构建基础工作流（复用之前的节点）"""
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")
return graph.compile()

def run_single_workflow(self, file_path: str, use_ocr: bool = False) -> PDFProcessState:
"""执行单个工作流（独立State）"""
execution_id = str(uuid.uuid4())
# 初始化独立State
state = PDFProcessState(
execution_id=execution_id,
file_path=file_path,
use_ocr=use_ocr,
current_node="start",
workflow_status="running"
)
# 保存初始快照
db_persistence.save_snapshot(state)
# 执行工作流
app = self.build_workflow()
final_state = app.invoke(state)
# 保存最终快照
db_persistence.save_snapshot(final_state)
return final_state

def run_concurrent_workflows(self, task_list: List[Dict[str, Any]]) -> List[PDFProcessState]:
"""
并发执行多个工作流任务
:param task_list: 任务列表，格式：[{"file_path": "…", "use_ocr": …}, …]
:return: 所有任务的执行结果
"""
futures = []
results = []
# 提交并发任务
for task in task_list:
task_id = str(uuid.uuid4())
future = self.executor.submit(self.run_single_workflow, task["file_path"], task.get("use_ocr", False))
self.running_tasks[task_id] = future
futures.append((task_id, future))

# 收集执行结果
for task_id, future in futures:
try:
result = future.result() # 获取任务结果
results.append(result)
except Exception as e:
# 记录失败任务状态
error_state = PDFProcessState(
execution_id=task_id,
file_path=task_list[futures.index((task_id, future))]["file_path"],
workflow_status="failed",
error_message=f"并发执行失败：{str(e)}"
)
db_persistence.save_snapshot(error_state)
results.append(error_state)
finally:
del self.running_tasks[task_id]

return results

def cancel_task(self, task_id: str) -> bool:
"""取消单个正在执行的并发任务"""
if task_id in self.running_tasks:
future = self.running_tasks[task_id]
if not future.done():
cancel_success = future.cancel()
if cancel_success:
# 更新任务状态为已取消
cancel_state = PDFProcessState(
execution_id=task_id,
workflow_status="cancelled",
error_message="任务已手动取消"
)
db_persistence.save_snapshot(cancel_state)
del self.running_tasks[task_id]
return cancel_success
return False

（二）并发任务调用示例

python

运行

# 初始化并发管理器（最大并发数4）
concurrent_manager = ConcurrentWorkflowManager(max_workers=4)

# 批量任务列表（5个PDF提取任务）
task_list = [
{"file_path": "C:/docs/1.pdf", "use_ocr": False},
{"file_path": "C:/docs/2.pdf", "use_ocr": True},
{"file_path": "C:/docs/3.pdf", "use_ocr": False},
{"file_path": "C:/docs/4.pdf", "use_ocr": True},
{"file_path": "C:/docs/5.pdf", "use_ocr": False},
]

# 执行并发任务
results = concurrent_manager.run_concurrent_workflows(task_list)

# 打印执行结果
for result in results:
print(f"任务ID：{result.execution_id}，文件：{result.file_path}，状态：{result.workflow_status}")

三、进程池并发（突破 GIL 限制）

针对 CPU 密集型任务（如 OCR 识别、文本处理），视频中使用multiprocessing.Pool实现进程级并发，规避 Python GIL（全局解释器锁）的限制：

（一）进程池并发实现

python

运行

import multiprocessing
from multiprocessing import Pool
import os

# 注意：进程池执行的函数需可序列化，建议单独定义
def run_worker_task(task: Dict[str, Any]) -> Dict[str, Any]:
"""进程池工作函数（独立进程执行）"""
# 初始化进程内的持久化工具（每个进程独立连接）
worker_persistence = DBWorkflowPersistence()
execution_id = str(uuid.uuid4())
try:
state = PDFProcessState(
execution_id=execution_id,
file_path=task["file_path"],
use_ocr=task.get("use_ocr", False),
workflow_status="running"
)
# 构建工作流并执行
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")
app = graph.compile()
final_state = app.invoke(state)
# 保存快照（进程内独立保存）
worker_persistence.save_snapshot(final_state)
return {
"success": True,
"execution_id": execution_id,
"state": final_state.dict() # 转换为字典便于进程间传递
}
except Exception as e:
error_state = PDFProcessState(
execution_id=execution_id,
file_path=task["file_path"],
workflow_status="failed",
error_message=str(e)
)
worker_persistence.save_snapshot(error_state)
return {
"success": False,
"execution_id": execution_id,
"error": str(e)
}

class ProcessPoolWorkflowManager:
"""进程池并发工作流管理器"""
def __init__(self, max_processes: int = None):
"""
:param max_processes: 最大进程数（默认CPU核心数）
"""
self.max_processes = max_processes or multiprocessing.cpu_count()
self.pool = Pool(processes=self.max_processes)

def run_concurrent_tasks(self, task_list: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""进程池执行并发任务"""
# 提交任务到进程池
results = self.pool.map(run_worker_task, task_list)
# 关闭进程池
self.pool.close()
self.pool.join()
return results

# 调用示例
if __name__ == "__main__": # 进程池必须在main函数中调用
process_manager = ProcessPoolWorkflowManager(max_processes=4)
task_list = [
{"file_path": "C:/docs/1.pdf", "use_ocr": True},
{"file_path": "C:/docs/2.pdf", "use_ocr": True},
]
results = process_manager.run_concurrent_tasks(task_list)
print(results)

四、分布式执行与状态同步（跨 Electron 实例）

视频中实现了基于共享 SQLite 数据库 + 文件锁的分布式执行方案，支持多个 Electron 实例 / 设备并行执行工作流，同时保证状态一致性：

（一）分布式架构设计

预览

查看代码

执行任务/更新状态

执行任务/更新状态

状态同步

状态同步

防止并发写入冲突

Electron实例1

共享SQLite数据库

Electron实例2

文件锁

flowchart LR
A[Electron实例1] –>|执行任务/更新状态| C[共享SQLite数据库]
B[Electron实例2] –>|执行任务/更新状态| C
C –>|状态同步| A
C –>|状态同步| B
D[文件锁] –>|防止并发写入冲突| C

执行任务/更新状态

执行任务/更新状态

状态同步

状态同步

防止并发写入冲突

Electron实例1

共享SQLite数据库

Electron实例2

文件锁

豆包

你的 AI 助手，助力每日工作学习

（二）分布式状态同步实现

python

运行

import fcntl
import os

class DistributedWorkflowManager:
"""分布式工作流管理器（跨实例）"""
def __init__(self, db_path: str = "./shared_workflow_db.sqlite", lock_file: str = "./workflow_lock.lock"):
self.db_path = db_path
self.lock_file = lock_file
self.persistence = DBWorkflowPersistence(db_path)

def _acquire_lock(self) -> None:
"""获取文件锁（防止并发写入）"""
self.lock_fd = open(self.lock_file, "w")
fcntl.flock(self.lock_fd, fcntl.LOCK_EX) # 排他锁

def _release_lock(self) -> None:
"""释放文件锁"""
fcntl.flock(self.lock_fd, fcntl.LOCK_UN)
self.lock_fd.close()

def fetch_pending_tasks(self, limit: int = 5) -> List[Dict[str, Any]]:
"""从共享数据库获取待执行任务（分布式任务分发）"""
self._acquire_lock()
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 查询状态为pending的任务，限制数量（避免单实例抢占过多）
cursor.execute('''
SELECT id, file_path, use_ocr
FROM distributed_tasks
WHERE status = 'pending'
LIMIT ?
''', (limit,))
tasks = cursor.fetchall()
# 将任务标记为running（防止其他实例重复执行）
task_ids = [task[0] for task in tasks]
if task_ids:
cursor.execute(f'''
UPDATE distributed_tasks
SET status = 'running', worker_id = ?
WHERE id IN ({",".join(["?"]*len(task_ids))})
''', (os.getpid(), *task_ids))
conn.commit()
conn.close()
# 转换为任务列表格式
return [{"id": t[0], "file_path": t[1], "use_ocr": t[2]} for t in tasks]
finally:
self._release_lock()

def run_distributed_tasks(self) -> None:
"""执行分布式任务"""
# 循环获取待执行任务，直到无任务
while True:
pending_tasks = self.fetch_pending_tasks(limit=2)
if not pending_tasks:
break
# 执行任务
for task in pending_tasks:
execution_id = str(uuid.uuid4())
try:
# 执行工作流
state = PDFProcessState(
execution_id=execution_id,
file_path=task["file_path"],
use_ocr=task["use_ocr"],
workflow_status="running"
)
app = self._build_workflow()
final_state = app.invoke(state)
# 更新任务状态为completed
self._update_task_status(task["id"], "completed", execution_id)
except Exception as e:
# 更新任务状态为failed
self._update_task_status(task["id"], "failed", execution_id, error=str(e))

def _update_task_status(self, task_id: int, status: str, execution_id: str, error: str = "") -> None:
"""更新分布式任务状态"""
self._acquire_lock()
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
UPDATE distributed_tasks
SET status = ?, execution_id = ?, error_msg = ?, update_time = CURRENT_TIMESTAMP
WHERE id = ?
''', (status, execution_id, error, task_id))
conn.commit()
conn.close()
finally:
self._release_lock()

（三）Electron 前端分布式任务管理

tsx

import { useState, useEffect } from 'react';
import { Button, Table, Input, Upload, Space } from 'antd';

const DistributedTaskPanel = () => {
const [taskList, setTaskList] = useState([]);
const [loading, setLoading] = useState(false);

// 加载分布式任务列表
const fetchTasks = async () => {
setLoading(true);
const list = await window.electronAPI.getDistributedTasks();
setTaskList(list);
setLoading(false);
};

// 提交批量任务到共享数据库
const uploadTasks = async (files) => {
const taskList = files.map(file => ({
file_path: file.path,
use_ocr: true,
status: 'pending'
}));
await window.electronAPI.submitDistributedTasks(taskList);
fetchTasks();
};

// 启动当前实例的任务执行
const startWorker = async () => {
await window.electronAPI.startDistributedWorker();
fetchTasks();
};

useEffect(() => {
fetchTasks();
// 定时刷新任务状态
const timer = setInterval(fetchTasks, 5000);
return () => clearInterval(timer);
}, []);

const columns = [
{ title: '任务ID', dataIndex: 'id', key: 'id' },
{ title: '文件路径', dataIndex: 'file_path', key: 'file_path', ellipsis: true },
{ title: '执行状态', dataIndex: 'status', key: 'status' },
{ title: '执行实例ID', dataIndex: 'worker_id', key: 'worker_id' },
{ title: '错误信息', dataIndex: 'error_msg', key: 'error_msg', ellipsis: true },
];

return (
<Space direction="vertical" size="large" style={{ width: '100%' }}>
<Space>
<Upload directory multiple beforeUpload={(file) => false} onChange={({ fileList }) => uploadTasks(fileList)}>
<Button>上传批量PDF文件</Button>
</Upload>
<Button type="primary" onClick={startWorker}>启动本地执行器</Button>
</Space>
<Table columns={columns} dataSource={taskList} loading={loading} rowKey="id" />
</Space>
);
};

五、并发控制常见问题与解决方案