BodySense Diagnosis 的 PydanticAI 执行边界

用 BodySense Diagnosis 实战串起 constructor DI、Protocol、PydanticAI deps/RunContext、Tool Calling、结构化输出与 targeted evidence trail 的完整执行边界。

#type / synthesis #status / growing #tech / ai #tech / lang / python #resource / python #resource / bodysense #resource / agent

[!info] related notes

BodySense Diagnosis 的 PydanticAI 执行边界

范围与核心结论

这篇笔记记录 BodySense Diagnosis 学习/重构阶段中已经实际写代码和测试验证的一整条 PydanticAI 执行边界。范围停在 typed Agent + targeted evidence tool 基础，不包含 production DiagnosisService 的模型路由切换，也不进入 Treatment。

这一节最重要的结论不是某个 API，而是把不同层级的数据和依赖分清：

Application / composition root
        │
        ├─ 注入长期能力：AIExecutor / EvidenceSearcher
        │
        ▼
DiagnosisDependencies（本轮 run context）
        │
        ├─ Input：BodyState revision / body_state / history / profile
        ├─ Capability：evidence_searcher
        └─ Run State：retrieved_evidence
        │
        ▼
PydanticAI Agent
        │
        ├─ RunContext[DiagnosisDependencies]
        ├─ @agent.tool search_evidence
        └─ output_type=DiagnosisAgentOutput
        │
        ▼
AgentRunResult
        ├─ output：模型结构化输出
        └─ application 可从 deps 读取本轮 evidence trail

因此需要区分五个概念：

  • Input：本轮推理要读取的业务输入，例如精确的 BodyState revision。
  • Dependency / Capability：代码执行所依赖的能力，例如 EvidenceSearcher。
  • Run State / Artifact：本轮执行过程中产生、由 runtime 确定性维护的数据，例如 retrieved_evidence。
  • Model Output：模型负责生成并经 Pydantic 校验的数据，例如 DiagnosisAgentOutput。
  • Execution Result：应用最终希望拿到的组合结果。后续可以进一步封装为 DiagnosisExecutionResult(output, retrieved_evidence)，不必让调用者知道内部是靠 deps accumulator 实现的。

Application DI 与 PydanticAI deps 不是一回事

本节先从传统依赖注入开始。DiagnosisService 这类 application service 的 constructor DI 回答的是：

这个长期对象依赖哪一种能力，由谁创建并注入？

例如 consumer-owned Protocol：

class AIExecutor(Protocol):
    async def generate(...) -> ...:
        ...

Service 只依赖能力抽象，composition root 负责注入真实实现或 Fake。

PydanticAI 的：

Agent(
    model,
    deps_type=DiagnosisDependencies,
    output_type=DiagnosisAgentOutput,
)

回答的是另一个问题：

一次具体 Agent run 可以访问什么 run-scoped context？

deps_type=DiagnosisDependencies 只是声明类型；真正的实例在运行时传入：

deps = DiagnosisDependencies(...)
result = await agent.run(prompt, deps=deps)

PydanticAI 再为本轮创建：

RunContext[DiagnosisDependencies]

Tool / instructions 通过：

ctx.deps

拿到同一个 run-scoped deps 对象。

所以更准确的原则是：

不要因为 PydanticAI 有 run-level DI，就放弃 application-level dependency inversion。两者解决的是不同依赖边和不同生命周期的问题。

Protocol 与结构化类型：能力边界不依赖具体类

本节为 targeted evidence retrieval 提取了最小能力接口：

from typing import Any, Protocol


class EvidenceSearcher(Protocol):
    async def search(
        self,
        query: str,
        *,
        top_k: int = 5,
    ) -> list[dict[str, Any]]:
        ...

关键点：

  1. Protocol 类似 Go interface：关注对象“有没有这些方法”，而不是“继承了谁”。
  2. Fake 不需要继承 EvidenceSearcher，只要结构匹配即可。
  3. 通过下面这种赋值可以让静态类型检查器验证 structural typing：
searcher: EvidenceSearcher = FakeEvidenceSearcher()
  1. * 让 top_k 变成 keyword-only：
await searcher.search("neck", top_k=5)  # OK
await searcher.search("neck", 5)        # 不允许

Protocol 的签名必须和实现实际承诺保持一致；如果真实实现只接受 keyword-only top_k，Protocol 也不应错误承诺 positional 调用。

DiagnosisDependencies：run-scoped context

当前学习实现：

@dataclass(slots=True)
class DiagnosisDependencies:
    body_state_revision: int
    body_state: dict[str, Any]
    relevant_history: list[dict[str, Any]] = field(default_factory=list)
    profile: dict[str, Any] = field(default_factory=dict)
    rag_context: str = ""
    evidence_searcher: EvidenceSearcher | None = None
    retrieved_evidence: list[dict[str, Any]] = field(default_factory=list)

这里实际上混合了三种角色：

输入数据
├─ body_state_revision
├─ body_state
├─ relevant_history
└─ profile

可调用能力
└─ evidence_searcher

本轮可变状态 / artifact accumulator
└─ retrieved_evidence

retrieved_evidence 严格说并不是传统意义上的 dependency，更像 run state / artifact collector。当前把它放在 deps 中，是为了让 Tool 可以在本轮运行里确定性记录证据，同时 application 在 Agent 结束后还能读取同一个 Python 对象。

这利用的是 Python 的对象引用语义：

外部变量 deps ─────────────┐
                           ▼
                 DiagnosisDependencies object
                           ▲
ctx.deps ──────────────────┘

agent.run(deps=deps) 不会为业务需要创建一个隔离副本。因此 Tool 执行：

ctx.deps.retrieved_evidence.append(item)

之后，外部的：

deps.retrieved_evidence

可以直接看到修改。

default_factory=list 则保证不同 DiagnosisDependencies 实例拥有不同 list：

deps_a.retrieved_evidence is not deps_b.retrieved_evidence

Typed Agent 与结构化输出

Agent 声明：

agent = Agent(
    model,
    deps_type=DiagnosisDependencies,
    output_type=DiagnosisAgentOutput,
    system_prompt=DIAGNOSIS_SYSTEM_PROMPT,
    name="bodysense_diagnosis",
)

output_type=DiagnosisAgentOutput 的价值不是简单替代 json.loads()，而是把结构化输出提升成 Agent 的执行契约：

旧路径
model text
  → json.loads
  → model_validate
  → application

PydanticAI 路径
model / output tool
  → PydanticAI structured-output runtime
  → DiagnosisAgentOutput
  → result.output

调用方可以直接：

result = await agent.run(...)
assert isinstance(result.output, DiagnosisAgentOutput)

DiagnosisAgentOutput 同时承载业务约束，例如：

  • Diagnosis candidate 数量是 0..N，不再固定 1..3；
  • completed 必须至少有一个 candidate；
  • insufficient_information / safety_blocked 可以零 candidate；
  • Python 不生成 durable analysis_id / candidate_id，这些身份仍由 Go 持有。

Structured Output 为什么也会出现 ToolCallPart

测试消息链时，一个重要发现是：使用默认 structured output 时，PydanticAI 会把最终输出 schema 暴露成 output tool。

因此一次 run 中可能同时看到：

ToolCallPart search_evidence   ← 普通 function tool
ToolCallPart final_result      ← structured output 的 output tool

二者语义不同：

  • search_evidence：模型请求 runtime 真正执行 Python 功能。
  • final_result：模型用符合 schema 的参数提交最终结构化输出，PydanticAI 再做验证并结束 run。

不能据此总结为“只要 structured output 就一定额外请求模型一次”。额外 model round 取决于前面是否发生 Tool Call、重试等。例如模型如果第一次响应就直接提交 output tool，可以直接完成。

测试也因此不应该断言“整次 run 只有一个 ToolCallPart”，而应过滤真正关心的 contract：

tool_calls = [
    part
    for message in messages
    for part in message.parts
    if isinstance(part, ToolCallPart)
    and part.tool_name == "search_evidence"
]

RunContext 与 @agent.tool 的真实调用关系

Tool：

@agent.tool
async def search_evidence(
    ctx: RunContext[DiagnosisDependencies],
    query: str,
    top_k: int = 5,
) -> list[dict[str, Any]]:
    ...

这里参数来自两个不同来源：

PydanticAI runtime
└─ ctx

模型 Tool Call
├─ query
└─ top_k

模型不会创建、也不会看到 application 如何构造 RunContext。Application 只传 deps，PydanticAI runtime 创建 RunContext 并令 ctx.deps 指向它。

完整调用心智模型：

Model
  │  ToolCallPart(name=search_evidence, args={query, top_k})
  ▼
PydanticAI runtime
  │  构造/提供 RunContext
  ▼
search_evidence(ctx, query, top_k)
  │
  ▼
ctx.deps.evidence_searcher.search(...)
  │
  ▼
ToolReturnPart
  │
  ▼
Model 继续推理

如果 evidence_searcher 是可选能力：

if searcher is None:
    return []

比直接 raise RuntimeError 更符合类型契约。否则类型虽然写了 EvidenceSearcher | None，运行时却把 None 当 fatal error，会造成 optionality 与 runtime semantics 不一致。

Targeted RAG：EvidenceSearcher 是能力，Tool 是模型入口

这两个概念不能混淆：

EvidenceSearcher
= application/runtime 提供给本轮的检索能力

@agent.tool search_evidence
= 暴露给模型的工具入口

组合关系：

composition root / caller
        ↓ 注入具体 EvidenceSearcher
DiagnosisDependencies
        ↓
RunContext.deps
        ↓
@agent.tool search_evidence
        ↓
EvidenceSearcher.search

这样 Agent 不直接依赖具体知识库实现，测试可注入 Fake，未来也可以换检索后端。

Targeted RAG 的目标不是每次 Diagnosis 都做 broad retrieval，而是在具体 evidence gap 会实质改变 candidate 时才检索。Tool 层保持读取知识的能力，不把 Go-owned BodyState / Diagnosis durable business ownership 移入 Python。

Evidence trail：Tool 结果为什么要走两条通道

仅仅：

return results

只保证 evidence 进入 ToolReturnPart 并回到模型，但 application 层不容易稳定知道这轮实际检索了哪些证据。

因此本节增加了一条 runtime-owned evidence trail：

searcher.search()
        ↓
      results
      /     \
     /       \
    ▼         ▼
ToolReturn    retrieved_evidence
给模型        给 application

Tool 中：

results = await searcher.search(query, top_k=top_k)
# 记录 trail
...
return results

不建议让模型自己在 DiagnosisAgentOutput 里“再抄一遍所有检索结果”，因为 evidence 是 runtime 已知事实，交给模型复述会引入遗漏、改写或编造风险。

同样，也不建议 application 为了拿 evidence 去解析 AgentRunResult.all_messages() 的 ToolReturnPart，否则业务层会耦合 PydanticAI message representation。retrieved_evidence 是更稳定的 application contract 候选。

Evidence trail 按 evidence_id 去重

直接：

ctx.deps.retrieved_evidence.extend(results)

会让重复检索产生重复 provenance。当前实现按稳定 evidence_id 去重：

known_ids = {
    str(item.get("evidence_id", ""))
    for item in ctx.deps.retrieved_evidence
}

for item in results:
    evidence_id = str(item.get("evidence_id", ""))

    if evidence_id and evidence_id not in known_ids:
        ctx.deps.retrieved_evidence.append(item)
        known_ids.add(evidence_id)

return results

注意：

  • 去重只影响 application evidence trail；
  • Tool 仍原样 return results 给模型；
  • 每 append 一个新 evidence 后立即 known_ids.add(evidence_id)，这样同一批 results 内部重复也能被挡住；
  • 没有稳定 evidence_id 的记录不进入 provenance trail。

这里的 set comprehension：

known_ids = {expr for item in items}

适合“从一批对象抽取唯一 key 集合”这种简单映射场景。复杂业务分支仍优先普通 for，Pythonic 不等于越短越好。

用 capture_run_messages() 观察 Agent loop

测试中使用：

with capture_run_messages() as messages:
    result = await agent.run(...)

可以直接观察 runtime 消息交换。

本节看到的关键顺序：

ModelRequest
  ├─ SystemPromptPart
  └─ UserPromptPart

ModelResponse
  └─ ToolCallPart search_evidence

ModelRequest
  └─ ToolReturnPart search_evidence

ModelResponse
  └─ ToolCallPart final_result

ModelRequest
  └─ ToolReturnPart final_result

业务 Tool Call / Return 可以通过：

part.tool_name == "search_evidence"

过滤，并验证：

tool_return.tool_call_id == tool_call.tool_call_id

tool_call_id 是 runtime 把一次 Tool Return 对应回具体 Tool Call 的关联键；不能只靠 tool name，因为同一工具可能多次调用。

AgentRunResult 除了 output，还提供 all_messages()、new_messages()、usage、run_id、metadata 等运行信息。测试时 capture_run_messages() 更方便观察 exchange，但业务代码不应无必要依赖消息内部结构。

测试策略：用 Fake 验 wiring，不把 TestModel 假数据当业务语义

本节使用 FakeEvidenceSearcher：

class FakeEvidenceSearcher:
    def __init__(self) -> None:
        self.calls: list[tuple[str, int]] = []

    async def search(
        self,
        query: str,
        *,
        top_k: int = 5,
    ) -> list[dict[str, Any]]:
        self.calls.append((query, top_k))
        return [
            {
                "evidence_id": "evidence-1",
                "content": "This is a piece of evidence.",
            }
        ]

测试关注的 contract：

  • Fake 不继承 Protocol 也能通过 structural typing；
  • Agent Tool 确实调用 run-scoped searcher；
  • query 是合法非空字符串，top_k == 5；
  • ToolCallPart → ToolReturnPart 的 tool_call_id 一致；
  • ToolReturn 中包含 Fake 返回的 evidence；
  • Agent 最终仍得到 DiagnosisAgentOutput；
  • Agent run 后，外部原始 deps.retrieved_evidence 可以看到 evidence；
  • 相同 evidence_id 不会重复记录。

不要写：

assert query == "a"

因为 TestModel 自动生成的参数只是满足 schema 的确定性测试数据，"a" 不是业务 contract。

这一节顺带学到的 Python 工程知识

from __future__ import annotations

这是 Python 标准语言特性，不是第三方库。它改变当前模块中类型注解的处理方式，方便前向引用和现代类型表达。

... 在 Protocol 中

async def search(...) -> ...:
    ...

... 是 Ellipsis，这里表示只声明接口形状，不提供实现体。

default_factory=list

不要用共享可变默认值。测试应验证不同实例拥有不同 list，而不是手工给两个实例各传一个 [] 后再证明它们不同。

dict 与对象属性访问

当前 evidence 是：

list[dict[str, Any]]

因此访问：

item["evidence_id"]
item.get("evidence_id", "")

而不是：

item.evidence_id

只有 dataclass / Pydantic model / 普通对象定义了该属性时才使用点访问。

append 与 extend

items.append(results)

会把整个 list 当一个元素加入，产生嵌套 list；

items.extend(results)

会逐项合入。但一旦需要按 ID 去重，就不能简单 extend，需要逐条判断。

comprehension 的工程边界

简单的映射/过滤/收集非常常见：

[x for x in values if x.enabled]

双层 comprehension 也合法，但当条件、转换、副作用继续增加时，应拆回普通 for。可读性优先于“最短代码”。

Ruff、Pyright 与编辑器类型体验

本节同时补齐了 Python 工程化工具的职责分工：

Ruff
├─ formatter
├─ lint
├─ import sorting
└─ safe fixes

Pyright / Pylance
├─ 静态类型推导
├─ hover
├─ autocomplete
├─ go to definition
└─ type diagnostics

ctx 明明声明成：

ctx: RunContext[DiagnosisDependencies]

但编辑器 hover 显示 Any，不代表代码类型设计一定有问题。实际排查发现学习快照没有自己的 .venv，编辑器没有正确解析安装了 pydantic_ai 的 Python 环境。显式使用主仓库 interpreter 后，Pyright 对相关文件可以做到 0 errors。

因此 Ruff 不能替代 Python language server；完整开发体验应配合 Python/Pylance（或等价 Pyright language server）并选择正确 interpreter。

实战中暴露出的设计与测试教训

Optional 类型要和运行语义一致

既然：

evidence_searcher: EvidenceSearcher | None = None

表示该能力可选，那么 Tool 在没有 searcher 时优先：

return []

而不是无条件 raise RuntimeError。如果没有 searcher 必须阻断 Diagnosis，就应从类型层把它设计成必填，而不是一边 Optional 一边运行时 fatal。

测试要锁 contract，不要锁框架内部偶然细节

第一次消息测试假设：

整次 run 只有一个 ToolCallPart

但 structured output 的 final_result 也是 ToolCallPart，测试因此失败。正确做法是过滤业务工具 search_evidence。

每次重构后必须跑 focused tests

最终替换 Tool 代码时曾意外删掉 create_diagnosis_agent() 的：

return agent

结果所有 Agent tests 变成 NoneType.run。Focused pytest 立刻定位到这一回归。恢复后：

Diagnosis agent + model focused tests: 14 passed
Ruff touched files: clean

这个例子说明：局部代码逻辑看起来正确，不等于 factory / wiring contract 仍完整。

当前边界与下一步

这一节完成后，已经跑通的学习链路是：

Protocol
  → structural typing
  → DiagnosisDependencies
  → Agent(deps_type, output_type)
  → agent.run(deps=...)
  → RunContext.deps
  → @agent.tool
  → EvidenceSearcher.search
  → ToolReturnPart
  → retrieved_evidence trail
  → evidence_id 去重
  → DiagnosisAgentOutput

还没有完成的 production 迁移：

现有
AIService
  → ModelRouter
  → OpenAICompatibleProvider
  → routing / fallback

下一阶段需要研究
PydanticAI Model / Provider / fallback
  → 哪些能力由 PydanticAI 接管
  → 哪些 BodySense routing/fallback 语义必须保留
  → 再把 DiagnosisService 普通候选生成路径切到 typed Agent

必须继续保护：

  • Go-owned BodyState 与精确 revision 语义；
  • Go-owned durable analysis/candidate IDs；
  • safety gate；
  • Diagnosis history persistence；
  • public compatibility response；
  • Consultation LangGraph runtime；
  • Treatment 暂不提前迁移。
创建于 2026/8/18 更新于 2026/8/18