ステップ 01 / 28

Action とパイプライン

AOA のすべては一つの図形を軸に回る — Action だ。クラスとして書き表された業務操作であり、型のついた入口が一つ(Params)、出口が一つ(Result)、そのあいだにはページを読むように上から下へたどれる一本の手順の連なりがある。以下の各節は examples/step_01_Action_and_pipeline/ にある実際のファイルに対応している — Run を押せば本物の出力が見られるし、Colab で開いて自分で走らせてもいい。

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
出力

これほど簡単な挨拶にしては行数が多い — だが儀式は一行もない。どれも、それなしでは AOA の Action が完成したと見なされないものを宣言している。すべては domain から始まる。@meta は Action の身分証。@check_roles(GuestRole) はアクセス権を暗黙にせず声に出して宣言する。そして @summary_aspect は唯一の出口だ。三つのどれか一つでも欠ければ、最初の呼び出しを待たずに機械が起動を拒む。

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
出力

boxprint よりずっと面白い — 今いる手順に結びついた構造化ロガーだ。チャネル、レベル、domain、action 名、aspect を運ぶので、絞り込みも振り分けも書き出しも容易にできる。出来事がどこへ行くかは Action 自身が決めることではない。それは機械の仕事であり、外から差し込まれる。

複数のアスペクト

一つの手順で足りることは少ない。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
出力

返された辞書は state丸ごと置き換える — 付け足すのではない。@result_string("cleaned", required=True)checker であり、アスペクトの出力に対する検査可能な契約だ。約束を破れば、機械はただちにアスペクトを止め、壊れた state を先へ通さない。

継承

Action は普通のクラスと同じように継承できる — ただし意図された例外が一つある。アスペクトはパイプラインに継承されない。機械はそのクラス自身に宣言されたものだけからパイプラインを組み立てる。

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
出力

ChildOrderAction はきちんと動くが、そのパイプラインに入っているのは child_summary だけだ — 親の validate_aspect は決して実行されない。ExtendedOrderAction は正しくやっている。アスペクトを改めて宣言し、super() を呼んで親の処理に乗っている。


実験

この章では規則がずいぶん積み上がった — が、それらには共通の性質がある。ほとんどが クラス宣言の時点で、モジュールの読み込み時に検査される。Action の呼び出し時ではない。ここでのコードは実行可能な仕様であり、契約違反は最初の実行より前に表に出る。ある夜の本番で、ではなく。自分で試してみてほしい — 最初の例の SayHelloAction を壊す方法を見つけて Run を押そう。

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
出力

まとめ

モデルの骨格はもう組み上がった。操作とは、型のついた境界(ParamsResult)と、アスペクトの一本のパイプラインと、その多くが初期化時に検査される契約とを備えた Action である。振る舞いはコードの構造そのもので言い表されていて、コードとは別に存在する文書の中にはない。


確認問題

  1. Action を「クラス一つで」読めるものにしている不変条件は何か。そしてそれはいつ検査されるか。
  2. @check_roles がないことが、黙って「全員に公開」ではなく誤りとして扱われるのはなぜか。
  3. AOA のパイプラインは一本道で、分岐も脇道の出口もない。この制約は何をもたらし、その代償は何か。
  4. アスペクトがパイプラインへ自動で継承されないのはなぜか。オブジェクト指向における通常のメソッド継承と比べてみよう。
  5. 「コードは実行可能な仕様である」とはどういう意味か。そして契約の大半が実行時ではなく初期化時に検査されるのはなぜか。
ステップ 01 — Action とパイプライン · aoa.run