从 microsoft/SkillOpt（9.5k⭐）出发，深度解析 Harness 6 件套中"Skill"组件的训练范式：把 Skill 文档当 trainable state，用 6 阶段 ReflACT 循环 + Held-out 验证门 + 文本学习率调度器，零推理时延地优化 Agent SOP。

Craton 核心 4 件套完整实现：基础类型 (String/Buffer/Expected/Any)、Logger + Console/File/AsyncChannel、Thread/ThreadPool/Mutex/SpinLock/Event/Semaphore/Condition、Timestamp/Stopwatch/Timer，附 Linux/QNX/Android 三平台编译实测

深度剖析 tensorzero/tensorzero (⭐11.7k) 的核心架构：Rust 实现的统一 LLMOps 平台，把 LLM Gateway、Observability、Evaluation、Optimization、Experimentation 五大能力熔于一炉，InferenceProvider trait 是抽象核心，41 个 crate 构成模块化生态。

从 agentsmd/agents.md（22.5k⭐）标准出发，深度解析 Harness 6 件套中"Rule"组件的设计哲学、加载机制与跨厂商适配，并以 oh-my-agent、steipete/agent-rules、awslabs/aidlc-workflows 三个真实项目做横向对比。

精选 6 个真正落地 Harness Engineering 思想的 GitHub 高 star 开源项目，从代码层拆解它们如何实现 Rule / Skill / Sub-Agent / Workflow / Hooks / Context Engineering。

Skills Hub 全景调研：从 27 个开源项目看 AI Agent 资产中心

调研项目：27 个真实开源项目 · 调研时间：2026-06-26 · 完整报告 ≈ 13000 字

去年 12 月我给团队部署 Claude Code 时，发现一个让人抓狂的事情：同一段”按 Linus 风格 review PR”的提示词，散落在 5 个开发者机器上的 5 个不同位置——.claude/skills/、.cursor/rules/、.codex/skills/、Slack 历史消息、还有一个人写在 Notion 里。没有人能回答”我们团队到底有哪些可复用的 skill？”。那个瞬间我就知道，Skills Hub 不是一个 nice-to-have，而是 AI 编码团队绕不开的工程基础设施。

这份报告是我用了一周时间，对 27 个真实开源项目进行结构化拆解后产出的——我把它当作”如果让我从零搭建一个 Skills Hub”的决策地图。

Craton 自研基础库开篇：为什么自研、命名由来、命名空间设计、跨平台架构、9 大组件、错误处理策略、构建系统。覆盖 Linux/QNX/Android 三大目标平台

深度剖析 MervinPraison/PraisonAI（⭐8.3k）的核心架构 五层分层 Interface/Control/Runtime/Observability/Data、四类错误分类器自动恢复路由、多 AuthProfile 优先级故障转移、RunEvent 结构化事件流、协议驱动核心加重量级外围实现 lazy-import 策略。

Pydantic AI 不是一个”加了 Pydantic 校验”的 LangChain 包装器——它把整个 Agent 循环拆成了一个由 4 个节点组成的有向状态机，每一次”思考—调用工具—再思考”都对应着状态机中一条显式的边。这套设计的核心收益是：可观测性、可重放、可中断/恢复，所有这些都是 LangChain 的回调链或 LlamaIndex 的事件总线很难做到的。

与 4 月文章的关系：本文不是 4 月那篇《【Pydantic AI】类型安全的 AI Agent 框架》的重复——那篇聚焦”为什么 Pydantic AI 用类型系统”以及”Capabilities/可观测性”等概览，本文聚焦当前最新版（18k⭐）的源码细节：_agent_graph.py 状态机的具体实现、与 LangChain/smolagents 的架构差异、生产环境的取舍。读 4 月那篇可获得全景，读这篇可获得动手能力。

一、为什么写这篇？

过去两年里我读过几乎所有主流 Python Agent 框架的源码：LangChain 的 Runnable 链、CrewAI 的角色编排、AutoGen 的群聊、LlamaIndex 的 Workflows、smolagents 的 ReAct 循环。它们都解决了一个共同问题——把”LLM 调用 + 工具执行”封装成可复用的组件——但实现方式五花八门。

直到看到 Pydantic AI（18k⭐，Pydantic 官方团队出品）的 _agent_graph.py 我才意识到一件事：Agent 循环本质上是一个有限状态机（Finite State Machine），而不是一个回调链。

这个判断有几个关键证据：

Pydantic AI 把”Agent 在做什么”这件事从一段隐式的 Python 代码升级成了一张可被静态分析、可被可视化、可被持久化的图。本文会用源码 + Mermaid 图把这套架构拆给你看。

二、问题与价值：Agent 循环到底难在哪？

先抛个具体问题：让你写一个能查天气的 Agent，加 5 个工具，支持中途改模型、限制 token、遇到 429 自动重试、最后输出结构化 JSON——你的最小代码量是多少？

Pydantic AI 的答案是 ~30 行（下面会贴完整代码）。但实现这 30 行背后要解决 6 个工程问题：

1. 工具的 JSON Schema 自动生成（不用手写）
2. 依赖注入（让工具拿到数据库连接、用户上下文，而不用全局变量）
3. 结构化输出（让 LLM 返回 Pydantic 模型而不是字符串）
4. 多模型 Provider 抽象（OpenAI / Anthropic / Gemini / Ollama 一套 API）
5. 重试与回退（工具失败、模型 429、输出校验失败都要处理）
6. 可观测性（OpenTelemetry trace，自动记录每一步的 prompt/response/cost）

大部分框架用”中间件 + 回调”解决，但 Pydantic AI 的解法是把它们全部塞进一个状态机的节点里。下面看这套设计的全貌。

三、整体架构：4 个节点 + pydantic_graph 状态机

Pydantic AI 的核心抽象在 pydantic_ai_slim/pydantic_ai/_agent_graph.py，整个 Agent 循环是一个由 4 个节点组成的有向图：

graph LR
    Start([🚀 Start]) --> UPN["🟣 UserPromptNode<br/>📝 处理输入<br/>🔧 收集系统提示"]
    UPN --> MRN["🔵 ModelRequestNode<br/>🤖 调用 LLM<br/>📦 构造请求"]
    MRN --> CTN["🟠 CallToolsNode<br/>⚙️ 执行工具<br/>✅ 校验结果"]
    CTN -->|"有工具调用<br/>或需重新思考"| MRN
    CTN -->|"最终输出"| SFR["🟢 SetFinalResult<br/>📤 打包结果"]
    SFR --> End([✅ End])

    style Start fill:#C7CEEA,stroke:#9FA8DA,stroke-width:2px,color:#333
    style UPN fill:#E8D5F5,stroke:#CE93D8,stroke-width:2px,color:#333
    style MRN fill:#C7CEEA,stroke:#9FA8DA,stroke-width:2px,color:#333
    style CTN fill:#FFDAB9,stroke:#F9A825,stroke-width:2px,color:#333
    style SFR fill:#B5EAD7,stroke:#80CBC4,stroke-width:2px,color:#333
    style End fill:#B5EAD7,stroke:#80CBC4,stroke-width:2px,color:#333

每个节点的职责单一到极致：

为什么这么切？ 因为每个节点都有一个单一输出类型（ModelRequestNode 输出 CallToolsNode 或 ModelRequestNode，CallToolsNode 输出 ModelRequestNode 或 End[FinalResult]），可以用 Pydantic 在边转移时做静态类型校验。

四、源码深挖：状态机是怎么跑起来的

4.1 状态机的构建：build_agent_graph

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# pydantic_ai_slim/pydantic_ai/_agent_graph.py
def build_agent_graph(
    name: str | None,
    deps_type: type[DepsT],
    output_type: OutputSpec[OutputT],
) -> Graph[
    GraphAgentState,
    GraphAgentDeps[DepsT, OutputT],
    UserPromptNode[DepsT, OutputT],
    result.FinalResult[OutputT],
]:
    g = GraphBuilder(
        name=name or 'Agent',
        state_type=GraphAgentState,
        deps_type=GraphAgentDeps[DepsT, OutputT],
        input_type=UserPromptNode[DepsT, OutputT],
        output_type=result.FinalResult[OutputT],
        auto_instrument=False,
    )

    g.add(
        g.edge_from(g.start_node).to(UserPromptNode[DepsT, OutputT]),
        g.node(UserPromptNode[DepsT, OutputT]),
        g.node(ModelRequestNode[DepsT, OutputT]),
        g.node(CallToolsNode[DepsT, OutputT]),
        g.node(SetFinalResult[DepsT, OutputT]),
    )
    return g.build(validate_graph_structure=False)

注意 auto_instrument=False——Pydantic AI 把可观测性做成显式能力（capability），而不是默认行为，避免污染用户。

4.2 状态定义：GraphAgentState

1
2
3
4
5
6
7
8
9
10
11

@dataclasses.dataclass
class GraphAgentState:
    message_history: list[_messages.ModelMessage]
    """完整消息历史（user/assistant/tool 三方对话）"""
    run_id: str
    """本次 run 的唯一 ID（用于跨进程恢复）"""
    usage: _usage.RunUsage
    """累计 token / request 计数"""
    retries: dict[str, int]
    """按 tool_name 索引的重试次数"""
    final_result: result.FinalResult[OutputT] | None = None

整个 Agent 的运行状态都在 GraphAgentState 里。这就是 Pydantic AI 能”暂停—恢复”的关键——把整个 state 序列化存到 Redis，3 天后从 Redis 读出来继续跑，state 里所有的 run_id 仍然匹配。

4.3 节点执行：_make_request 的核心

ModelRequestNode.run() 调用的 _make_request 是最精彩的一段：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

async def _make_request(self, ctx):
    model, model_settings, model_request_parameters, message_history, run_context = (
        await self._prepare_request(ctx)
    )

    async def model_handler(req_ctx):
        # 实际调用 LLM
        response = await req_ctx.model.request(
            req_ctx.messages, req_ctx.model_settings, req_ctx.model_request_parameters
        )
        # 把通用 ToolCallPart 升级为类型化子类（ToolSearchCallPart 等）
        response = _narrow_tool_call_parts(response, req_ctx.model_request_parameters)
        return response

    request_context = ModelRequestContext(
        model=model, messages=message_history,
        model_settings=model_settings,
        model_request_parameters=model_request_parameters,
    )

    # 关键：通过 capability 中间件包裹 model 调用
    model_response = await ctx.deps.root_capability.wrap_model_request(
        run_context, request_context=request_context, handler=model_handler,
    )
    return await self._finish_handling(ctx, model_response)

三个关键设计：

1. req_ctx.model.request(...) 把”调 LLM”封装成一个可被替换的 handler——测试时可以换成 TestModel，生产环境换 OpenAI/Anthropic，不影响业务代码。
2. _narrow_tool_call_parts 在响应回来后统一升级所有 ToolCallPart 的类型。模型适配器只关心”emit base parts”，类型化由框架统一处理——这是解耦的精髓。
3. root_capability.wrap_model_request 是洋葱模型——重试、logfire、限流、人在回路都以”能力”形式插进来，不是硬编码。

4.4 数据流：消息在节点间的传递

graph TB
    subgraph "🌐 外部输入"
        UserPrompt["📥 UserPrompt<br/>(str / 多模态)"]
        Deps["📦 Deps<br/>(数据库/用户上下文)"]
    end

    subgraph "🟣 UserPromptNode"
        UPN["合并:<br/>• user_prompt<br/>• instructions<br/>• system_prompts<br/>• history"]
    end

    subgraph "🔵 ModelRequestNode"
        MRP["ModelRequest<br/>{parts: [...]}<br/>timestamp + run_id"]
    end

    subgraph "🤖 LLM Provider"
        LLM["OpenAI / Anthropic<br/>Gemini / Ollama"]
    end

    subgraph "🟠 CallToolsNode"
        MR["ModelResponse<br/>{parts: [TextPart,<br/>ToolCallPart, ...]}"]
        TE["Tool Execution<br/>并发/重试/审批"]
        TR["ToolReturnPart<br/>RetryPromptPart"]
    end

    subgraph "🟢 SetFinalResult"
        Val["Pydantic 校验<br/>output_type"]
        FR["FinalResult[T]"]
    end

    UserPrompt --> UPN
    Deps -.->|"注入 RunContext"| TE
    UPN -->|"构造"| MRP
    MRP -->|"调用"| LLM
    LLM -->|"返回"| MR
    MR -->|"解析 parts"| TE
    TE -->|"成功"| TR
    TE -->|"失败"| TR
    TR -->|"追加到 history"| MRP
    MR -->|"无工具调用<br/>纯文本输出"| Val
    Val -->|"通过"| FR
    Val -->|"校验失败<br/>重试"| MRP

    style UserPrompt fill:#C7CEEA,stroke:#9FA8DA,color:#333
    style Deps fill:#C7CEEA,stroke:#9FA8DA,color:#333
    style UPN fill:#E8D5F5,stroke:#CE93D8,color:#333
    style MRP fill:#E8D5F5,stroke:#CE93D8,color:#333
    style LLM fill:#FFDAB9,stroke:#F9A825,color:#333
    style MR fill:#FFDAB9,stroke:#F9A825,color:#333
    style TE fill:#FFDAB9,stroke:#F9A825,color:#333
    style TR fill:#FFF9C4,stroke:#F9A825,color:#333
    style Val fill:#B5EAD7,stroke:#80CBC4,color:#333
    style FR fill:#B5EAD7,stroke:#80CBC4,color:#333

4.5 工具的自动注册与 Schema 生成

Pydantic AI 最爽的地方是：工具函数直接是 Python 函数，Schema 自动从类型注解生成。

1
2
3
4
5
6
7
8
9
10
11

# pydantic_ai_slim/pydantic_ai/_function_schema.py（简化）
@dataclass(kw_only=True)
class FunctionSchema:
    function: Callable
    name: str
    description: str | None
    validator: SchemaValidator      # Pydantic Core 的验证器
    json_schema: ObjectJsonSchema   # 喂给 LLM 的 JSON Schema
    takes_ctx: bool                 # 第一个参数是否是 RunContext
    is_async: bool
    # ...

_function_schema.py 用了 Pydantic Core 的内部 API（pydantic._internal._generate_schema）来：

1. 把 async def my_tool(ctx: RunContext[DepsT], city: str) -> str 解析成 Pydantic TypeAdapter
2. 生成 {type: "object", properties: {city: {type: "string"}}, required: ["city"]}
3. 拿到 SchemaValidator 用于运行时校验

这意味着：如果你在工具函数签名里写错类型（比如把 int 写成 str），你的 IDE 会在你保存文件的瞬间报错，而不是等运行时 LLM 调过来才发现。

五、可运行的真实代码：从 0 到天气查询 Agent

下面是一段完整可运行的代码（用 TestModel 不需要真实 API key），展示 Pydantic AI 的核心用法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

# 安装：pip install 'pydantic-ai[examples]'
from dataclasses import dataclass
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext

# === 1. 定义依赖（替代全局变量）===
@dataclass
class WeatherDeps:
    api_key: str
    default_unit: str = "celsius"

# === 2. 定义结构化输出 ===
class WeatherReport(BaseModel):
    city: str
    temperature: float
    unit: str
    summary: str

# === 3. 创建 Agent（带类型参数）===
agent = Agent(
    "test",                            # 用 TestModel，不需要 API key
    deps_type=WeatherDeps,
    output_type=WeatherReport,
    instructions="你是一个天气助手，根据用户提问查询天气并返回结构化报告。",
)

# === 4. 用 @agent.tool 装饰器注册工具 ===
@agent.tool
async def get_weather(ctx: RunContext[WeatherDeps], city: str) -> str:
    """根据城市名查询当前天气。

    Args:
        city: 城市名称，例如 "北京"
    """
    # 实际场景这里会调外部 API
    return f"{city} 当前 22°C，晴，微风"

# === 5. 动态 system prompt（可以读 deps）===
@agent.instructions
async def add_unit_pref(ctx: RunContext[WeatherDeps]) -> str:
    return f"用户偏好使用 {ctx.deps.default_unit} 显示温度。"

# === 6. 运行 Agent ===
if __name__ == "__main__":
    deps = WeatherDeps(api_key="fake-key")
    result = agent.run_sync("北京今天天气怎么样？", deps=deps)
    print(result.output)
    # WeatherReport(city='北京', temperature=22.0, unit='celsius', summary='晴，微风')
    print(result.usage())
    # RunUsage(input_tokens=0, output_tokens=0, requests=1)

对比一下用 LangChain 实现同样功能需要写多少：

把这段代码保存成 weather.py 运行 python weather.py，就能看到 WeatherReport(...) 对象（不需要任何 LLM key，因为 test model 会返回固定响应）。

六、多 Agent 编排：把 Agent 当成 Tool 用

Pydantic AI 还支持Agent-as-Tool——一个 Agent 把另一个 Agent 当成 Tool 调用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

from pydantic_ai import Agent

# 子 Agent：专科医生
cardiology_agent = Agent(
    "openai:gpt-5.2",
    output_type=MedicalReport,
    deps_type=PatientInfo,
    instructions="你是心脏病专家。",
)

# 主 Agent：分诊台
triage_agent = Agent(
    "openai:gpt-5.2",
    output_type=TriageOutput,
    deps_type=PatientInfo,
    instructions="你是分诊医生，需要时调用专科医生。",
)

# 把子 Agent 变成主 Agent 的工具
triage_agent.toolset(cardiology_agent.as_tool(
    tool_name="consult_cardiologist",
    tool_description="向心脏病专家咨询",
))

这背后的机制是 pydantic_ai/toolsets/wrapper.py——Agent.as_tool() 实际返回一个 WrapperToolset，把子 Agent 的 run_sync() 调用包成一次 ToolCall。

七、优缺点：架构简洁性 vs 性能开销

7.1 架构优点

7.2 架构代价

7.3 一个真实场景的取舍

假设你要做：客服 Agent，调 5 个内部 API，处理 30 种意图，4 个 LLM Provider，2 个 Region，每天 200k 次调用。

• Pydantic AI：60 个源码文件就够。type-safe 模型 + 结构化输出 + 状态机可观测 → 写一次跑一年不重构。
• LangChain：需要 LCEL + custom callback + output parser + 自建 retrier。光工具集就能写 200 行 glue code。

八、与 3 个主流框架的对比

8.0 抽象层面差异总览

graph TB
    subgraph "🔵 Pydantic AI 状态机"
        P1["Node: UserPrompt"] --> P2["Node: ModelRequest"]
        P2 --> P3["Node: CallTools"]
        P3 -->|"有工具"| P2
        P3 -->|"无工具/最终输出"| P4["Node: SetFinalResult"]
    end

    subgraph "🟣 LangChain LCEL"
        L1["Prompt Template"] --> L2["Chat Model"]
        L2 --> L3["Output Parser"]
        L3 -->|"字符串"| L4["Agent Executor"]
        L4 -->|"循环"| L2
    end

    subgraph "🟠 smolagents ReAct"
        S1["system prompt + tools"] --> S2{"while-loop<br/>has_tool_call?"}
        S2 -->|"yes"| S3["执行工具"]
        S3 -->|"追加 step"| S2
        S2 -->|"no"| S4["final_answer()"]
    end

    style P1 fill:#E8D5F5,stroke:#CE93D8,color:#333
    style P2 fill:#E8D5F5,stroke:#CE93D8,color:#333
    style P3 fill:#E8D5F5,stroke:#CE93D8,color:#333
    style P4 fill:#B5EAD7,stroke:#80CBC4,color:#333
    style L1 fill:#C7CEEA,stroke:#9FA8DA,color:#333
    style L2 fill:#C7CEEA,stroke:#9FA8DA,color:#333
    style L3 fill:#C7CEEA,stroke:#9FA8DA,color:#333
    style L4 fill:#C7CEEA,stroke:#9FA8DA,color:#333
    style S1 fill:#FFDAB9,stroke:#F9A825,color:#333
    style S2 fill:#FFDAB9,stroke:#F9A825,color:#333
    style S3 fill:#FFDAB9,stroke:#F9A825,color:#333
    style S4 fill:#FFDAB9,stroke:#F9A825,color:#333

三种抽象的核心差别：显式状态机 vs 字符串管道 vs 命令式 while-loop。

8.1 Pydantic AI vs LangChain

核心差异：LangChain 是”字符串管道”，Pydantic AI 是”类型化状态机”。前者灵活但容易写出 Any 流；后者严格但更难 hack。

8.2 Pydantic AI vs smolagents

smolagents 是 HuggingFace 团队为了”极简”做出的设计取舍——单文件 2k 行，任何人都能读完。但代价是不能暂停/重放/多 Agent。Pydantic AI 是 smolagents 的工业级放大版。

8.3 Pydantic AI vs LlamaIndex Workflows

LlamaIndex 的 Workflows 更适合”文档 → 多步处理 → 答案”的 RAG 流水线；Pydantic AI 更适合”用户输入 → 工具调用 → 结构化输出”的通用 Agent 场景。两者可结合——Pydantic AI 处理 agent 循环，LlamaIndex 处理文档检索。

九、趋势：状态机会成为 Agent 框架的标准抽象吗？

回顾 Agent 框架的演化：

1
2
3
4
5

2023 H1:  while-loop + 函数调用（OpenAI Function Calling 兴起）
2023 H2:  LCEL pipe（LangChain 统一 API）
2024 H1:  Graph 抽象（LlamaIndex Workflows / Pydantic Graph / LangGraph）
2024 H2:  状态机 + 类型系统（Pydantic AI / Temporal Agents）
2025+:    显式持久化 + 中断恢复（durable execution）

Pydantic AI 的 pydantic_graph 不只是一个 Agent 库，它正在演化成通用工作流引擎（同仓库下的 pydantic_graph 子项目已经被外部项目用作 DAG 执行器）。一旦你接受”Agent 循环是状态机”这个前提，整个软件工程里所有”有状态、有循环、有重试”的问题都可以套这个模型。

对开发者的启示：

• 如果你正在选型 Agent 框架：优先选有显式状态抽象的（Pydantic AI / LangGraph），因为后期”加观测、加中断、加审计”几乎是必然需求。
• 如果你已经在用 LangChain：不必立刻迁移，但开始把 callback 逻辑搬到 RunnableConfig 里，为状态化做准备。
• 如果你正在自研 Agent：先画一张状态机图（4-6 个节点），再去写代码——你会省下 50% 的调试时间。

十、对你项目里的 Agent 启发

1. 先画状态机再写代码：一个 Agent 循环至少有”接收输入 / 调 LLM / 执行工具 / 校验输出”4 步，每一步就是一个状态。如果你的 Agent 代码不能用状态图描述，说明耦合度太高了。
2. 工具签名就是 API 文档：@agent.tool async def get_weather(ctx: RunContext[DepsT], city: str) -> str 的 docstring + 类型注解就是喂给 LLM 的 schema。不要再手写 JSON Schema。
3. 依赖注入胜过全局变量：把数据库连接、用户配置、feature flag 全部塞进 deps，工具函数从 ctx.deps 拿，测试时可以随便换 mock。
4. 结构化输出要趁早：output_type=BaseModel 不只是”返回 JSON”，而是把 LLM 的模糊输出编译期类型化——下游消费者再也不用写 try/except。

一句话总结：Pydantic AI 用 pydantic_graph 把 Agent 循环拆成 4 个节点，强制类型校验每一条边，让可观测/可重放/可中断从”附加特性”变成”基础设施”。 如果你的 Agent 还在用回调地狱，是时候看看状态机了。

参考资料

• 仓库：https://github.com/pydantic/pydantic-ai
• 文档：https://ai.pydantic.dev
• 核心源码：pydantic_ai_slim/pydantic_ai/_agent_graph.py（82KB，含完整 4 节点定义）
• 状态机引擎：pydantic_graph/pydantic_graph/graph_builder.py（96KB，通用 DAG 引擎）
• 工具装饰器：pydantic_ai_slim/pydantic_ai/agent/__init__.py 中 Agent.tool（97k 行位置）
• 示例代码：examples/pydantic_ai_examples/bank_support.py（银行客服完整 demo）

附：本文涉及的 Pydantic AI 关键源码位置

POCO 交叉编译完整指南：Linux ARM/QNX Neutrino/Android NDK 三平台 toolchain 文件、CMake 裁剪选项、IVI/IoT 工业实战案例、性能与包大小基准