[!info] related notes

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

BodySense Python Async / RAG Engineering

L4 的定位：不是重新背 async/await

L4 的目标不是再讲一遍 async def、await、gather 的语法，而是把 BodySense 已经运行在 FastAPI / LangGraph / Agent runtime 中的真实 RAG 数据路径当成一条生产执行链来审查：

1. 这段代码真的会把执行权还给 Event Loop 吗？
2. DB 连接、线程池、Embedding 模型与 Evidence 的生命周期由谁拥有？
3. Retrieval 返回的数据怎样变成可追溯、可采纳、可验证的 Evidence？
4. 模型最终提出的 Diagnosis/Treatment claim，怎样证明真的被 Evidence 支持？

所以 L4 是前面三层的工程地基：

L1 Diagnosis
→ Agent reasoning / Evidence / Governance / Authority

L2 Treatment
→ Proposal / Acceptance / Intervention / Outcome

L3 Consultation
→ Multi-turn Runtime / Streaming / Replay / HITL

L4 Async / RAG
→ DB / Embedding / Retrieval / Evidence / Grounding 的资源与执行正确性

通用 asyncio 语法先看 Python 异步编程 与 asyncio 任务、超时与取消；本篇只学习 BodySense 的差分。

第一条核心不等式：async def != non-blocking execution

最容易产生的误解：

async def search(...):
    ...

看到 async def 就认为函数已经“异步化”。

但 Event Loop 真正在意的是：

当前 Task 从一次可调度点恢复以后，到下一次真正 suspend 之前，在 loop thread 上运行了什么？

BodySense 审查时的旧 KnowledgeLibrary.search() 正好是典型反例：

async def search(...):
    embedding = await self.embedding_generator.generate(query)

    conn = self._get_connection()   # sync psycopg.connect
    with conn.cursor() as cur:      # sync cursor
        cur.execute(...)            # sync database I/O
        rows = cur.fetchall()

外层是 async API，但数据库段仍然是同步 I/O。

同样：

async def generate(...):
    embedding = model.encode([text])[0]

SentenceTransformer.encode() 是同步 CPU/GPU 工作，也不会因为外层写了 async def 就自动移出 Event Loop。

因此：

async API
≠ async implementation
≠ non-blocking execution

详见 BodySense Event Loop Blocking 边界。

从一条真实 RAG 路径看工作负载分类

Consultation / Diagnosis / Treatment Agent
        │
        │ EvidenceGap / search_knowledge
        ▼
KnowledgeLibrary.search()
        │
        ├─ Query Embedding
        │    ├─ remote API → network I/O
        │    ├─ local transformer → sync CPU/GPU
        │    └─ hashing → local CPU
        │
        ├─ PostgreSQL / pgvector query
        │    └─ database I/O
        │
        ├─ rerank / normalize
        ▼
Retrieved Evidence candidates
        │
        ├─ provenance/admissibility
        └─ grounding/faithfulness

不同工作应该使用不同 execution model：

工作	本质	L4 方向
Remote Embedding API	网络 I/O	native async client
PostgreSQL 查询	网络 I/O + DB wait	AsyncConnectionPool + async cursor
Local Transformer	CPU/GPU 同步计算	bounded thread/executor
Hashing Embedding	本地 CPU	benchmark 后决定 inline/offload
轻量 normalization/rerank	本地 CPU	通常 inline，仍看规模

关键不是“所有东西都改成 async”，而是识别资源属性，再放到正确的执行边界。

Event Loop 为什么会被一个局部 blocking section 拖住

在一个典型 asyncio worker 里，许多 Task 共用同一个 loop thread 来执行 Python control flow：

Task A: send HTTP → await ──────────┐
Task B: DB query   → await ─────────┤
Task C: SSE work   → await ─────────┤
                                     ▼
                              Event Loop schedules

当 A/B/C 都在等待真正的异步 I/O 时，CPU 可以去运行其他 ready Task。

但如果某个 Task 进入：

sync psycopg execute 400ms

或：

SentenceTransformer.encode 900ms

这段时间受影响的不只是“这个搜索慢一点”，而是同一个 loop 上：

• 另一个 Agent run；
• SSE progress；
• health endpoint；
• interrupt/resume；
• timeout/cancellation 回调；

都可能得不到及时调度。

因此 async correctness 是并发系统正确性，不只是单请求 latency 优化。

L4.1 Database Boundary：从同步单连接到 AsyncConnectionPool

旧模型近似：

process
  └─ global KnowledgeLibrary
       └─ one sync psycopg.Connection
            └─ async methods 包 sync cursor

问题包括：

• DB wait 阻塞 Event Loop；
• 一个长期连接隐藏故障状态；
• 多个 concurrent search 的 resource semantics 模糊；
• lazy connect 把依赖可用性推迟到第一位用户请求；
• transaction / connection lifetime 混在业务对象里。

North-Star：

FastAPI lifespan
  └─ AsyncConnectionPool
       ├─ search A checkout connection
       ├─ search B checkout connection
       └─ ingestion checkout transaction connection

示意：

async with self._pool.connection() as conn:
    async with conn.cursor() as cur:
        await cur.execute(...)
        rows = await cur.fetchall()

这里 Pool 长期存在，connection 只是短 lease。

详见 BodySense KnowledgeLibrary 的 Async Postgres Pool。

Pool、Connection、Transaction 是三个不同生命周期

Pool
= application lifespan resource

Connection lease
= one DB operation / short unit of work

Transaction
= one domain atomic operation

Search 通常只需要短只读 lease；Ingestion 却可能要求：

knowledge source
+ segments
+ units
+ clips

作为一个 intentional transaction。

异步重构不能为了“每个 await 都很漂亮”把原本业务原子性拆碎。

[!important] Execution model 可以从 sync 改成 async；domain atomicity 不能因此消失。

Pool Exhaustion 是 Backpressure，不是 Event Loop Blocking

如果 pool max=5，同时 20 个搜索：

5 个获得 connection
15 个 await pool availability

后 15 个 coroutine 会排队，但不会占住 loop thread。

这说明 async 并没有制造无限资源，只是把等待方式改成 cooperative。

所以还需要：

• acquire timeout；
• pool usage metrics；
• query latency；
• capacity budget。

non-blocking
≠ infinite throughput

L4.2 Embedding Boundary：同一个业务能力，不同执行模型

BodySense 的 EmbeddingGenerator 对外是统一 async capability，但内部至少三种 provider：

remote API
local SentenceTransformer
hashing fallback

它们不应该被最低公分母地统一成同一种执行方式。

Remote

await AsyncOpenAI(...).embeddings.create(...)

适合 native async network I/O。

Local Transformer

model.encode(texts)

是同步 CPU/GPU 边界，应从 Event Loop thread 移出：

await asyncio.to_thread(model.encode, texts)

但只做 to_thread 还不够。

为什么 to_thread 仍然需要并发上限

如果 40 个请求同时触发：

40 × model.encode

可能产生：

• CPU oversubscription；
• PyTorch/BLAS 内部线程叠加；
• GPU contention；
• OOM；
• thread pool saturation；
• tail latency 暴涨。

所以更完整：

requests
  ↓
Semaphore / bounded executor
  ↓
limited local encode concurrency

Semaphore.acquire() 的排队是 awaitable backpressure；它不会像同步 encode 那样堵 loop。

详见 BodySense Embedding 的 Async 执行边界。

Lazy Model Initialization 也属于 Blocking / Concurrency Boundary

如果：

if self._local_model is None:
    self._local_model = SentenceTransformer(...)
    self._local_model.encode(["test"])

首次加载本身就可能很重，而且多个 worker thread 同时首次进入可能重复加载模型/GPU memory。

所以 offload generate() 后仍要继续追调用链：

model construction
warm-up
dimension probing

所有重工作都要有明确 owner 和并发语义。

Thread、Process、Durable Job 不解决同一个问题

机制	更适合
native async I/O	DB、HTTP、socket
to_thread / bounded executor	短同步 SDK、本地 encode、保护 Event Loop
Process Pool	重纯 Python CPU、多核隔离
Dedicated model worker	高负载共享模型/GPU batching
Durable Job Runtime	OCR、批 ingestion、长任务、需要 recovery/progress

不能因为“local encode 会 block”就把每一次 interactive RAG search 变成后台 Job。

L4 的原则是：

先用最窄的 execution adapter 修正资源边界；只有生命周期真的要求 durable background execution 时才上 Job。

L4.3 Targeted Retrieval：RAG 不应该退回 Broad Preloading

L1 已经建立：

Decision-Relevant EvidenceGap
→ targeted acquisition
→ EvidenceAttempt
→ Evidence

L4 负责把这条语义真正落到 retrieval plumbing，而不是退回：

每次请求
→ broad search
→ top-k 全拼进 rag_context
→ model 自己找重点

Targeted Retrieval 的核心不是换向量数据库，而是调用语义改变：

不是“RAG 应用所以先搜”
而是“这个明确 Gap 值得从 external knowledge 获取证据，所以搜”

这样每次 search 都有：

• gap identity；
• rationale；
• source kind；
• budget；
• stopping reason；
• provenance。

详见 BodySense Targeted RAG 与 Evidence Provenance。

外部 RAG 不能补成 User Fact

这是 L1 与 L4 之间最重要的一条 epistemic boundary。

知识库说：

“某类神经问题可能伴随小腿外侧放射痛。”

不能因为搜到这段，就把当前用户状态写成：

user has lateral calf radiating pain

正确：

External knowledge:
这种表现与某机制有关

User fact:
当前是否存在仍 UNKNOWN

Gap:
需要 ask_user / user-owned observation

所以 retrieval 能补知识，不会凭空补用户事实。

三层 Evidence State

Retrieved Evidence
= Retriever 找到了

Admissible Evidence
= 当前 source/version/fact-type policy 允许使用

Gap Resolved
= 语义上已经足够解决该 uncertainty

必须记住：

retrieved
≠ admissible
≠ sufficient
≠ resolved

这让“搜到相似文章”不会被错误升级成“已经有充分决策依据”。

Citation 与 Provenance 不一样

Citation 更偏用户解释：

“这条建议参考了哪里？”

Provenance 更偏系统审计：

“本轮实际检索到的是哪个 evidence_id、哪个 source version、为哪个 gap、在哪个 run？”

漂亮的 citation 卡片不自动等于可 replay 的 provenance。

Normalized Evidence Contract

Retriever raw result 可以变化，但 Agent/Governance 层最好消费稳定 Evidence contract：

Evidence
├─ evidence_id
├─ source identity
├─ source version/snapshot
├─ observed excerpt/content
├─ acquired_for_gap_id
├─ attempt identity
├─ score/rank（若决策需要）
└─ policy-relevant metadata

这样以后从 pure vector search 换 hybrid retrieval，治理层不需要跟着 raw payload 重写。

L4.4 Grounding：有 Citation 还不够

早期 MVP Faithfulness Checker 主要做：

Treatment exercise title
→ substring / alias match in RAG results

它能挡住一部分明显幻觉，但不能证明完整 intervention claim。

例如 Evidence 只说：

“臀桥可以作为基础训练动作。”

模型却输出：

臀桥，每天 10 组，每组 100 次，疼痛加重也继续

动作名出现过，但：

• dosage；
• frequency；
• stop condition；

并没有被 Evidence 支持。

所以 production grounding 的评估单元应该是 material claim：

InterventionClaim
├─ kind/title
├─ goal
├─ dosage/frequency/duration
├─ progression
├─ stop conditions
└─ supporting evidence IDs

详见 RAG Grounding / Faithfulness 校验。

Grounding 的三层 Evaluator

Layer 1
Deterministic provenance/contract
  ↓
Layer 2
Structured semantic support / contradiction
  ↓
Layer 3
Optional LLM Judge for uncertain cases

Layer 1 必须机器直接验证：

• evidence ID 是否本轮实际 observed；
• 是否 cross-run；
• source/version 是否 admissible；
• user_fact 来源是否合法；
• evidence budget/tool policy 是否被绕过。

不要让 LLM Judge 来判断这些 runtime 已经确定知道的事实。

为什么 Embedding Similarity 不等于 Grounding Support

Evidence: “疼痛加重时停止臀桥”
Claim:    “疼痛加重时继续臀桥”

两段文本 embedding 可能非常相似，但语义互相矛盾。

所以：

semantic similarity
≠ entailment
≠ support

Embedding 可以做候选信号，不能单独成为 production truth gate。

L4.5 Lifespan Ownership

资源最好显式归 Application Lifespan：

FastAPI startup
├─ open checkpoint pool
├─ open KnowledgeLibrary async pool
├─ initialize long-lived clients/models as designed
└─ readiness

serve requests
└─ acquire/release short-lived leases

shutdown
├─ stop accepting work
├─ close pools
└─ drain owned resources

而不是：

first search request
→ global singleton silently connect/load

显式 lifespan 的价值：

• startup failure 可见；
• readiness 可测；
• shutdown 可测；
• 测试可注入 fake/test resources；
• ownership 清楚。

L4.6 Async 重构必须保护 Cancellation / Timeout

当远端 API、pool acquire、DB query 都变成 awaitable 后，仍然需要：

request budget
→ downstream timeout
→ cancellation propagation

否则只是从“阻塞”变成“可以无限 await”。

尤其：

await asyncio.to_thread(blocking_fn)

外层 coroutine 被取消，不等于 worker thread 中已经运行的同步函数能被安全强杀。

因此 blocking task 应：

• bounded；
• 单次有限；
• 避免不可重复副作用；
• 过长则升级到更适合的 Job/worker 模型。

L4.7 Heartbeat Test：验证行为，不相信函数名

只检查：

函数是不是 async

没有价值。

更好的测试：

Task A: 执行被怀疑 block 的 DB / encode stub
Task B: 每 10ms await sleep + record heartbeat

如果 A 直接在 loop thread：

heartbeat —— 300ms gap —— heartbeat

如果 A 使用 native async / offload：

heartbeat heartbeat heartbeat ...

真正锁住的 contract 是：

被测工作进行时，无关 lightweight Task 仍可被 Event Loop 调度。

L4 测试矩阵

Database

• pool startup/shutdown exactly once；
• concurrent search checkout independent leases；
• no sync psycopg.Connection in interactive path；
• async pgvector registration；
• ingestion mid-failure rolls back whole logical pack；
• startup/acquire failure bounded by timeout。

Embedding

• local encode runs off loop thread；
• heartbeat remains responsive；
• concurrency limiter caps simultaneous encode；
• lazy initialization concurrency-safe；
• vector dimension/order unchanged；
• remote provider remains native async。

Retrieval / Evidence

• query linked to explicit Gap；
• user-information gap cannot generic RAG；
• returned IDs become runtime provenance；
• inadmissible evidence cannot close critical gap；
• budget exhausted stops further retrieval but does not mark resolved。

Grounding

• exact supported action；
• synonym action；
• action supported but dosage unsupported；
• explicit contradiction；
• same body part but wrong intervention；
• no evidence；
• cross-run evidence ID；
• Chinese short-token false positive；
• stop-condition / progression support。

L4 与 L1 Failure Attribution 的连接

假设 critical EvidenceGap 最终 unresolved。

表面可能像：

“模型没找到答案。”

但 Execution Provenance / Attempt trace 可能显示：

local embedding blocked loop
→ search timed out
→ budget consumed
→ Gap unresolved

于是第一个 contract violation 在 async resource boundary，而不是 semantic reasoning。

所以 L4 不是孤立“性能课”。它直接进入：

Evidence quality
→ SafetyEnvelope
→ DecisionAuthority

L4 与 L2 Treatment 的连接

Treatment Grounding 依赖：

valid Evidence
+ stable provenance
+ complete material claim

如果 retrieval 只是 broad context、Evidence ID 不稳定、source version 不可追溯，那么：

Treatment qualification / replay / Outcome audit

都无法真正解释一个 intervention 为什么被允许进入 proposal。

因此 RAG Engineering 是 action governance 的组成部分。

L4 与 L3 Consultation 的连接

Consultation 的 async runtime 还有：

• streaming；
• interrupt/resume；
• durable recovery；

如果 local encode / sync DB block loop，就可能出现：

SSE stall
interrupt response delayed
recovery status delayed
health endpoint timeout

所以 L3 的 runtime correctness 依赖 L4 的 event-loop health。

一组值得直接记忆的不等式

async def
!= non-blocking execution

await
!= guaranteed scheduling yield

to_thread
!= unlimited safe parallelism

non-blocking
!= infinite capacity

retrieved evidence
!= admissible evidence

admissible evidence
!= gap resolved

citation exists
!= claim grounded

embedding similarity
!= support

budget exhausted
!= evidence sufficient

accepted Treatment
!= actually executed Intervention

学完 L4 应该能独立回答的问题

1. 为什么同步 psycopg 写在 async def 里仍会阻塞整个 Event Loop？
2. Remote Embedding API 与 local SentenceTransformer.encode() 为什么需要不同并发模型？
3. to_thread 解决什么，为什么还要 Semaphore / bounded executor？
4. Pool、Connection lease、Transaction 的 owner 分别是什么？
5. 为什么 async pool wait 是 backpressure，不是 thread blocking？
6. 为什么 interactive targeted retrieval 不应该为了“异步化”就全部改成后台 Job？
7. 为什么外部 RAG 不能补成 user fact？
8. retrieved / admissible / resolved 为什么必须三层分开？
9. Citation 与 Provenance 分别回答什么？
10. 为什么动作名被 Evidence 提到仍不足以证明 dosage/stop condition grounded？
11. 哪些 grounding check 必须 deterministic，哪些才适合 semantic matcher / Judge？
12. 怎样用 heartbeat test 证明一段工作没有堵 Event Loop？
13. 为什么 async 重构仍要保留 ingestion transaction atomicity？
14. 为什么 Execution Provenance 能帮助区分“模型不知道”和“基础设施没完成取证”？

L4 完成标准

不是机械检查“代码里没有同步函数”，而是：

interactive DB wait
→ native async

local heavy encode
→ bounded blocking executor

resource lifecycle
→ explicit lifespan owner

transaction
→ domain atomicity preserved

retrieval
→ gap-targeted + budgeted

Evidence
→ stable identity + provenance + admissibility

Grounding
→ material claim support, not string presence

Testing
→ heartbeat + concurrency + rollback + provenance + semantic support

做到这些，才算真正从“会写 async/await”升级到“能设计 production RAG 的异步资源与证据边界”。