两只熊骑着双人自行车aoa.run
教程 →
课程 01 / 28

Action 与流水线

AOA 的一切都围绕一个核心概念展开——Action。它把一次业务操作变成一个类:一个带类型的输入(Params)、一个输出(Result),中间是一条自上而下、像阅读页面一样逐行读下去的步骤链。下面每个小节都对应 examples/step_01_Action_and_pipeline/ 中的一个真实文件——点击运行看它实际捕获的输出,或者在 Colab 中打开、自己动手跑一遍。

← 课程 00 — 开始使用课程 02 — State：操作的 X 光片 →

Hello, world!

让我们从一个什么正经事都不做的 Action 开始——它只打印一行,然后返回一个空的占位结果。它的价值不在于此:它展示的是,在 AOA 里要让任何代码跑起来,最少需要声明哪些东西。

01_hello_world.py
class GreetingDomain(BaseDomain):    name = "greeting"    description = "Greetings domain"@meta(description="Say hello to the world", domain=GreetingDomain)@check_roles(GuestRole)class SayHelloAction(BaseAction[ParamsStub, ResultStub]):    @summary_aspect("Print greeting and return stub")    async def output_summary(self, params, state, box, connections):        print("Hello, world!")        return ResultStub()
$ uv run python examples/step_01_Action_and_pipeline/01_hello_world.py
▶ 在 Colab 中试运行 ↗在 GitHub 中打开 ↗
输出

对于一句简单的问候来说,这行数不算少——但没有一行是走过场:每一行都在声明一样东西,少了它,AOA 中的 Action 就不算完整。一切始于一个Domain。@meta 是 Action 的身份护照;@check_roles(GuestRole) 把访问权限明确声明出来,而不是默认开放。@summary_aspect 则是唯一的出口——这三者只要少一个,机器就会在第一次调用之前拒绝运行这个 Action。

Params、Result 与 box

占位代码适合用来入门,但真正的 Action 要处理数据。我们给它加上一个输入和一个输出——顺便把 print 换成 AOA 中 Action 与外部世界对话时真正会用的工具。

02_params_result_and_box.py
class GreetParams(BaseParams):    name: str = Field(description="Name of the person to greet")class GreetResult(BaseResult):    message: str = Field(description="Assembled greeting message")@meta(description="Greet a person by name", domain=GreetingDomain)@check_roles(GuestRole)class GreetPersonAction(BaseAction[GreetParams, GreetResult]):    @summary_aspect("Build greeting and return result")    async def greet_summary(self, params, state, box, connections):        await box.info(            Channel.business,            "Greeting: Hello, {%var.name|cyan}!",            name=params.name,        )        return GreetResult(message=f"Hello, {params.name}!")
$ uv run python examples/step_01_Action_and_pipeline/02_params_result_and_box.py
▶ 在 Colab 中试运行 ↗在 GitHub 中打开 ↗
输出

box 比 print 有意思得多——它是绑定到当前步骤的结构化日志器,携带 channel、level、domain、action 与 aspect 名称等信息,因此天然适合过滤、路由与导出。事件最终流向哪里并不由 Action 决定——那是机器的事,由外部接入。

多个 aspect

一个步骤往往是不够的。AOA 不允许中间数据藏身于局部变量或对象字段中——它以显式的 state 流经整条流水线,而 Action 本身在两次调用之间始终保持为空。

03_multiple_aspects.py
@meta(description="Process input string through multiple steps", domain=ProcessingDomain)@check_roles(GuestRole)class ProcessInputAction(BaseAction[ProcessParams, ProcessResult]):    @regular_aspect("Step 1: Strip whitespace and lowercase")    @result_string("cleaned", required=True)    async def validate_aspect(self, params, state, box, connections):        cleaned = params.raw_input.strip().lower()        return {"cleaned": cleaned}    @regular_aspect("Step 2: Enrich data")    @result_string("cleaned", required=True)    @result_string("enriched", required=True)    async def enrich_aspect(self, params, state, box, connections):        enriched = f"enriched::{state['cleaned']}"        return {"cleaned": state["cleaned"], "enriched": enriched}    @summary_aspect("Step 3: Assemble final result")    async def assemble_summary(self, params, state, box, connections):        return ProcessResult(            cleaned=state["cleaned"],            enriched=state["enriched"],            final=f"{state['cleaned']} → {state['enriched']}",        )
$ uv run python examples/step_01_Action_and_pipeline/03_multiple_aspects.py
▶ 在 Colab 中试运行 ↗在 GitHub 中打开 ↗
输出

返回的字典会整体替换 state——而不是在其基础上扩展。@result_string("cleaned", required=True) 是一个checker:对 aspect 输出的可验证契约。一旦违反承诺,机器会当场终止该 aspect,绝不让被污染的 state 继续向下传递。

继承

Action 的继承方式和普通类没什么两样——但有一个刻意为之的例外:aspect 不会被继承进流水线。机器只根据类自身声明的内容来构建流水线。

04_inheritance.py
# Parent: two aspects in the pipelineclass BaseOrderAction(BaseAction[OrderParams, OrderResult]):    @regular_aspect("Validate order")    async def validate_aspect(self, ...): ...    @summary_aspect("Base result")    async def base_summary(self, ...): ...# Child: declares only its own summary — validate_aspect will NOT runclass ChildOrderAction(BaseOrderAction):    @summary_aspect("Child result")    async def child_summary(self, ...): ...# The right way: declare the aspect explicitly and call super()class ExtendedOrderAction(BaseOrderAction):    @regular_aspect("Validate order")    @result_instance("steps", list, required=True)    async def validate_aspect(self, params, state, box, connections):        result = await super().validate_aspect(params, state, box, connections)        return {**result, "extended": True}    @summary_aspect("Extended result")    async def extended_summary(self, ...): ...
$ uv run python examples/step_01_Action_and_pipeline/04_inheritance.py
▶ 在 Colab 中试运行 ↗在 GitHub 中打开 ↗
输出

ChildOrderAction 能正常运行,但它的流水线里只有 child_summary——父类的 validate_aspect 永远不会执行。ExtendedOrderAction 的做法才对:它重新声明该 aspect,并调用 super() 在父类逻辑的基础上继续构建。

实验

这一章积累了不少规则——但它们有一个共同点:几乎所有规则都是在类声明时、也就是模块导入阶段被检查的,而不是在 Action 调用时。这里的代码就是可执行的规范;违反契约会在第一次运行之前暴露出来,而不是在生产环境的某个深夜。动手试试——从第一个例子中选一种方式破坏 SayHelloAction,然后点击运行。

01_hello_world.py
class GreetingDomain(BaseDomain):    name = "greeting"    description = "Greetings domain"@meta(description="Say hello to the world", domain=GreetingDomain)@check_roles(GuestRole)class SayHelloAction(BaseAction[ParamsStub, ResultStub]):    @summary_aspect("Print greeting and return stub")    async def output_summary(self, params, state, box, connections):        print("Hello, world!")        return ResultStub()
$ uv run python examples/step_01_Action_and_pipeline/01_hello_world.py
▶ 在 Colab 中试运行 ↗在 GitHub 中打开 ↗
输出

小结

模型的核心你已经掌握了:一次操作就是一个 Action,拥有带类型的边界(Params 与 Result)、一条线性的 aspect 流水线,以及大多在初始化阶段就被检查的契约。行为体现在代码结构本身,而不是躺在旁边的文档里。

复习题

  1. 哪一条不变量保证了 Action “单一类可读”的特性?它又是在什么时刻被检查的?
  2. 为什么缺少 @check_roles 会被视为错误,而不是默默地“对所有人开放”?
  3. AOA 的流水线是线性的——没有分支,也没有旁路出口。这个约束换来了什么?又付出了什么代价?
  4. 为什么 aspect 不会自动被继承进流水线?对比一下 OOP 中普通的方法继承。
  5. “代码即可执行的规范”是什么意思?为什么大多数契约是在初始化阶段被检查,而不是在运行时?
← 课程 00 — 开始使用课程 02 — State：操作的 X 光片 →
第 01 步 01 — Action 与流水线 · aoa.run