[!info] related notes

• 所属 MOC:
• 相关概念:
• 易混淆概念:
• 相关资源:

BodySense Embedding 的 Async 执行边界

核心结论

“生成 Embedding”是一个业务能力名，不代表只有一种执行模型。

BodySense 实际支持的路径至少可以分成：

EmbeddingGenerator
├─ remote OpenAI-compatible API
│    └─ network I/O
├─ local SentenceTransformer
│    └─ synchronous CPU/GPU computation
└─ deterministic hashing
     └─ local CPU computation

因此不能给三者套同一种“async”处理方式。

更准确的设计是：

对外保持统一 async capability，对内根据 provider 的真实资源属性选择 native async I/O、bounded blocking executor 或轻量 inline CPU。

为什么统一 async def generate() 容易让人误判

接口：

async def generate(self, text: str) -> list[float]: ...
async def generate_batch(self, texts: list[str]) -> list[list[float]]: ...

看起来所有 provider 都是 async。

但 local branch：

if self.provider == "local_transformer":
    model = self._get_local_model()
    embedding = model.encode([text])[0]

model.encode() 是同步调用。

调用链实际：

Agent / KnowledgeLibrary coroutine
→ await EmbeddingGenerator.generate()
    → coroutine starts on Event Loop
    → model.encode() runs synchronously
    → only returns after compute finishes

外层 await 不会自动把 encode() 移到别的线程。

所以：

async capability interface
≠
all provider implementations are non-blocking

Remote Embedding：典型 Async I/O

Remote provider：

response = await self.client.embeddings.create(
    model=self.model,
    input=texts,
)

主要时间在：

serialize request
→ send network
→ remote service computes
→ wait socket
→ receive response

Async HTTP client 能把 network wait 交还 Event Loop。

这类路径的工程重点：

• async connection reuse；
• timeout；
• retry/backoff；
• rate limit；
• cancellation；
• provider error taxonomy。

不应该为了抽象统一，再把它塞进 thread executor。

Local Transformer：同步计算边界

SentenceTransformer.encode() 可能经历：

tokenization
→ tensor preparation
→ transformer forward
→ pooling
→ normalization/materialization

真实计算可能发生在：

• Python；
• PyTorch native code；
• CPU BLAS/OpenMP threads；
• CUDA GPU。

无论底层是否释放 GIL，对 asyncio coroutine 来说它仍是同步调用：返回前没有自然的 await suspend point。

因此首先要保护：

Event Loop responsiveness

最窄修复：Blocking Encode 移出 Loop Thread

示意：

embeddings = await asyncio.to_thread(model.encode, texts)

它表达：

loop thread
→ submit synchronous encode
→ await Future
→ continue scheduling unrelated Tasks

这样：

local encode slow
≠
entire worker event loop freezes

但这只是第一层修复。

为什么还必须 Bounded Concurrency

假设 30 个请求同时：

await asyncio.to_thread(model.encode, texts)

没有显式限制时可能出现：

30 encode work items
× PyTorch / BLAS internal threads
× tensor allocations
× GPU memory

后果：

• CPU oversubscription；
• context switch 暴增；
• GPU kernel/memory contention；
• CUDA OOM；
• thread pool 被占满；
• tail latency 全部变差。

所以更完整：

incoming requests
     ↓
async Semaphore / dedicated bounded executor
     ↓
limited local encode concurrency
     ↓
SentenceTransformer

Semaphore 为什么不会再次“堵 Event Loop”

如果令牌用完：

await semaphore.acquire()

当前 coroutine suspend，Event Loop 仍可运行其他任务。

这叫：

backpressure

而不是：

thread blocking

生产系统需要允许有限排队，并结合 timeout/metrics 判断容量是否足够。

为什么默认 ThreadPool 上限不等于业务容量策略

asyncio.to_thread 使用 executor；executor 本身会有 worker 数限制。

但仅依赖默认上限仍然不够表达：

Embedding capacity = ?

原因：

• executor 可能与其他 blocking work 共用；
• 一个 encode 内部又会启动 native threads；
• GPU capacity 与 OS thread 数不是一个概念；
• 需要可配置、可观测的 resource budget。

所以独立 Semaphore 或 dedicated executor 更清楚地表达：

local_embedding_concurrency = N

Lazy Model Initialization 是第二个并发边界

旧模式：

def _get_local_model(self):
    if self._local_model is None:
        self._local_model = SentenceTransformer(self.model)
    return self._local_model

顺序调用没问题。

但如果多个 worker thread 在首次访问时同时看到：

self._local_model is None

可能：

Thread A loads model
Thread B also loads model

导致：

• 重复读取/初始化；
• 重复 GPU allocation；
• duplicate warm-up；
• dimension 并发修改。

所以把 encode 移到 thread 后，lazy init 必须重新审查线程安全。

初始化内部的 encode(["test"]) 也可能 Block

如果 _get_local_model() 为了探测 dimension：

test_embedding = self._local_model.encode(["test"])

那么即使正式 generate() 已经 offload，首次 model init 如果仍发生在 Event Loop thread，第一次请求照样会 block。

所以 async review 要追：

model construction
warm-up
dimension probing

不能只搜正式 generate_batch()。

Preload vs Thread-safe Lazy Load

两种常见策略：

Startup Preload

application startup
→ load model once
→ warm up
→ mark ready

优点：

• 首请求不会承担冷启动；
• initialization failure 早发现。

缺点：

• startup 慢；
• 即使 provider 暂时不用也占内存。

Lock-protected Lazy Init

first local request
→ one initializer
→ others await/wait

优点：按需。

缺点：首请求 latency 高，implementation 更复杂。

选择取决于：

• local provider 是否 production 常用；
• model size；
• deployment memory；
• cold-start SLO。

Hashing Embedding：不要机械 Offload

Hashing fallback 也做 CPU 工作：

• normalize；
• n-gram；
• SHA-256；
• bucket accumulation；
• vector normalization。

但它通常比 transformer 轻得多。

不应该写规则：

CPU work → 一律 to_thread

因为 thread queue/Future/context switch 也有成本。

更合理：

benchmark representative input/batch
→ observe latency + heartbeat gap
→ if negligible: inline
→ if material: offload via same blocking adapter

这体现：

异步工程目标是控制可观察的 blocking/capacity，不是让代码表面每个分支都有 await。

Thread、Process、Model Worker、Durable Job 怎么选

Thread / to_thread

适合：

• 同步库；
• 单次 relatively short；
• 需要共享一个已加载模型；
• 首要目标是保护 Event Loop。

Process Pool

适合：

• 重纯 Python CPU；
• 多核并行价值高；
• serialization / model duplication 可接受。

风险：

• 每进程模型副本；
• GPU context 更复杂；
• IPC overhead。

Dedicated Embedding Service/Worker

适合：

• Embedding 负载变大；
• 多应用共享；
• GPU batching；
• 独立扩缩容。

Durable Job Runtime

适合：

• 很长；
• 用户不需要当前请求内结果；
• 需要 retry/recovery/progress；
• OCR / batch ingestion / posture analysis 等。

不应因为：

local encode = 300ms

就把每次 targeted search 变成后台 Job。

Batching 的吞吐 / 延迟 Trade-off

Transformer 往往 batch 越大吞吐越好：

model.encode(texts)

但 interactive retrieval 通常只有 1 条 query。

如果为了 batch：

wait 20ms collect requests
→ one GPU batch

可能提升 throughput，但增加单请求 queue latency。

是否 micro-batching 应由 SLO/throughput 决定，不是“AI 模型就应该 batch”。

Cancellation 的现实边界

当 coroutine：

await asyncio.to_thread(model.encode, texts)

被取消，外层可以停止等待，但已运行的 worker thread 通常不能被 Python 安全强制终止。

所以：

coroutine cancelled
≠
model.encode instantly stopped

这意味着 local blocking work 必须：

• bounded；
• 单次 finite；
• 避免不可重复 side effect；
• 真正长任务使用 worker/job model。

Timeout 也不等于 Worker Stop

await asyncio.wait_for(to_thread(...), timeout=0.5)

超时后 caller 可以继续，但底层 thread 可能还在跑。

所以如果本地 encode 经常超过 request budget，真正方案不是无限加 timeout wrapper，而是：

• 降低 workload；
• 限制并发；
• 预热；
• dedicated service；
• workload redesign。

Dimension 是数据 Contract，不只是 Model 属性

pgvector schema 的 vector dimension 与 embedding provider/model 必须一致。

把执行从 loop thread 移到 worker thread，不应该改变：

vector count
vector ordering
dimension
normalization semantics

如果切换 model 导致：

384 → 768

这不是简单 execution refactor，而是：

knowledge index/schema migration

可能要求重新 embedding 全库。

所以 provider execution 与 embedding semantic identity 要分开。

Remote Provider Retry 不应被 Local Executor 抽象误伤

当前 remote path 可以：

native async HTTP
→ async backoff
→ retry

不要为了做一个“统一 blocking adapter”把 remote call 也丢到 thread。

好的 abstraction 应允许：

Provider Strategy
├─ AsyncRemoteEmbedding
├─ BoundedLocalEmbedding
└─ InlineHashingEmbedding

而不是把所有 provider 降级成最低公分母同步接口。

Error Taxonomy

至少区分：

provider auth/rate limit
remote timeout
local model load failure
local encode failure
executor queue timeout
concurrency budget timeout
dimension mismatch

上层 EvidenceAttempt 看到：

embedding failed

最好还能定位 resource/provider class，而不是被误解释成：

knowledge base had no matching evidence

Metrics

Local embedding 至少值得观察：

queue_wait_ms
encode_ms
concurrency_inflight
semaphore_waiters
model_load_ms
error_total
timeout_total

如果 GPU：

GPU memory/utilization

这样才能决定：

• concurrency limit 太小？
• encode 真慢？
• 大部分时间在排队？
• 冷启动问题？

测试矩阵

Thread Identity + Heartbeat

Fake model.encode 阻塞 200ms：

• encode thread != loop thread；
• heartbeat 仍持续调度。

Concurrency Cap

Fake 记录 simultaneous executions：

max_observed <= configured_limit

Lazy Init

多个并发首次调用：

model construction count == 1

或符合明确的 preload design。

Result Parity

Offload 前后 fake model：

same input
→ same vector order/value/dimension

Remote Isolation

remote provider 仍然直接 await async client，不意外进入 thread path。

Hashing Benchmark

用代表性 text/batch 证明 inline 不制造显著 heartbeat gap；如果失败再 offload。

Timeout/Cancellation

caller timeout 后：

• outer coroutine returns/propagates correctly；
• concurrency permit 最终释放；
• no leaked permanently-held semaphore。

Model Init Failure

startup/lazy initialization error 不应留下“半初始化但被标记 ready”的 model object。

常见误区

“CPU-bound 一定 ProcessPool”

过度绝对。to_thread 的首要目标可以只是保护 Event Loop；底层 native/GPU 情况也不等同纯 Python CPU。

“用了 GPU 就不会堵 Event Loop”

错。Python 调用、数据准备、同步等待都可能让 model.encode() 长时间不返回。

“ThreadPool 会自动帮我做正确限流”

默认 worker 数不是 Embedding 业务容量策略。

“请求 cancel 后 thread 也 cancel”

通常不成立。

“Hashing 也是 CPU，所以必须线程化”

先测，再决定。

“换 Embedding Model 只改配置”

如果 dimension/semantic space 变化，往往需要重新构建知识向量索引。

自测题

1. 为什么 remote API 与 local transformer 虽然都叫 Embedding，却需要不同 execution model？
2. await generate() 为什么不能证明 model.encode() 已 offload？
3. to_thread 为什么还要 Semaphore/bounded executor？
4. Lazy model initialization 在 offload 后会出现什么新 race？
5. 为什么 _get_local_model() 内的 warm-up encode 也要审查 blocking？
6. Hashing provider 为什么应该 benchmark-driven，而不是形式统一地 offload？
7. Thread、Process、Dedicated Service、Durable Job 分别适合什么？
8. coroutine cancel 为什么不等于同步 worker function 被强杀？
9. Embedding dimension 变化为什么属于数据 migration，而不只是 runtime refactor？
10. 如何用 heartbeat + concurrency-cap + parity tests 共同证明新边界正确？

最终不变量

Remote wait → native async
Local heavy encode → bounded executor/thread
Light CPU → benchmark-driven inline/offload
Model init → concurrency-safe
Cancellation → caller can stop waiting, worker semantics understood
Result → dimension/order stable
Capacity → observable and bounded

学到这里，重点已经从“Python 能不能 await EmbeddingGenerator”升级成：

每种 provider 真正消耗什么资源、在哪执行、由谁限流、取消能做到什么，以及如何证明异步重构没有改变 Embedding 的业务/数据语义。