Apache Burr 快速入门与集成示例

Apache Burr 0.42.0 快速入门与集成示例：环境准备、状态机应用构建与条件转移、本地 tracker 与 Web UI 可视化、pytest 测试用例生成、FastAPI 服务形态集成，均含真实命令与实测回显。

Apache BurrBurr2026/10/14

Apache Burr 快速入门与集成示例

目标：以 burr 0.42.0（实际安装主体 apache-burr 0.42.0）为基线，从环境准备开始快速构建第一个状态机应用，并给出与本地跟踪、Web UI 可视化、pytest 测试用例生成、FastAPI 服务形态的四类集成示例。以下命令与回显均在真实环境逐步实测（Ubuntu 22.04，Python 3.10.12）。

一、框架定位与核心概念

Apache Burr（Incubating）是一个用 Python 构建"会做决策的应用"的轻量状态机框架，典型场景包括聊天机器人、智能体、模拟仿真等。

💡 技术拓展：Python 工作流框架生态中，Prefect、Airflow 侧重大规模调度与分布式编排，Temporal 主打长期运行工作流与持久化执行，而 Apache Burr 主打轻量级状态机与可观测性，单机 pip 安装即可快速跑通，无需调度集群。

核心概念：

  • Action（动作）：被 @action 装饰的函数，通过 reads/writes 声明状态读写契约；
  • State（状态）：不可变状态对象，动作通过 state.update 产生新状态；
  • Transition（转移）：(from, to, condition) 三元组定义状态机边，条件用 expr 表达式或 default 兜底；
  • ApplicationBuilder：链式构建应用的入口；
  • Tracker（跟踪器）：记录执行轨迹，供 Web UI 可视化。

二、环境准备

创建虚拟环境并安装带 start 扩展的 burr（start 提供 CLI 与 Web UI 依赖）：

python3 -m venv /opt/burr-demo/.venv
. /opt/burr-demo/.venv/bin/activate
pip install --proxy http://192.168.1.62:7897 "burr[start]"

安装实测回显（关键行）：

Collecting burr
  Downloading burr-0.42.0-py3-none-any.whl (2.3 kB)
Collecting apache-burr==0.42.0
  Downloading apache_burr-0.42.0-py3-none-any.whl (4.7 MB)
Successfully installed apache-burr-0.42.0 burr-0.42.0

三、第一个状态机应用

程序示例 app.py：定义 counter（累加器）与 summarize（汇总器）两个动作，counter 计数满 3 后转移至 summarize 并暂停。

from burr.core import action, State, ApplicationBuilder, expr, default

@action(reads=[], writes=["count"])
def counter(state: State) -> State:
    return state.update(count=state["count"] + 1)

@action(reads=["count"], writes=["result"])
def summarize(state: State) -> State:
    return state.update(result=f"count reached {state['count']}")

app = (
    ApplicationBuilder()
    .with_actions(counter=counter, summarize=summarize)
    .with_transitions(
        ("counter", "counter", expr("count < 3")),
        ("counter", "summarize", default),
    )
    .with_state(count=0, result="")
    .with_entrypoint("counter")
    .with_tracker(project="burr-quickstart")
    .build()
)

last_action, result, state = app.run(halt_after=["summarize"])
print("stopped after action:", last_action.name)
print("result:", result)
print("final state:", state)

运行命令与实测回显：

python3 app.py
stopped after action: summarize
result: {}
final state: {'count': 3, 'result': 'count reached 3', '__SEQUENCE_ID': 3, '__PRIOR_STEP': 'summarize'}

执行解读：入口动作 counter 连续执行 3 次，每次 count 加 1；第三次执行后条件 count < 3 不成立，命中 default 兜底转移至 summarize；halt_after 使应用在 summarize 完成后暂停，返回动作名、结果与最终状态。__SEQUENCE_ID 与 __PRIOR_STEP 为框架内部字段，分别记录执行序列号与上一步动作。

四、条件转移与默认分支

expr("count < 3") 将字符串表达式编译为条件对象，default 作为无匹配时的兜底。实测坑：第三项直接传字符串会在构建时报 AttributeError: 'str' object has no attribute 'name'，必须使用 expr 或 default 包装。落盘的 graph.json 中转移结构实测回显：

{"type": "transition", "from_": "counter", "to": "counter", "condition": "count < 3"}
{"type": "transition", "from_": "counter", "to": "summarize", "condition": "default"}

五、集成示例一：本地跟踪与 Web UI 可视化

with_tracker(project="burr-quickstart") 将每次运行的图结构、元数据与逐步骤日志写入 ~/.burr/burr-quickstart/<app_id>/，实测回显：

ls ~/.burr/burr-quickstart/1b853a06-13d9-44e2-83a0-55897f602302/
graph.json
log.jsonl
metadata.json

启动 Web UI 服务（端口 7241）：

burr --no-open --port 7241 --host 0.0.0.0

启动日志实测回显（关键行）：

INFO:     Uvicorn running on http://0.0.0.0:7241 (Press CTRL+C to quit)

可视化接口实测回显：

curl -s -o /dev/null -w "%{http_code}\n" http://127.0.0.1:7241/ready
200

curl -s http://127.0.0.1:7241/api/v0/projects
[{"name":"demo_counter","id":"demo_counter","last_written":"2026-10-13T15:48:22.019192Z","created":"2026-10-13T15:52:37.170526Z","num_apps":5,"uri":"https://github.com/apache/burr/tree/main/examples/hello-world-counter"}

验证结果：接口返回项目列表，其中 burr-quickstart 即本应用，可通过 Web UI 查看运行轨迹。

六、集成示例二：生成 pytest 测试用例

burr-test-case 可从已落盘的运行轨迹抽取输入/期望状态，生成 pytest 测试用例。实测命令：

burr-test-case create --project-name burr-quickstart --app-id 1b853a06-13d9-44e2-83a0-55897f602302 --sequence-id 3 --target-file-name /tmp/tc.json

实测回显：

Action was successful so loading initial and expected state into test fixture.

Writing data to file /tmp/tc.json

Add the following to your test file:

import pytest
from burr.core import state
from burr.testing import pytest_generate_tests  # noqa: F401

@pytest.mark.file_name("/tmp/tc.json")
def test_summarize(input_state, expected_state):
    """Function for testing the action"""
    input_state = state.State.deserialize(input_state)
    expected_state = state.State.deserialize(expected_state)
    _, output_state = summarize(input_state)  # exercise the action
    assert output_state == expected_state

补充说明：生成代码中保留了两处提示性注释（动作导入位置与断言方式选择），实际使用前按需替换；测试数据已写入 /tmp/tc.json，包含输入状态与期望状态。验证结果：用例生成成功，可直接接入 pytest 运行。

七、集成示例三：以 FastAPI 服务形态运行

Web UI 后端本身即 FastAPI 应用，可脱离 burr CLI 直接以 uvicorn 启动，便于与既有 Python 服务统一托管：

. /opt/burr-demo/.venv/bin/activate
python3 -m uvicorn burr.tracking.server.run:app --port 7241 --host 0.0.0.0

启动实测回显：

INFO:     Started server process [196]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:7241 (Press CTRL+C to quit)

验证结果：/ready 返回 200，/api/v0/version 返回 {"version":"0.42.0"}，与 burr CLI 启动的服务行为一致。

八、常见问题与参考链接