范叶亮写智能体、模型和数据系统时，习惯先把定义和边界钉死。把「进程，线程和协程」整理成可落地的中文笔记：问题在哪、默认做法会踩什么坑、该怎么选。原站导航和广告已去掉。

进程和线程

理论篇请参见： 进程，线程和协程 (Process, Thread and Coroutine) – 理论篇

本文将介绍进程，线程和协程在 Python 中的实现，代码详见 这里 ，部分参考自「Python 并发编程」 1 :。

利用函数创建线程

在 Python 中可以使用 multiprocessing.Process 和 threading.Thread 来实现进程和线程。我们采用 CPU 密集型 、 磁盘 IO 密集型 、 网络 IO 密集型 和 模拟 IO 密集型 任务类型来测试单线程，多线程和多进程之间的性能差异。

import requests # CPU 密集型 def cpu_bound_task ( x = 1 , y = 1 ): c = 0 while c < 1500000 : c += 1 x += x y += y # 磁盘 IO 密集型 def disk_io_bound_task (): with open ( 'tmp.log' , 'w' ) as f : for idx in range ( 5000000 ): f . write ( ' {} \n ' . format ( idx )) # 网络 IO 密集型 def web_io_bound_task (): try : requests . get ( 'https://www.baidu.com' ) except Exception as e : pass # 模拟 IO 密集型 def simulation_io_bound_task (): time . sleep ( 2 ) 为了方便统计运行时间，定义如下一个运行时间装饰

利用类创建线程

import time def timer ( task_mode ): def wrapper ( func ): def decorator ( * args , ** kwargs ): task_type = kwargs . setdefault ( 'task_type' , None ) start_time = time . time () func ( * args , ** kwargs ) end_time = time . time () print ( '耗时（ {} – {} ）: {} ' . format ( task_mode , task_type , end_time – start_time )) return decorator return wrapper 单线程，多线程和多进程的测试代码如下：

from threading import Thread from multiprocessing import Process @timer ( '单线程' ) def single_thread ( func , task_type = '' , n = 10 ): for idx in range ( n ): func () @timer ( '多线程' ) def multi_threads ( func , task_type = '' , n = 10 ): threads = {} for idx in range ( n ): t = Thread ( target = func ) threads [ idx ] = t t . start () for thread in threads . values (): thread . join () @timer ( '多进程' ) def multi_processes ( func , task_type = '' , n = 10

锁

# 单线程 single_thread ( cpu_bound_task , task_type = 'CPU 密集型任务' ) single_thread ( disk_io_bound_task , task_type = '磁盘 IO 密集型任务' ) single_thread ( web_io_bound_task , task_type = '网络 IO 密集型任务' ) single_thread ( simulation_io_bound_task , task_type = '模拟 IO 密集型任务' ) # 多线程 multi_threads ( cpu_bound_task , task_type = 'CPU 密集型任务' ) multi_threads ( disk_io_bound_task , task_type = '磁盘 IO 密集型任务' ) multi_threads ( web_io_bound_task , task_type = '网络 IO 密集型任务' ) m

Python 中的 threading.Thread() 接受两个参数： 线程函数 ，用于指定线程执行的函数； 线程函数参数 ，以元组的形式传入执行函数所需的参数。

互斥锁

import time from threading import Thread # 自定义函数 def func ( name = 'Python' ): for idx in range ( 2 ): print ( 'Hello, {} ' . format ( name )) time . sleep ( 1 ) # 创建线程 thread_1 = Thread ( target = func ) thread_2 = Thread ( target = func , args = ( 'Leo' , )) # 启动线程 thread_1 . start () thread_2 . start () 可以得到如下输出：

Hello, Python Hello, Leo Hello, Python Hello, Leo 利用类创建线程 利用类创建线程需要自定义一个类并继承 threading.Thread 这个父类，同时重写 run 方法。最后通过实例化该类，并运行 start() 方法执行该线程。

可重入锁

# 自定义类 class MyThread ( Thread ): def __init__ ( self , name = 'Python' ): super ( MyThread , self ) . __init__ () self . name = name def run ( self ): for idx in range ( 2 ): print ( 'Hello, {} ' . format ( self . name )) time . sleep ( 1 ) # 创建线程 thread_1 = MyThread () thread_2 = MyThread ( 'Leo' ) # 启动线程 thread_1 . start () thread_2 . start () 可以得到同上面一样的输出：

Hello, Python Hello, Leo Hello, Python Hello, Leo 线程的一些常用方法和属性如下所示：

值得单独记下的点

• 多线程和多进程相比单线程速度整体上有很大提升。
• 对于 CPU 密集型任务，由于 GIL 加锁和释放问题，多线程相比单线程更慢。
• 多线程更适合在 IO 密集场景下使用，例如：爬虫等。
• 多进程更适合在 CPU 密集场景下使用，例如：大数据处理，机器学习等。
• GEN_RUNNING ：正在执行（只在多线程应用中能看到该状态）
• 委托生成器：包含 yield from 表达式的生成器函数
• 子生成器： yield from 后面的生成器函数
• event_loop 事件循环 ：程序开启一个无限的循环，协程将注册到事件循环上，当满足事件发生时，调用相应的协程函数。

落地时建议先做的 5 件事

1. 先写清任务能不能被自动验证：能验证的交给系统和评测，不能验证的留给人审。
2. 本地部署先算显存、延迟和失败回滚，不要只看能跑通一次。
3. 多智能体只在单智能体触到上下文或专业边界时再拆。
4. Token、微调和压缩都要有对照数字，避免口号式优化。
5. 结论写成可检查清单：接口、超时、评测集、回滚版本。

和智能体产品怎么接

龙虾PRO做 OpenClaw 落地时，最该拿走的是「单智能体先做好工具和提示，再谈编排」。数字员工、技能市场和网关应共用同一套评测与权限，而不是各写一套角色人设。

本文侧重全链路风控方法论。落地时请用自身业务单据做回放验证，不要把示例阈值直接当生产策略。 相关：风控体检 · 方案资源