声明式前端: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() 里调工厂、把实例装进 AgentApp | src/fast_agent/core/managed_runtime.py:168 |
2.3 主线走一遍(高层)
fast = FastAgent("app")建对象,解析 CLI 参数,准备好空的agents字典。- 每个
@fast.agent(...)装饰器执行时,把一份AgentConfig连同元数据存进fast.agents[name]。 async with fast.run():先做前置校验,再调工厂create_agents_in_dependency_order,按依赖顺序逐个把配置实例化成真正的 agent。- 实例装进
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)。 顺序很清楚,每步一件事:
self.args = argparse.Namespace()并调_initialize_runtime_options,记下 quiet / 环境目录 / skills 目录等运行选项(:213)。- 若
parse_cli_args=True,_parse_constructor_cli_args解析命令行(:232)。 _load_instance_settings从fast-agent.yaml+ secrets 加载配置(:394)。self.app = Core(...)建底层上下文(持有 config、MCP、skill registry 等)。_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.agent | BASIC | child_agents、agents_as_tools_options | direct_decorators.py:434 |
@fast.smart | SMART | 同上(智能编码 agent) | :529 |
@fast.custom | CUSTOM | agent_class(用户自带类) | :601 |
@fast.orchestrator | ORCHESTRATOR | child_agents、plan_type、plan_iterations | :684 |
@fast.iterative_planner | ITERATIVE_PLANNER | child_agents、plan_iterations | :740 |
@fast.router | ROUTER | router_agents | :789 |
@fast.chain | CHAIN | sequence、cumulative | :848 |
@fast.parallel | PARALLEL | fan_out、fan_in、include_request | :889 |
@fast.evaluator_optimizer | EVALUATOR_OPTIMIZER | generator、evaluator、min_rating、max_refinements | :931 |
@fast.maker | MAKER | worker、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_type | AgentType 枚举,标明这是哪种 agent |
default / tool_only | 是否默认 agent / 是否仅作为工具暴露 |
命名坑(源码注释专门点出)。 tools 这个名字在框架里有两种含义,容易混:
AgentConfig.tools= MCP 工具过滤器(按 server 名筛选发现到的工具)。- 运行时构造器如
ToolAgent(..., tools=...)的tools== 解析好的、可执行的函数工具对象。 - 而
function_tools才是「本地 Python 工具 」的声明。
__post_init__ 的两条校验(:134)。 dataclass 一构造就跑,做两件事:
save_trajectory and use_history同时为真 → 直接抛AgentConfigError(二者互斥)。- 若没给
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 映射到一个专属构造函数:
| AgentType | builder 函数 |
|---|---|
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.py | FastAgent |
| 应用初始化 | src/fast_agent/core/fastagent.py | FastAgent.__init__ |
| CLI 参数定义 | src/fast_agent/core/fastagent.py | _constructor_arg_parser |
| 运行快照 / 共享资源 | src/fast_agent/core/fastagent.py | RunSettings / RunRuntime |
| 运行前置校验 | src/fast_agent/core/fastagent.py | _validate_run_preconditions |
run() 生命周期 | src/fast_agent/core/run_runtime.py | FastAgentRunMixin.run |
| 生命周期 enter/exit | src/fast_agent/core/run_lifecycle.py | FastAgentRunLifecycle.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.py | DecoratorMixin.agent / .router / .chain … |
| instruction 解析 | src/fast_agent/core/direct_decorators.py | _resolve_instruction / _apply_templates |
| 配置数据类 | src/fast_agent/agents/agent_types.py | AgentConfig / AgentConfig.__post_init__ |
| 类型枚举 | src/fast_agent/agents/agent_types.py | AgentType |
| 类型→builder 分派表 | src/fast_agent/core/direct_factory.py | _AGENT_TYPE_BUILDERS |
| 按依赖顺序实例化 | src/fast_agent/core/direct_factory.py | create_agents_in_dependency_order / create_agents_by_type |
| basic agent 装配 | src/fast_agent/core/direct_factory.py | _create_basic_agent / _finalize_agent |
| 挂 LLM / 函数工具 / 子 agent | src/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 |