BodySense Python Async / RAG Engineering
BodySense L4 的差分学习总览：从 async 外观与 Event Loop 阻塞开始，串起 Async Postgres Pool、Embedding 执行边界、Targeted RAG、Evidence provenance 与 Grounding Eval。
[!info] related notes
- 所属 MOC: bodysense-moc、rag-engineering-moc
- 相关概念: hybrid-search、reranker、rag-intent-aware-reranking、bodysense-targeted-rag-evidence-provenance、rag-faithfulness-checker
- 易混淆概念: retrieval failure vs reranking failure vs grounding failure；Event Loop blocking vs backpressure
- 相关资源:
BodySense Python Async / RAG Engineering
L4 的定位：不是重新背 async/await
L4 的目标不是再讲一遍 async def、await、gather 的语法，而是把 BodySense 已经运行在 FastAPI / LangGraph / Agent runtime 中的真实 RAG 数据路径当成一条生产执行链来审查：
- 这段代码真的会把执行权还给 Event Loop 吗？
- DB 连接、线程池、Embedding 模型与 Evidence 的生命周期由谁拥有？
- Retrieval 返回的数据怎样变成可追溯、可采纳、可验证的 Evidence？
- 模型最终提出的 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”，而是识别资源属性，再放到正确的执行边界。
对话学习时最有用的并发图：100 个 coroutine、10 个 DB connection
仅仅说“Pool 会复用连接”很容易停留在概念层。实际学习时更清楚的心智模型是：
100 concurrent coroutines want DB work
│
↓
AsyncConnectionPool(max=10)
│
┌────────┴────────┐
↓ ↓
10 coroutines 90 coroutines
checkout connection await pool availability
│ │
↓ ↓
await DB response SUSPENDED
│ │
└────────┬────────┘
↓
Event Loop stays free
to schedule other ready work这里要避免两个误解。
第一：
100 coroutine
≠
100 DB queries simultaneously executing真正数据库并发被 pool capacity 限制在大约 10 个 connection lease；其余 coroutine 是异步排队。
第二：
90 coroutine waiting for pool
≠
90 threads blocked它们在 await pool availability 时会 suspend，Event Loop 仍然可以继续调度：
SSE progress
other Agent runs
health endpoint
interrupt / resume
timeouts对比同步单连接/同步 DB call：
Coroutine A
→ sync psycopg execute()
→ current loop thread blocks
Coroutine B/C/D/... + SSE + timeout callbacks
→ all wait for loop thread to come back所以最重要的区分是：
Async capacity limit
→ backpressure / queueing
Sync blocking call on loop thread
→ scheduler starvation这也是为什么：
non-blocking
≠
infinite throughputAsync Pool 解决“怎么等待”，Pool Size 决定“同时允许多少数据库工作”。
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 operationSearch 通常只需要短只读 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 throughputL4.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 concurrencySemaphore.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
→ EvidenceL4 负责把这条语义真正落到 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。
从 Hybrid Search 到 Faithfulness：一张图定位 RAG 到底在哪一步失败
这轮实际学习里，一个很有用的例子是：目标文档/证据 142 已经被 Retriever 找到了，但最终没有排进模型真正消费的前列。 这不是 Retrieval Failure，而是典型的 Reranking Failure。
把整个链画出来会清楚很多：
Decision-Relevant EvidenceGap
↓
Targeted Query
↓
┌──────────────────────────────┐
│ Retrieval Backend │
│ │
│ Semantic / Vector Search │
│ + │
│ Lexical / Keyword Search │
│ ↓ │
│ Hybrid Search │
└──────────────┬───────────────┘
↓
Candidate Set
e.g. [..., 142, ...]
↓
Rerank
↓
Top-k cutoff
↓
Raw SearchResult objects
↓
Normalized Evidence Boundary
↓
Evidence ID / source version / excerpt
gap_id / attempt_id / provenance
↓
Admissibility
↓
Gap resolved / unresolved?
↓
Model material claim
↓
Grounding / Faithfulness
↓
DecisionAuthority / Safety这张图对应一组非常实用的 Failure Attribution：
| 失败类型 | 142/正确证据在哪里丢了？ | 第一检查层 |
|---|---|---|
| Retrieval Failure | Candidate Set 里根本没有 | query / embedding / lexical / filters / knowledge coverage |
| Reranking Failure | Candidate Set 里有，但被排到 Top-k 外 | reranker / intent boost / score fusion / cutoff |
| Normalization / Provenance Failure | raw result 有，但没有正确变成稳定 Evidence | normalized contract / identity / source version / attempt link |
| Admissibility Failure | Evidence 已规范化，但当前 policy 不允许使用 | source/version/fact-type policy |
| Resolution Failure | Evidence 可采纳，但不足以关闭 Gap | semantic sufficiency / conflicting or partial evidence |
| Grounding / Faithfulness Failure | Evidence 已存在，但最终 claim 没被它支持 | material claim support / contradiction |
| Authority Failure / Block | claim 可以 grounded，但当前安全/业务规则仍不允许动作 | SafetyEnvelope / DecisionAuthority |
所以实际排障时不要笼统说：
“RAG 没找到答案”而应该沿着：
retrieve
→ rerank
→ normalize
→ admit
→ resolve
→ ground
→ authorize找第一处违背 contract 的位置。
Hybrid Search 解决的是 Recall，不保证最后排序正确
Hybrid Search 通常把：
semantic similarity
+
lexical / keyword matching结合起来，提高候选集 recall。
它解决的问题更接近：
“正确候选有没有机会被捞进来？”
而 Reranker / 意图感知 RAG 重排 回答的是：
“候选都在以后，谁应该排在真正喂给模型的前面？”
因此：
correct doc retrieved
≠
correct doc survives reranking/top-kNormalized Evidence Boundary 隔离 Retrieval Backend
Retriever 可能以后从：
vector only
→ hybrid vector + BM25
→ cross-encoder reranker
→ another search backend但 Agent / Governance 层不应该跟着读取每种 backend 的 raw payload。
边界应该是：
Retriever-specific SearchResult
↓
normalize
↓
Stable Evidence Contract
├─ evidence_id
├─ source identity/version
├─ observed excerpt
├─ acquired_for_gap_id
├─ attempt_id
├─ score/rank if relevant
└─ policy metadata这就是 Normalized Evidence Boundary：下游依赖的是稳定 Evidence 语义，而不是某个 retriever 的实现细节。
Faithfulness 解决的是“模型有没有忠于证据”，不是“检索有没有成功”
Faithfulness / Grounding 位于更下游：
Evidence already retrieved/admitted
+
Model generated claim
→ does evidence actually support this claim?例如：
Evidence:
“疼痛加重时停止臀桥”
Claim:
“疼痛加重时继续臀桥”两者 lexical/embedding similarity 都可能很高，但语义上是 contradiction。
因此：
Retrieval relevance
≠
Reranking quality
≠
Evidence admissibility
≠
Gap resolution
≠
Faithfulness / Grounding外部 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 casesLayer 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
≠ supportEmbedding 可以做候选信号，不能单独成为 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.Connectionin 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
→ DecisionAuthorityL4 与 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 应该能独立回答的问题
- 为什么同步 psycopg 写在
async def里仍会阻塞整个 Event Loop？ - Remote Embedding API 与 local
SentenceTransformer.encode()为什么需要不同并发模型？ to_thread解决什么，为什么还要 Semaphore / bounded executor？- Pool、Connection lease、Transaction 的 owner 分别是什么？
- 为什么 async pool wait 是 backpressure，不是 thread blocking？
- 为什么 interactive targeted retrieval 不应该为了“异步化”就全部改成后台 Job？
- 为什么外部 RAG 不能补成 user fact？
retrieved / admissible / resolved为什么必须三层分开？- Citation 与 Provenance 分别回答什么？
- 为什么动作名被 Evidence 提到仍不足以证明 dosage/stop condition grounded？
- 哪些 grounding check 必须 deterministic，哪些才适合 semantic matcher / Judge？
- 怎样用 heartbeat test 证明一段工作没有堵 Event Loop？
- 为什么 async 重构仍要保留 ingestion transaction atomicity？
- 为什么 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 的异步资源与证据边界”。
Digital Biome