跳到主要内容

Burr — 图、转移与核心执行循环

这是 Burr 的「主算法」。本章讲三件事:图怎么建、怎么按转移边选下一个动作、核心循环 _step → iterate → run 每一步发生了什么(含读写校验和判停逻辑)。

1. Graph:动作 + 转移的容器​

它要解决的小问题​

给定「我刚跑完动作 X、当前状态是 S」,下一个该跑谁? 这是状态机的核心问询。Graph 就是为高效回答它而存在的数据结构。

思路/直觉:一张邻接表​

Graph 是个 dataclass,持有 actions 和 transitions(burr/core/graph.py:113)。建好后,__post_init__ 预计算两张表:

_action_map:     动作名 → Action 对象
_adjacency_map:  动作名 → [(目标动作名, 条件), (目标动作名, 条件), ...]   ← 按声明顺序

(burr/core/graph.py:122-126 __post_init__;:128 _create_adjacency_map)

核心方法:get_next_node​

「选下一步」就一个方法(burr/core/graph.py:150):

# 真实源码 burr/core/graph.py:150-160 Graph.get_next_node
def get_next_node(self, prior_step, state, entrypoint) -> Optional[Action]:
    if prior_step is None:
        return self._action_map[entrypoint]          # 还没跑过 → 入口
    possibilities = self._adjacency_map[prior_step]  # 从上一步出发的所有边
    for next_action, condition in possibilities:     # 按声明顺序逐条试
        if condition.run(state)[Condition.KEY]:       # 第一个为真的就赢
            return self._action_map[next_action]
    return None                                      # 没边可走 → 停机

三个关键性质,务必记牢:

  1. 顺序敏感:边按你 with_transitions 的声明顺序求值,第一个条件为真的边胜出,其余不再看。所以「特殊情况」的边要写在「default 兜底」边前面。
  2. None = 停机:没有任何边的条件为真(或这个动作没有出边),状态机就停下来。
  3. 入口靠 prior_step is None:第一步没有「上一步」,所以直接走入口动作。

GraphBuilder:怎么把它建起来​

GraphBuilder(burr/core/graph.py:276)收集动作和转移,build()(:366)时做校验后造出 Graph:

  • with_actions(*args, **kwargs):位置参数要求动作自带名(函数用函数名),关键字参数 name=action 显式命名(graph.py:286)。
  • with_transitions(*transitions):每条是 (from, to) 或 (from, to, condition);from 可以是 list(一对多)(graph.py:314)。
  • build():_validate_actions(查重名,graph.py:41)+ _validate_transitions(查 from/to 都存在、无重复 default,graph.py:57)。

注意 builder 与 Application 的关系:ApplicationBuilder.with_actions/with_transitions 其实是转发给一个内部 GraphBuilder(burr/core/application.py:2373/2390)。你也可以单独建 Graph 再 with_graph 复用——这就是「图与应用关注点分离」的设计(graph.py:276 docstring)。

2. 核心循环:_step 干的四件事​

它要解决的小问题​

「按一步」= 选动作 → 跑它 → 把结果并进状态 → 通知所有观察者。这四步必须有顺序、有校验、有错误兜底。

真实实现:_step​

Application._step(burr/core/application.py:949)是整个库的心脏。去掉细节后骨架是:

# 真实源码骨架 burr/core/application.py:949-1003 Application._step
with self.context:
    next_action = self.get_next_action()      # ① 选
    if next_action is None:
        return None                           #    没下一步 → 停
    action_inputs = self._process_inputs(inputs, next_action)  # 注入/校验输入
    if _run_hooks:
        self._adapter_set.call_all_lifecycle_hooks_sync("pre_run_step", ...)  # ④a 前钩子
    try:
        if next_action.single_step:
            result, new_state = _run_single_step_action(next_action, self._state, action_inputs)
        else:
            result = _run_function(next_action, self._state, action_inputs, ...)   # ② 算
            new_state = _run_reducer(next_action, self._state, result, ...)        # ③ 并
        new_state = self._update_internal_state_value(new_state, next_action)      # 记 __PRIOR_STEP
        self._set_state(new_state)
    except Exception as e:
        exc = e; raise
    finally:
        if _run_hooks:
            self._adapter_set.call_all_lifecycle_hooks_sync("post_run_step", ...)  # ④b 后钩子(含 exception)
    return next_action, result, new_state

逐步拆解​

① 选动作 — 调 get_next_action()(application.py:2034),它读内部状态里的 __PRIOR_STEP 传给 Graph.get_next_node。

②a 算(多步动作) — _run_function(application.py:161)。两个关键动作:

  • 状态裁剪:state.subset(*function.reads) ——只把声明读的键喂进去,别的看不到(application.py:177)。这是 reads 契约的运行时落地。
  • 异步守卫:同步上下文里碰到 async 动作直接报错,提示用 arun/aiterate(application.py:171-176)。

②b/③ 算+并(单步动作) — _run_single_step_action(application.py:301)一次调 run_and_update,然后做输出归一化和写键校验。

③ 并(多步动作的 reducer) — _run_reducer(application.py:243)。这里有两道写键校验:

# 真实源码节选 burr/core/application.py:255-262 _run_reducer
new_keys = keys_in_new_state - set(state.keys())
extra_keys = new_keys - set(reducer.writes)
if len(extra_keys) > 0:
    raise ValueError(f"Action {name} attempted to write to keys {extra_keys} that it did not declare...")
_validate_reducer_writes(reducer, new_state, name)  # 反过来:声明要写的键真的存在吗

第一道防「写了没声明的键」,第二道(application.py:233 _validate_reducer_writes)防「声明要写却没写出来」。

记账 — _update_internal_state_value(application.py:1010)把刚跑的动作名写进内部键 __PRIOR_STEP(常量 PRIOR_STEP,application.py:85),下一轮 get_next_action 靠它知道「从哪出发」。

④ 钩子 — 前后各一发 pre_run_step / post_run_step;后钩子在 finally 里,即使动作抛异常也会带着 exception= 触发(追踪能记录失败步,持久化能存「failed」状态)。详见第 4 章。

state_update:删键也要正确传播​

动作返回的新状态可能删了某些键(wipe)。_state_update(application.py:199)负责把「删除」正确合并回主状态:它先抹掉所有 __ 开头的内部键(动作不该碰这些),再算出被删的键,merge 后 wipe。注释坦白这是「次优实现」,理想方案是 delta 分层(又指向 issue #33)。

3. step → iterate → run:三层 API​

思路:一层套一层​

run(halt_before, halt_after)
  └─ 把 iterate() 这个生成器跑到尽头,取它的 return 值
       iterate(...)  ← while has_next_action: step(); yield; 判停
            └─ step()  ← 增 sequence_id,调 _step(_run_hooks=True)
                 └─ _step()  ← 上面那四步
  • step()(application.py:916):public,自增 sequence_id 后调 _step。每次 step,序号 +1(从 0 起,第一步看到的是 1,application.py:861 注释)。
  • iterate()(application.py:1259):生成器。循环 step() 并 yield (动作,结果,状态),每轮后查 _should_halt_iterate 决定 break。它用 return 返回最终三元组(Python 生成器的 return 值藏在 StopIteration.value 里)。
  • run()(application.py:1336):就是把 iterate() 跑干,捕获 StopIteration 取它的 .value(application.py:1356-1361)。

halt_before vs halt_after​

判停逻辑在 _should_halt_iterate(application.py:1213):

条件含义谁优先
halt_before=[X]在即将执行 X 之前停(X 不跑)halt_before 先判
halt_after=[Y]在执行完 Y 之后停其次
# 真实源码 burr/core/application.py:1213-1223 _should_halt_iterate
if self.has_next_action() and self.get_next_action().name in halt_before:
    return True                          # 下一步在黑名单 → 现在停
elif prior_action.name in halt_after:
    return True                          # 刚跑完的在停名单 → 停
return False

关键纪律(docstring application.py:1276 点明):每轮至少执行一次 step 才检查判停,且 halt_before 优先于 halt_after。

标签(tag):一次指一批动作​

halt 列表和转移里都能用 @tag:<name> 指代「一组带同 tag 的动作」。_parse_action_list(application.py:1156)把 @tag:x 展开成对应动作名;tag→动作映射在 Graph._create_action_tag_map(graph.py:137)预建。

4. 同步 vs 异步:几乎对称​

每个同步方法都有 a 前缀的异步孪生:step/astep、iterate/aiterate、run/arun。_astep(application.py:1089)有个巧妙处理:如果下一个动作是同步的,它直接委托给 self._step(避免乱开线程),只有真正 async 的动作才走 await(application.py:1111-1120)。validate_correct_async_use(application.py:2160)还会在你用错(同步 API 跑 async 应用)时拦下来。

5. 代码地图(导航索引)​

主题文件符号名
图与邻接表burr/core/graph.pyGraph、Graph._create_adjacency_map
选下一个动作burr/core/graph.pyGraph.get_next_node
图构建与校验burr/core/graph.pyGraphBuilder、_validate_actions、_validate_transitions
核心单步burr/core/application.pyApplication._step、Application.step
跑函数(算)burr/core/application.py_run_function、_arun_function
跑 reducer(并)+ 写键校验burr/core/application.py_run_reducer、_validate_reducer_writes
单步动作执行burr/core/application.py_run_single_step_action、_adjust_single_step_output
删键传播burr/core/application.py_state_update
循环与判停burr/core/application.pyiterate、run、_should_halt_iterate、_return_value_iterate
记账内部键burr/core/application.pyPRIOR_STEP、_update_internal_state_value
标签解析burr/core/application.py_parse_action_list
异步委托burr/core/application.py_astep、validate_correct_async_use