python中add()和__add__()的用法小结

先看区别：

方案 1：add()原地修改

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。

方案 2：__add__()返回新对象

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

现在我们看原地更改的问题： 可以直接看3.

1. 引用共享导致意外修改（复制引用）

例如：

usage = Usage(prompt=100)

backup = usage

usage.add(Usage(prompt=50))

print(backup.prompt)

结果：

150

很多人会以为：

backup = usage

是复制，但 Python 只是复制引用。

图示：

backup ─┐
        ├──> Usage(prompt=100)
usage ──┘

add() 修改的是这个对象本身。

2. 作为函数参数传递时 （更改全局状态）

例如：

def calculate_total(usage):
    usage.add(Usage(prompt=100))
    return usage

u = Usage(prompt=10)

result = calculate_total(u)

print(u.prompt)

输出：

110

3. 多线程环境下可能出现数据竞争

例如你的场景：

• 多个 LLM 请求同时完成
• 都更新一个全局 Usage

类似：

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() 本身不是线程安全的.

不过重点是：当前这份代码里，add() 暂时没有并发问题。

每个线程里，各自跑自己的 Workflow

原因是现在的并发结构是这样的：

• 每个并发任务都会在 run_source() 里创建自己的 Workflow
• 每个 Workflow 有自己的 self.usage
• source_usage 也是每个任务自己的局部变量
• global_usage 只在主线程的 for future in as_completed(futures) 里合并

# 每个线程：
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，主线程统一合并。