跳到主要内容

数据截至 (上游 commit 610a2f575c46)

声明式前端:FastAgent 应用、装饰器与工厂

30 秒导读: fast-agent 让你用 @fast.agent(...) 这样的装饰器来「声明」一个 agent,而不是手写它的构造过程。装饰器当场并不创建任何 agent 对象——它只是往一个字典里塞一份配置(AgentConfig)。真正把配置变成活的 agent 实例,发生在你 async with fast.run() 的那一刻,由一套工厂函数按依赖顺序统一完成。本章讲的就是这条「声明 → 登记 → 实例化」的流水线。

本章聚焦「用户怎么声明 agent、框架怎么把声明变成活的实例」。各 agent 类的内部实现见 Agent 类栈;一次 turn 怎么跑见 工具循环引擎;各工作流(chain/parallel/router…)的语义见 工作流模式。

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

一句话定义: 声明式前端是 fast-agent 面向用户的那层 API——你用装饰器「描述」想要什么样的 agent,框架负责把描述变成实例。

它解决什么问题。 想象你要搭一个多 agent 应用:一个负责写代码、一个负责评审、再来一个 router 决定把请求发给谁。如果每个 agent 都要你手动 new 出对象、挂上 LLM、连好 MCP server、还要处理谁依赖谁的初始化顺序,代码会很啰嗦。fast-agent 把这些收进装饰器背后,你只写「声明」。

用起来什么样。 一个最小应用长这样:

from fast_agent import FastAgent   # 示意,非源码

fast = FastAgent("my-app")                      # ① 建应用对象

@fast.agent(name="assistant",                   # ② 声明一个 agent
            instruction="You are helpful.",
            servers=["fetch"])                   #    要连的 MCP server
async def main():
    async with fast.run() as agent:              # ③ 此刻才真正构造实例
        await agent.assistant.send("hi")

# asyncio.run(main())

一句话直觉: 把装饰器想成「点菜单」——你在菜单上勾选想要的菜(声明),但厨房(工厂)要等到 fast.run() 下单那一刻才开火做菜(实例化)。菜单和成品是两回事,这就是本章反复强调的**「声明与实例分离」**。

本节不出现底层代码。记住三步:建应用 → 挂装饰器登记配置 → run() 时工厂造实例。

2. 顶层全景(声明与实例分离)​

2.1 一张图看清流水线​

怎么读这张图:从上到下是时间顺序。左半边是「你写的声明」,在 import/装饰阶段就跑完;右半边是「框架造实例」,直到 fast.run() 才发生。中间的 self.agents 字典是两半的交接点。

  声明阶段(import 时,一装饰就跑)          实例化阶段(async with fast.run() 时)
  ─────────────────────────────           ──────────────────────────────────
   @fast.agent(...)                          fast.run()
        │  只登记,不构造                          │
        ▼                                        ▼
   _decorator_impl                          FastAgentRunLifecycle.enter
        │  组装 AgentConfig                       │  校验 / 载入 skills / 建 runtime
        ▼                                        ▼
   self.agents["name"] = {                  create_agents_in_dependency_order
     "config":  AgentConfig,   ◄────────────────┤  按依赖分组
     "type":    "agent",                         ▼
     "func":    <fn>,                        _AGENT_TYPE_BUILDERS[type]
     ...extra_kwargs                             │  每种类型一个 builder
   }                                             ▼
        └──────── 交接点 ────────►           活的 Agent 实例 → AgentApp

核心洞察: self.agents 是一个普通字典,键是 agent 名,值是「一份配置 + 元数据」,不是 agent 对象。声明阶段只写这个字典;实例化阶段只读这个字典。二者靠它解耦。

2.2 部件一句话职责​

部件干什么在哪个文件
FastAgent 应用对象用户入口:持有 agents 字典、解析 CLI、提供 run()src/fast_agent/core/fastagent.py:153
装饰器族(DecoratorMixin)@fast.agent/router/chain…,每个只登记一份配置src/fast_agent/core/direct_decorators.py:376
AgentConfig / AgentType声明的数据载体与类型枚举src/fast_agent/agents/agent_types.py:93 :24
工厂(direct_factory)读配置字典,按类型 + 依赖顺序造实例src/fast_agent/core/direct_factory.py:1088
运行时(ManagedRuntimeMixin)在 run() 里调工厂、把实例装进 AgentAppsrc/fast_agent/core/managed_runtime.py:168

2.3 主线走一遍(高层)​

  1. fast = FastAgent("app") 建对象,解析 CLI 参数,准备好空的 agents 字典。
  2. 每个 @fast.agent(...) 装饰器执行时,把一份 AgentConfig 连同元数据存进 fast.agents[name]。
  3. async with fast.run():先做前置校验,再调工厂 create_agents_in_dependency_order,按依赖顺序逐个把配置实例化成真正的 agent。
  4. 实例装进 AgentApp(agent.assistant.send(...) 里那个 agent 就是它),交给你用。

3. 核心机制​

3.1 FastAgent 应用对象:一个「组合」出来的门面​

它要解决的小问题: 用户只想面对一个 fast 对象,但它背后的能力(装饰、运行、服务器托管、AgentCard 热重载)其实分散在好几块。fast-agent 用**多继承混入(mixin)**把它们拼成一个类。

FastAgent 的定义就是一串 mixin 的组合:

class FastAgent(AgentCardRuntimeMixin, ManagedRuntimeMixin,
                FastAgentRunMixin, DecoratorMixin):
    ...

真实源码见 core/fastagent.py:153。每个 mixin 各管一摊,FastAgent 自己只管「初始化 + 配置加载 + 错误分类」:

Mixin负责的能力
DecoratorMixin提供 @fast.agent/smart/router/chain… 全部装饰器
FastAgentRunMixin提供 run() / start_server() 的运行生命周期
ManagedRuntimeMixin在 run() 内部调工厂、管理 agent 实例的生老病死
AgentCardRuntimeMixin从 Markdown/YAML AgentCard 文件加载与热重载 agent

__init__ 做了什么(core/fastagent.py:159)。 顺序很清楚,每步一件事:

  1. self.args = argparse.Namespace() 并调 _initialize_runtime_options,记下 quiet / 环境目录 / skills 目录等运行选项(:213)。
  2. 若 parse_cli_args=True,_parse_constructor_cli_args 解析命令行(:232)。
  3. _load_instance_settings 从 fast-agent.yaml + secrets 加载配置(:394)。
  4. self.app = Core(...) 建底层上下文(持有 config、MCP、skill registry 等)。
  5. _initialize_agent_registries() 把 self.agents = {} 等一堆注册表清空初始化(:426)。

CLI 参数解析(_constructor_arg_parser,:240)。 框架自带一套 argparse 参数,常用的几个:

参数作用
--model覆盖所有 agent 的默认模型
--agent + -m/--message给指定 agent 发一条消息后退出(headless)
-p/--prompt-file用一个 prompt 文件驱动
--transport {http,stdio,acp,a2a}以服务器模式运行,并选传输协议
--quiet关掉进度/日志显示,输出更干净

注意一个细节:只要命令行里出现了 --transport,_normalize_constructor_cli_server_flags 就把 self.args.server = True(:337),即「给了传输协议 = 想当服务器」。

fast.run() 的生命周期。 run() 是个异步上下文管理器(core/run_runtime.py:266),它把重活委托给 FastAgentRunLifecycle。enter() 的关键步骤(core/run_lifecycle.py:33),依次:

app.initialize()                     # 起底层上下文
  → _prepare_run_settings()          # 算出本次运行的 RunSettings(:688)
  → app.run() 进 exit_stack          # MCP 连接等
  → _load_default_skills_for_run()   # 载入 skills
  → _apply_skills_to_agent_configs() # 把 skills 灌进各 AgentConfig
  → _validate_run_preconditions()    # 前置校验(见 §3.4)
  → _create_run_runtime()            # 建 RunRuntime(模型工厂闭包等)

这里出现两个 frozen dataclass,值得点名:

  • RunSettings(core/fastagent.py:88)——本次运行的只读快照:quiet、模型覆盖、是否服务器模式、传输协议、是否 ACP 等,从 self.args + config 一次性算出(_prepare_run_settings,:688)。
  • RunRuntime(core/fastagent.py:102)——本次运行的共享资源:模型工厂闭包 model_factory_func、全局 prompt 上下文、shell executor、已托管实例列表等。

拿到这两个后,run() 才去 _initialize_managed_run_state 真正造 agent(下面 §3.4)。

3.2 装饰器族:每个只登记一份配置,绝不当场构造​

它要解决的小问题: 用户希望「写一行装饰器就有一个 agent」,但此刻 LLM、MCP server 都还没连上,依赖顺序也没算。所以装饰器不能当场构造——它只能先把「你想要什么」记下来。

统一入口 _decorator_impl。 所有装饰器(agent、smart、router、chain、parallel…)最终都汇到同一个函数 _decorator_impl(core/direct_decorators.py:310)。它返回一个真正的装饰器 decorator(func),而 decorator 干的事只有三件:

def decorator(func):                                   # 示意,精简自 :345
    config = _agent_config_from_decorator(...)         # ① 组装 AgentConfig
    self.agents[name] = _agent_data_from_decorator(    # ② 塞进注册字典
        config, agent_type, func, extra_kwargs)
    return _attach_agent_tool_decorator_if_supported(  # ③ 挂 .tool 子装饰器
        func, agent_type, config, extra_kwargs)

重点看第 ②步:登记的是一个 dict,不是 agent 对象。 _agent_data_from_decorator(:231)返回的结构是:

{
    "config": config,          # AgentConfig 实例
    "type":   agent_type.value,# 字符串,如 "agent" / "router" / "chain"
    "func":   func,            # 被装饰的那个 async 函数
    **extra_kwargs,            # 各类型特有的参数(见下)
}

装饰器族一览。 每种装饰器只是往 _decorator_impl 传不同的 AgentType 和不同的 extra_kwargs,把「这种 agent 需要的额外声明」记进字典:

装饰器登记的 AgentType特有 extra_kwargs(登记进字典的键)源码
@fast.agentBASICchild_agents、agents_as_tools_optionsdirect_decorators.py:434
@fast.smartSMART同上(智能编码 agent):529
@fast.customCUSTOMagent_class(用户自带类):601
@fast.orchestratorORCHESTRATORchild_agents、plan_type、plan_iterations:684
@fast.iterative_plannerITERATIVE_PLANNERchild_agents、plan_iterations:740
@fast.routerROUTERrouter_agents:789
@fast.chainCHAINsequence、cumulative:848
@fast.parallelPARALLELfan_out、fan_in、include_request:889
@fast.evaluator_optimizerEVALUATOR_OPTIMIZERgenerator、evaluator、min_rating、max_refinements:931
@fast.makerMAKERworker、k、max_samples、match_strategy:979

这些参数此刻只是被记下来;它们要到工厂阶段才被读出来用(§3.4)。工作流类装饰器(chain/parallel/router 等)的语义——比如 chain 怎么把上一个输出喂给下一个——见 工作流模式,本章不展开。

声明期就做的两件小事。 装饰器虽不构造实例,但会当场处理两类「纯声明」工作:

  • instruction 解析。 _resolve_instruction(:141)在装饰时就把 instruction 读出来:它可以是字符串、Path(读文件)、或 URL(拉取内容),并套用 {{currentDate}}、{{url:...}} 模板(_apply_templates,:98)。注意 {{file:...}} 模板故意留到运行时才解析,以便相对 workspaceRoot。
  • .tool 子装饰器。 对支持函数工具的类型,_attach_agent_tool_decorator_if_supported(:274)给被装饰函数挂一个 .tool,让你能 @my_agent.tool 注册只属于这个 agent 的本地 Python 工具,存进 config.function_tools。而 @fast.tool(DecoratorMixin.tool,:402)注册的是全局工具,进 self._registered_tools。

3.3 AgentConfig 与 AgentType:声明的数据载体​

它要解决的小问题: 「一个 agent 的所有声明」需要一个统一、可校验的容器。这就是 AgentConfig(agents/agent_types.py:93),一个 @dataclass。

字段速览(节选)。 一份 AgentConfig 就是一个 agent 的完整「规格书」:

字段含义
name / instruction名字与系统指令
servers要连的 MCP server 名列表
tools / resources / prompts按 server 名过滤 MCP 能力的映射(注意不是工具本身)
function_tools本地 Python 函数工具(可调用、字符串 spec 或 ScopedFunctionToolConfig)
skills / skill_manifests声明的 skills 与解析后的清单
model模型字符串(见 LLM 抽象)
use_history / save_trajectory是否保留对话历史 / 是否存轨迹
agent_typeAgentType 枚举,标明这是哪种 agent
default / tool_only是否默认 agent / 是否仅作为工具暴露

命名坑(源码注释专门点出)。 tools 这个名字在框架里有两种含义,容易混:

  • AgentConfig.tools = MCP 工具过滤器(按 server 名筛选发现到的工具)。
  • 运行时构造器如 ToolAgent(..., tools=...) 的 tools= = 解析好的、可执行的函数工具对象。
  • 而 function_tools 才是「本地 Python 工具」的声明。

__post_init__ 的两条校验(:134)。 dataclass 一构造就跑,做两件事:

  1. save_trajectory and use_history 同时为真 → 直接抛 AgentConfigError(二者互斥)。
  2. 若没给 default_request_params,就用 use_history 和 instruction 现造一个 RequestParams;若给了,也强制把它的 use_history/systemPrompt 对齐当前 config——保证 instruction 始终是权威系统提示。

AgentType 枚举(:24)。 一个 StrEnum,列出所有支持的类型:LLM / BASIC / SMART / CUSTOM / ORCHESTRATOR / PARALLEL / EVALUATOR_OPTIMIZER / ROUTER / CHAIN / ITERATIVE_PLANNER / MAKER / A2A。它是装饰器与工厂之间的路由键:装饰器写入 type,工厂据此选 builder。

3.4 工厂:把配置字典变成活的实例​

它要解决的小问题: 现在 self.agents 里躺着一堆配置,还有依赖关系(parallel 依赖它的 fan-out agents、router 依赖被路由的 agents)。工厂要按正确顺序把它们一个个造出来。

类型 → builder 的分派表。 工厂的核心是一张字典 _AGENT_TYPE_BUILDERS(core/direct_factory.py:1072),把每个 AgentType 映射到一个专属构造函数:

AgentTypebuilder 函数
BASIC / LLM_create_basic_agent(:687)
SMART_create_smart_agent(:728)
CUSTOM_create_custom_agent(:781)
ORCHESTRATOR / ITERATIVE_PLANNER_create_planner_agent(:834)
PARALLEL_create_parallel_workflow_agent(:860)
ROUTER_create_router_workflow_agent(:901)
CHAIN_create_chain_workflow_agent(:926)
EVALUATOR_OPTIMIZER_create_evaluator_optimizer_agent(:957)
MAKER_create_maker_agent(:988)
A2A_create_a2a_agent(:1041)

依赖顺序是怎么保证的。 入口是 create_agents_in_dependency_order(:1165):它先用 get_dependencies_groups 把 agents 拓扑分组,再一组组地实例化;每组内部 create_agents_by_type(:1088)按类型调对应 builder。这样当 router 被造时,它要路由的子 agent 早已在 active_agents 里就绪。

一个 basic agent 是怎么被装配出来的。 看 _create_basic_agent(:687)→ _finalize_agent(:350),这是最典型的路径。_finalize_agent 四步走,每步一件事:

async def _finalize_agent(...):                 # 示意,精简自 :350
    await agent.initialize()                     # ① 起 MCP 聚合器等
    llm_factory = model_factory_func(model=config.model)
    await agent.attach_llm(llm_factory, ...)      # ② 挂上 LLM
    _apply_tool_hooks(agent, config, ...)         # ③ 装工具钩子
    _register_loaded_agent(result_agents, name, agent)  # ④ 登记为「已就绪」

关键装配 hook。 工厂在这里把散落的声明「焊」到实例上:

  • 挂 LLM。 _initialize_agent_with_llm(:318)/_finalize_agent 都会用 model_factory_func 造出 LLM 工厂,再 agent.attach_llm(...),把 config.default_request_params、config.api_key 传进去。模型字符串怎么解析见 LLM 抽象。
  • 函数工具落地。 _resolve_function_tools_with_globals(:173)决定这个 agent 用哪些函数工具:若它显式声明了 function_tools(哪怕空列表)就只用自己的;否则回退到全局 @fast.tool 注册的工具。
  • agents-as-tools。 若一个 basic/smart agent 声明了 child_agents,_attach_child_agents_as_tools(:332)会把子 agent 包成工具挂上去(否则走普通 McpAgent 路径)。
  • 工具钩子。 _apply_tool_hooks(:566)按 config 装上历史裁剪、自动压缩、会话历史持久化等 after-turn 钩子。

工厂从哪被调起。 闭环发生在 ManagedRuntimeMixin._instantiate_agent_instance(core/managed_runtime.py:168):它在 run() 生命周期里调 create_agents_in_dependency_order(self.app, self.agents, ..., global_function_tools=self._registered_tools)(:176),拿到 agents_map 后包进 AgentApp —— 这就是你 async with fast.run() as agent 里的 agent。

别忘了前置校验。 造实例前,_validate_run_preconditions(core/fastagent.py:788)先把关:没有任何 agent → 报错;还会 validate_server_references(引用的 MCP server 都存在吗)和 validate_workflow_references(chain/router 引用的子 agent 都声明了吗)。声明期允许你引用还不存在的名字,校验期统一兜底。

4. 巧妙之处(可借鉴的技术)​

  • 声明与实例彻底分离,交接点是一个普通 dict。 装饰器只写 self.agents[name] = {...},工厂只读它。好处:声明可以乱序、可以互相前向引用(router 先声明、被路由的 agent 后声明都行),因为解析/校验/拓扑排序全推迟到 run()。见 direct_decorators.py:364 与 direct_factory.py:1165。

  • 所有装饰器复用同一个 _decorator_impl。 十来个装饰器差别只在 AgentType + 少量 extra_kwargs,主体逻辑零重复(direct_decorators.py:310)。新增一种 agent 类型,基本只需加一个薄装饰器 + 一个 builder + 在 _AGENT_TYPE_BUILDERS 注册一行。

  • 类型枚举当路由键。 AgentType 把「装饰器写入」和「工厂读取」用一个字符串键对齐(agent_types.py:24 ↔ direct_factory.py:1072),两端解耦又不失一致。

  • 配置对象自我校验、自我兜底。 AgentConfig.__post_init__(:134)在构造瞬间就拦下互斥选项、并保证 instruction 永远是权威系统提示——非法状态根本无法存在。

  • RunSettings 是不可变快照。 一次运行的所有开关在 _prepare_run_settings 里一次算清、冻结成 frozen dataclass(core/fastagent.py:88),运行途中不再受 self.args 变动影响,行为可预期。

5. 边界与局限​

  • 装饰器不校验依赖是否存在。 你可以 @fast.chain(sequence=["a","b"]) 引用尚未声明的 a;错误直到 run() 期 validate_workflow_references 才暴露(core/fastagent.py:794)。声明期唯一的即时校验是空 sequence(direct_decorators.py:871)和 custom agent 的函数工具兼容性(:174)。

  • save_trajectory 与 use_history 互斥。 同时开会在 AgentConfig.__post_init__ 直接抛错(agent_types.py:136)。

  • tools= 的双重含义是已知的命名陷阱。 源码用注释反复澄清(agent_types.py:97、direct_factory.py:184):AgentConfig.tools 是 MCP 过滤器,构造器 tools= 是解析后的函数工具对象——阅读时务必分清。

  • 本章不覆盖的部分: 各 agent 类内部如何跑(→ Agent 类栈)、一次 turn 的工具循环(→ 工具循环引擎)、工作流语义(→ 工作流模式)、MCP 聚合细节(→ MCP 集成)。

6. 代码地图(导航索引)​

主题文件路径符号名
应用对象(mixin 组合)src/fast_agent/core/fastagent.pyFastAgent
应用初始化src/fast_agent/core/fastagent.pyFastAgent.__init__
CLI 参数定义src/fast_agent/core/fastagent.py_constructor_arg_parser
运行快照 / 共享资源src/fast_agent/core/fastagent.pyRunSettings / RunRuntime
运行前置校验src/fast_agent/core/fastagent.py_validate_run_preconditions
run() 生命周期src/fast_agent/core/run_runtime.pyFastAgentRunMixin.run
生命周期 enter/exitsrc/fast_agent/core/run_lifecycle.pyFastAgentRunLifecycle.enter
装饰器统一实现src/fast_agent/core/direct_decorators.py_decorator_impl
登记进字典src/fast_agent/core/direct_decorators.py_agent_data_from_decorator
装饰器族src/fast_agent/core/direct_decorators.pyDecoratorMixin.agent / .router / .chain …
instruction 解析src/fast_agent/core/direct_decorators.py_resolve_instruction / _apply_templates
配置数据类src/fast_agent/agents/agent_types.pyAgentConfig / AgentConfig.__post_init__
类型枚举src/fast_agent/agents/agent_types.pyAgentType
类型→builder 分派表src/fast_agent/core/direct_factory.py_AGENT_TYPE_BUILDERS
按依赖顺序实例化src/fast_agent/core/direct_factory.pycreate_agents_in_dependency_order / create_agents_by_type
basic agent 装配src/fast_agent/core/direct_factory.py_create_basic_agent / _finalize_agent
挂 LLM / 函数工具 / 子 agentsrc/fast_agent/core/direct_factory.py_initialize_agent_with_llm / _resolve_function_tools_with_globals / _attach_child_agents_as_tools
工厂调用点(run→factory)src/fast_agent/core/managed_runtime.py_instantiate_agent_instance