FastPilot

FastPilot 让一个持续存在的 Python 业务对象，拥有统一的控制、观察与通信方式。

它适合这样的开发过程：你先在 IPython 里写出一个普通业务对象，然后逐步让 它能够接收类型化命令、保存需要继续判断的数据、发布领域事实、接受未来动作， 并在需要时由浏览器、远程 Python 或自动程序继续推动。

这段过程中，业务保持为你的 Python 代码。FastPilot 提供的是对象周围那条 可以继续观察和控制的公共因果链。

安装

FastPilot 需要 Python 3.13 或更高版本：

pip install fastpilot

核心包不强制安装网络、服务或 IPython 依赖。需要对应能力时，再安装可选依赖：

pip install "fastpilot[ipython]"
pip install "fastpilot[remote]"
pip install "fastpilot[service]"

Command
    ↓
Participant
    ↓
Runtime
    ↓
Fact / Event / Snapshot / CommandReceipt
    ↓
IPython、远程 Python、浏览器、自动程序

先看一种日常写法

假设应用已经有一个普通的目录 SDK。现在我们想把“刷新目录”建模成一项可以 观察的业务动作，并保留后续判断需要的计数。

下面的代码可以按 IPython 的 # %% cell 逐段执行，也可以保存成普通 Python 脚本运行。每个 cell 都继续使用前面已经创建的对象。

第一格：声明控制输入和领域事实

# %%
from dataclasses import dataclass
from typing import Literal

from fastpilot import (
    Command,
    Fact,
    ParticipantContext,
    command,
    emits,
    fact,
    handles,
    make_runtime,
)


@command("catalog.refresh")
@dataclass(frozen=True, slots=True)
class RefreshCatalog(Command):
    scope: Literal["recent", "all"]


@fact("catalog.refreshed")
@dataclass(frozen=True, slots=True)
class CatalogRefreshed(Fact):
    count: int
    scope: str

RefreshCatalog 表达“希望目录刷新”的控制输入；CatalogRefreshed 表达 “目录已经完成刷新”的领域事实。它们都是普通的不可变 Python dataclass，字段 就是业务需要传递的信息。

第二格：把已有业务对象接入 Runtime

# %%
class CatalogClient:
    def fetch_catalog(self, scope: str) -> list[str]:
        records = ["book", "pen", "lamp"]
        return records if scope == "all" else records[:2]


class CatalogRefresher:
    initial_data = {"refresh_count": 0}

    def __init__(self, client: CatalogClient) -> None:
        self.client = client

    @handles
    @emits(CatalogRefreshed)
    def refresh(
        self,
        context: ParticipantContext,
        command: RefreshCatalog,
    ) -> None:
        records = self.client.fetch_catalog(command.scope)
        refresh_count = context.read_data("refresh_count", 0)
        assert isinstance(refresh_count, int)

        context.write_data("last_count", len(records))
        context.write_data("refresh_count", refresh_count + 1)
        context.emit(CatalogRefreshed(len(records), command.scope))


catalog = make_runtime(CatalogRefresher(CatalogClient()))

CatalogRefresher 是普通 Python 类。它通过构造参数持有已有的 CatalogClient，业务依赖由应用自己组织。

@handles 把 RefreshCatalog 交给 refresh；@emits 声明这个 handler 可以发布 CatalogRefreshed。context 是这次控制轮次使用的工作材料： 成功结束以后，写入的数据和发布的事实一起进入 Runtime 的新边界。

第三格：观察并推动一次控制轮次

# %%
refreshed = catalog.subscribe(CatalogRefreshed)

catalog.start()
receipt = catalog.dispatch(RefreshCatalog("recent"))

print(receipt.outcome)
print(dict(receipt.snapshot.data))
print(refreshed.next(timeout=1).fact)

你会得到类似这样的结果：

completed
{'refresh_count': 1, 'last_count': 2}
CatalogRefreshed(count=2, scope='recent')

这里有三种互相配合的观察方式：

• receipt 说明这一轮命令的结果；
• receipt.snapshot 保存这一轮结束时的完整边界；
• subscribe() 让另一个消费者沿自己的游标读取事实。

下一格可以继续使用 catalog、receipt 和 refreshed。这就是 FastPilot 最常见的工作感受：先让对象活在工作台里，再根据当前观察决定下一步。

这段代码解决了什么问题

普通函数通常适合这样的调用：

result = refresh_catalog(scope="recent")

当业务开始需要持续存在时，调用者还会遇到另外一些问题：

• 上一次动作留下了什么数据？
• 这次命令实际产生了哪些事实？
• 另一个消费者怎样从自己的位置继续观察？
• 一个动作能不能稍后执行、重排或取消？
• 浏览器和远程 Python 怎样继续控制同一个对象？

FastPilot 把这些信息组织成 Runtime 的控制与观察边界。业务代码决定 自己的 Command、Fact 和数据；Runtime 负责把每次控制动作收束为可以继续 判断的 CommandReceipt、Event 和 Snapshot。

三个需要记住的角色

Participant：你的业务规则

Participant 是普通 Python 对象。它拥有领域依赖和处理 Command 的方法：

class CatalogRefresher:
    @handles
    @emits(CatalogRefreshed)
    def refresh(self, context: ParticipantContext, command: RefreshCatalog) -> None:
        ...

你可以从一个只有事件处理的 Participant 开始；当业务确实需要逻辑时间、 状态图或未来队列时，再让它声明对应能力。

Runtime：持续存在的业务世界

catalog = make_runtime(CatalogRefresher(CatalogClient()))

catalog.start()
catalog.dispatch(RefreshCatalog("recent"))
catalog.snapshot()
catalog.history()

Runtime 保存当前生命周期中的领域数据、事实历史、观察游标、队列、转换草稿、 可选逻辑时间和子 Runtime。它是同一条控制链的权威拥有者。

Driver：根据观察选择下一步的人或程序

Driver 是一个角色，可以由下面这些对象承担：

• 持久 IPython 工作台；
• 远程 Python 程序；
• 浏览器控制台；
• 根据事件自动发送下一条命令的程序。

它们都可以沿着同一个 Runtime 的 dispatch()、snapshot()、history() 和 subscribe() 继续工作。

让未来动作先成为一个值

Command 可以立即执行，也可以登记为未来意图：

# %%
from fastpilot import AtTime
from fastpilot.contrib import SetEnergy, make_robot

robot = make_robot()
robot.start()
entry = robot.enqueue(
    SetEnergy(50),
    trigger=AtTime(20),
)

robot.move_entry(entry.id, 0)
robot.cancel_entry(entry.id, "计划改变")

队列条目拥有稳定身份，因此未来动作可以被观察、重排、取消，并在历史中留下 自己的终态。事件驱动业务可以直接停留在前面的几格代码；需要时间能力的业务 再进入 Trigger 和队列。

需要状态图时，再加入转换站

状态图用于表达领域允许的状态边：

# %%
from fastpilot import StateChart

statechart = StateChart.create(
    initial="IDLE",
    transitions={
        "IDLE": {"WALKING"},
        "WALKING": {"RESTING"},
        "RESTING": {"WALKING"},
    },
)

这个值可以由 Participant 作为自己的状态图提供：

class RobotParticipant:
    statechart = statechart

    # 这里继续放置 Walk、Obstacle 等领域 Command 的 handler。
    ...

当领域 Command 提出状态变化时，Runtime 可以把下一条状态边停在一个可观察、 可编辑的 TransitionDraft：

# %%
from fastpilot.contrib import Walk, make_robot

robot = make_robot()
robot.dispatch(Walk())

assert robot.pending is not None
robot.pending.context["operator_note"] = "沿北侧通道"
robot.commit()
robot.resume()

这里的日常表达很直接：先看提案，再补充工作材料，提交状态边，最后恢复后续 调度。需要这类人工介入的业务，可以在 examples/03-robot 中继续阅读。

需要网络时，让同一个对象打开入口

本地 Runtime 可以在当前进程中打开 HTTP/SSE 入口：

# %%
service = catalog.serve(port=0)
print(service.url)

另一个 Python 进程使用同一个协议连接它：

from fastpilot import connect

with connect(service.url) as remote:
    remote.dispatch(RefreshCatalog("all"))

远程调用发送同一种 RefreshCatalog Command，服务端由同一个 Runtime 完成解释、记录事件和生成快照。

如果应用本来就有 FastAPI，可以把控制面挂入已有 ASGI 应用：

from typing import Literal

from fastapi import FastAPI
from pydantic import BaseModel
from fastpilot.service import mount_runtime, serve_app


class RefreshRequest(BaseModel):
    scope: Literal["recent", "all"]


app = FastAPI()
mount_runtime(app, catalog, prefix="/_pilot")


@app.post("/catalog/refresh")
def refresh(request: RefreshRequest) -> dict[str, object]:
    receipt = catalog.dispatch(RefreshCatalog(request.scope))
    return {
        "outcome": receipt.outcome,
        "sequence": receipt.snapshot.sequence,
    }


service = serve_app(app, port=0)

这段组合里：

• RefreshRequest 是 HTTP 边界模型；
• RefreshCatalog 是 Runtime 控制意图；
• /catalog/refresh 是应用自己的业务地址；
• /_pilot/v1/* 是同一个 Runtime 的通用控制与观察地址。

普通 FastAPI 路由与 FastPilot 控制面因此可以在一个应用里共同工作。

从哪里继续

这份 README 先覆盖安装和最小使用旅程。仓库中的 examples/、docs/、 tests/ 和 CONTRIBUTING.md 属于开发维护材料，当前不随 PyPI 发布包提供； 它们会在整理成熟后再决定是否公开。