None: self.prompt += other.prompt self.completion += other.completion self.total += other.total ">
发布于2026-07-27 阅读(0)
扫一扫,手机访问
def add(self, other: "Usage") -> None:
self.prompt += other.prompt
self.completion += other.completion
self.total += other.total
self.cached += other.cached
self.calls += other.calls
self.cost += other.cost
使用:

total = Usage() total.add(usage1) total.add(usage2)
不会创建新对象,适合大量累加:
usage = Usage()
for item in usages:
usage.add(item)
避免大量临时 Usage() 对象。
例如:
a = Usage(prompt=100) b = a a.add(Usage(prompt=50)) print(b.prompt)
结果:
150
因为 a 和 b 指向同一个对象。
这可能导致隐藏 bug。
def __add__(self, other: "Usage") -> "Usage":
return Usage(
prompt=self.prompt + other.prompt,
completion=self.completion + other.completion,
total=self.total + other.total,
cached=self.cached + other.cached,
calls=self.calls + other.calls,
cost=self.cost + other.cost,
)
使用:
total = usage1 + usage2
例如:
total = usage1 + usage2 print(usage1)
usage1 不会改变。
这符合函数式思想:
输入不变,输出新值
支持链式计算
total = u1 + u2 + u3 + u4
result = Usage()
for u in usages:
result = result + u
实际上:
第1次:
Usage对象A + u1 -> B
第2次:
B + u2 -> C
第3次:
C + u3 -> D
例如:
usage = Usage(prompt=100) backup = usage usage.add(Usage(prompt=50)) print(backup.prompt)
结果:
150
很多人会以为:
backup = usage
是复制,但 Python 只是复制引用。
图示:
backup ─┐
├──> Usage(prompt=100)
usage ──┘
add() 修改的是这个对象本身。
例如:
def calculate_total(usage):
usage.add(Usage(prompt=100))
return usage
u = Usage(prompt=10)
result = calculate_total(u)
print(u.prompt)
输出:
110
例如你的场景:
类似:
global_usage.add(response_usage)
线程 A:
读取 prompt=100
+50
写回150
线程 B:
读取 prompt=100
+30
写回130
最终:
130
而不是:
180
为:
self.prompt += other.prompt
不是原子操作。
实际上等价:
tmp = self.prompt tmp = tmp + other.prompt self.prompt = tmp
多线程会丢更新。
解决: 加锁
from threading import Lock
lock = Lock()
with lock:
usage.add(other)
是的,add() 本身不是线程安全的.
不过重点是:当前这份代码里,add() 暂时没有并发问题。
每个线程里,各自跑自己的 Workflow
原因是现在的并发结构是这样的:
# 每个线程: run_source() -> Workflow() -> workflow.usage -> source_usage # 主线程: as_completed(futures) -> 拿到某个 source_usage -> merge_usage_map(global_usage, source_usage)
现在没有多个线程同时对同一个 Usage 对象执行:
usage.add(other)
所以当前代码的数据统计不会因为 add() 并发写而乱掉。
真正会出问题的是这种写法:(在最后全局统计时)
global_usage = {}
# 多个 worker 线程里同时执行
merge_usage_map(global_usage, workflow.usage)
两个线程同时加同一个模型时,可能出现丢计数。
现在的最后统计:
for future in as_completed(futures):
title, success, error, usage = future.result()
merge_usage_map(global_usage, usage)
虽然前面 5 个 worker 是并发跑的,但它们只是各自返回自己的usage,真正合并到 global_usage 的动作,是主线程一个 future 一个 future 地处理。
worker 1 跑完 -> 返回 usage
worker 2 跑完 -> 返回 usage
worker 3 跑完 -> 返回 usage
主线程:
合并 worker 2 的 usage
合并 worker 1 的 usage
合并 worker 3 的 usage
哪个任务先结束,就先汇总哪个;没结束的任务暂时不会汇总。
# 5 个并发任务启动
第 1 本完成 -> add 到 global_usage
第 3 本完成 -> add 到 global_usage
第 5 本还在跑 -> 暂时没 add
第 2 本还在跑 -> 暂时没 add
第 4 本还在跑 -> 暂时没 add
串行汇总? 会不因为前面的某个线程很慢,而卡住后面的线程
as_completed(futures)的特点是:哪个 future 先完成,就先把哪个吐出来。
同时跑:1 2 3 4 5
任务 3 先完成
-> 主线程立刻拿到任务 3 的 usage
-> merge_usage_map(global_usage, 任务3_usage)
-> 线程池空出一个位置,开始任务 6
任务 1 完成
-> 立刻 add 任务 1 的 usage
-> 开始任务 7
任务 5 很慢
-> 它没完成前不会 add 它自己的 usage
-> 但不会阻塞任务 1/2/3/4/6/7/... 的 add
只有一种等待:最后总函数返回前,会等全部 100 个 future 都完成。因为最终全局总合必须包含所有任务。
当前代码:没有并发问题。
因为全局合并发生在主线程。
add() 本身:不是线程安全的。
最稳的写法是保持现在这个架构:worker 只返回自己的 usage,主线程统一合并。
售后无忧
立即购买>office旗舰店
售后无忧
立即购买>office旗舰店
售后无忧
立即购买>office旗舰店
售后无忧
立即购买>office旗舰店
正版软件
正版软件
正版软件
正版软件
正版软件
1
2
3
7
8