跳到主要内容

数据截至 (上游 commit f9be26e5e26d)

Atomic Agents — 架构与原理

30 秒导读: Atomic Agents 是一个极薄的 Python agent 框架。它的赌注只有一句话:“一个 agent 的输入和输出,都是一个带文档的 Pydantic 模型(schema)。” 把 LLM 调用包成“吃一个 schema、吐一个 schema”的纯函数,于是 agent 和工具就能像乐高一样,靠对齐 schema拼接起来。底层结构化输出交给 Instructor,框架自己只管:提示生成、历史、schema 拼装、工具协议、MCP 桥接。

1. 这是什么(零基础也能懂)​

一句话定义: Atomic Agents 是一个把“调用大模型”包装成类型安全的输入→输出函数的轻量框架——你定义两个 Pydantic 模型(输入长什么样、输出长什么样),它负责让模型严格按输出模型回话。

解决什么问题 / 给谁用。 假设你在写一个 AI 应用,要让模型“生成 3 个搜索查询”,然后把查询喂给搜索工具,再把结果交给下一个 agent 总结。

  • 如果模型自由发挥地返回一段文本,你得自己解析、容错、对接下一步——脆弱且难维护。
  • Atomic Agents 让你声明:这个 agent 的输出 schema 就是“一个 queries: list[str] 列表”。模型被强制按这个结构回。下一个工具的输入 schema 如果也是这个结构,两者直接对接。

给谁用:想用普通软件工程方法(类型、组合、可测)搭 AI 管线的开发者,而不是想要一个大而全、帮你决定一切的 agent 平台的人。

它能做什么(功能):

  • 定义自带 schema 的 agent(同步 / 异步 / 流式三套 run)。
  • 用分节模板(背景 / 步骤 / 输出指令)生成系统提示,并能动态注入运行时上下文。
  • 管理多轮对话历史(带 turn 概念、可序列化、支持图片/音频/PDF 多模态)。
  • 定义工具,并把任意 MCP 服务器的工具/资源/提示自动变成框架里的工具类。
  • 按 token 预算自动裁剪历史;挂 hook 监控调用与重试。

用起来什么样。 一个最小聊天 agent(取自 atomic-examples/quickstart/quickstart/1_0_basic_chatbot.py 的同款写法):

import instructor, openai
from atomic_agents import AtomicAgent, AgentConfig, BasicChatInputSchema, BasicChatOutputSchema

client = instructor.from_openai(openai.OpenAI())          # 结构化输出由 instructor 负责
agent = AtomicAgent[BasicChatInputSchema, BasicChatOutputSchema](
    AgentConfig(client=client, model="gpt-5-mini")        # 中括号里就是 [输入schema, 输出schema]
)
resp = agent.run(BasicChatInputSchema(chat_message="你好"))
print(resp.chat_message)                                   # resp 是一个强类型对象,不是裸字符串

注意那个 AtomicAgent[InputSchema, OutputSchema] —— 方括号里的两个类型,就是这个 agent 的“接口签名”。这是整个框架的灵魂。

一句话直觉/类比。 把每个 agent / 工具想成一块乐高积木:它的“凸点和凹槽”就是输入 schema 和输出 schema。只要凸点配凹槽(A 的输出 schema = B 的输入 schema),就能拼。框架本身不是大厦,而是一盒标准化的积木接口。

2. 顶层全景(它大概怎么转)​

先看一次 agent.run() 内部的数据流。怎么读这张图:从上到下是一次同步调用,左边是你给的输入,右边是框架内部三块状态(提示 / 历史 / schema)如何汇成发给模型的 messages。

            你的代码
               │  agent.run(InputSchema(...))
               ▼
     ┌───────────────────────┐
     │      AtomicAgent       │   核心循环(agents/atomic_agent.py)
     └───────────────────────┘
        │        │        │
        │        │        └──► output_schema ──┐  (要模型按这个结构回)
        │        │                              │
        │        └──► ChatHistory ──────────┐   │  (历史多轮 + 本次 user 输入)
        │                                   │   │
        └──► SystemPromptGenerator ─────┐   │   │  (背景/步骤/输出指令 + 上下文providers)
                                        ▼   ▼   ▼
                              _prepare_messages() 汇成 messages[]
                                        │
                                        ▼
                           client.chat.completions.create(
                               messages, model, response_model=output_schema)
                                        │   ← Instructor 在这里强制结构化、校验、重试
                                        ▼
                          OutputSchema 实例(强类型)→ 写回 history → 返回

部件一句话职责:

部件干什么在哪个文件
AtomicAgent核心循环:组装消息、调 Instructor、写回历史atomic-agents/atomic_agents/agents/atomic_agent.py
BaseIOSchema所有输入/输出 schema 的基类,强制带 docstringatomic-agents/atomic_agents/base/base_io_schema.py
SystemPromptGenerator把背景/步骤/输出指令 + 动态上下文拼成系统提示atomic-agents/atomic_agents/context/system_prompt_generator.py
ChatHistory多轮历史:turn 模型、序列化、多模态、溢出管理atomic-agents/atomic_agents/context/chat_history.py
BaseTool工具的标准接口(输入 schema → 输出 schema)atomic-agents/atomic_agents/base/base_tool.py
MCPFactory把远程 MCP 服务器的工具/资源/提示动态生成为 BaseTool 子类atomic-agents/atomic_agents/connectors/mcp/mcp_factory.py
TokenCounter用 LiteLLM 算 token,支撑上下文裁剪atomic-agents/atomic_agents/utils/token_counter.py

主线走一遍(高层,不进代码):

  1. 你用输入 schema 包一个消息,调 agent.run(...)。
  2. 框架先按 max_context_tokens 裁剪老历史(若设了),再把本次输入加进 ChatHistory。
  3. _prepare_messages() 把“系统提示 + 历史”拼成 messages 列表。
  4. 调 client.chat.completions.create(..., response_model=output_schema) —— Instructor 负责让模型按 output schema 结构化输出、校验、失败重试。
  5. 得到强类型 OutputSchema 实例,写回历史,返回给你。

关键认知:框架本身不碰 token、不碰 JSON 解析。 那些脏活都委托给 Instructor + Pydantic。Atomic Agents 的代码量小,正是因为它只做“组织”,不做“底层结构化”。

3. 阅读地图(按这个顺序往下钻)​

  • 01-core-loop.md — 先读这章。一次 run() 到底发生了什么:_prepare_messages / _build_system_messages / 写回历史 / 同步·异步·流式四个入口的差异。理解了循环,其余都是它的零件。
  • 02-schema-and-chaining.md — “原子”哲学的硬核:BaseIOSchema 为什么强制 docstring、generic 类型参数怎么在 __init_subclass__ 里被偷出来当 schema、以及“对齐 schema = 拼装管线”怎么落地。
  • 03-prompt-history-context.md — 输入侧三块状态:系统提示的分节模板、ContextProvider 动态注入、ChatHistory 的 turn 模型与多模态序列化/反序列化。
  • 04-tools-mcp-tokens.md — 输出侧与外部世界:BaseTool 标准接口、MCPFactory 如何把远程 MCP 工具运行时编译成本地工具类、以及 token 计数与历史裁剪。

4. 一句话精华(读完应带走的)​

Atomic Agents 的全部魔法,是把“agent”降维成一个有类型签名的纯函数 f: InputSchema -> OutputSchema。 一旦 agent 和工具都长成这个样子,“编排”就退化成了“对齐类型”——而类型对齐是编译器/IDE/人脑都擅长的事。框架因此可以很薄。

详见各章。代码地图集中在 04-tools-mcp-tokens.md 末尾。