async_btree 3.0.0

Async behavior tree

Async Behaviour Tree for Python

Versions following Semantic Versioning

See documentation.

Requires Python 3.11+.

• For Python >= 3.11, (asyncio/curio) use 2.x releases.
• For Python 3.9/3.10, (asyncio/curio) use 1.x releases.

Overview

What's a behavior tree ?

Unlike a Finite State Machine, a Behaviour Tree is a tree of hierarchical nodes that controls the flow of decision and the execution of "tasks" or, as we will call them further, "Actions". -- behaviortree

If you're new (or not) to behavior tree, you could spend some time on this few links:

• Behavior trees for AI: How they work by Chris Simpson
• Introduction to BTs

Few implementation libraries:

• task_behavior_engine A behavior tree based task engine written in Python
• pi_trees a Python/ROS library for implementing Behavior Trees
• pr_behavior_tree A simple python behavior tree library based on coroutines
• btsk Behavior Tree Starter Kit
• behave A behavior tree implementation in Python

Why another library so ?

SIMPLICITY

When you study behavior tree implementation, reactive node, dynamic change, runtime execution, etc ... At a moment you're build more or less something that mimic an evaluator 'eval/apply' or a compilator, with a complex hierarchical set of class.

All complexity came with internal state management, using tree of blackboard to avoid global variable, multithreading issue, maybe few callback etc ...

This break the simplicity and beauty of your initial design.

What I find useful with behavior tree:

• clarity of expression
• node tree representation
• possibility to reuse behavior
• add external measure to dynamicaly change a behavior, a first step on observable pattern...

As I've used OOP for years (very long time), I will try to avoid class tree and prefer using the power of functional programming to obtain what I want: add metadata on a semantic construction, deal with closure, use function in parameters or in return value... And a last reason, more personal, it that i would explore python expressivity.

SO HOW ?

This library uses coroutines and functional composition instead. No class trees. No configuration files (no XML, no JSON, no YAML). Business logic is plain Python functions. Composition is plain Python.

By this way:

• we reuse simple language idiom to manage state, parameter, etc
• no design constraint on action implementation
• most of language build block could be reused

No need to introduce an extra level of abstraction to declare a composition of functions. I think it's true for most of main use case (except using an editor to wrote behaviour tree for example). So "If you wrote your function with python, wrote composition in python"... (remember that you did not need XML to do SQL, just write good sql...)

See Concepts for a deeper explanation of the design principles.

The rest is just implementation details..

You could build expression like this:

import async_btree as bt

async def a_func():
    return "a"

async def b_decorator(child_value, other=""):
    return f"b{child_value}{other}"

assert bt.run(bt.decorate(a_func, b_decorator)) == "ba"

This expression apply b_decorator on function a_func. Note that decorate(a_func, b_decorator) is not an async function, only action, or condition are async function.

Want an abstract tree of our behaviour tree ?

Functions from async-btree build an abstract tree for you. If you lookup in code, you should see an annotation "node_metadata" on internal implementation. This decorator add basic information like function name, parameters, and children relation ship.

This abstract tree can be retrieved and stringified with analyze and stringify_analyze.

For example:

# your behaviour tree, or a sub tree:
my_func = alias(child=repeat_while(child=action(hello), condition=success_until_zero), name="btree_1")

# retrieve meta information and build a Node tree
abstract_tree_tree_1 = analyze(my_func) 

# output the tree:
print(stringify_analyze(abstract_tree_tree_1))

This should print:

 --> btree_1:
     --(child)--> repeat_while:
         --(condition)--> success_until_zero:
         --(child)--> action:
                      target: hello

Note about action and condition method:

• you could use sync or async function
• you could specify a return value with SUCCESS or FAILURE
• function with no return value will be evaluated as FAILURE until you decorate them with a always_successor always_failure

Key design decisions

• Truthy/falsy as node status — SUCCESS / FAILURE are True / False. Exceptions are wrapped in ControlFlowException to give them falsy meaning without losing the original cause.
• ContextVar as blackboard — no custom blackboard class needed. Use Python's built-in contextvars.
• Node metadata — @node_metadata decorates inner functions with name, parameters, and child relationships. This builds the abstract tree used by analyze().

Core primitives

Leaves

Primitive	Role
action	Wrap sync or async function as BT node; exceptions become ControlFlowException
condition	Wrap sync or async predicate; result coerced to SUCCESS/FAILURE

Control flow

Primitive	Role
sequence	Run children in order; stop early once enough succeed or too many fail (success_threshold)
fallback / selector	OR — run children in order, stop on first success
decision	If/else — evaluate success_tree or failure_tree based on condition
condition_guard	Run child only if condition is truthy; return SUCCESS otherwise
repeat_while	Loop child while condition is truthy
repeat_until	Loop child until condition becomes truthy
do_while	Run child at least once, then repeat while condition is truthy
repeat_n	Run child exactly N times
random_selector	Fallback with children shuffled on every tick
switch	Route to a child based on return value of condition
parallele	Run children concurrently; succeed if enough succeed (success_threshold)
parallel_race	Run children concurrently; first to finish wins, others cancelled

Decorators

Primitive	Role
decorate	Apply a decorator function to child output
alias	Name a subtree
ignore_exception	Turn exceptions into falsy ControlFlowException
always_success / always_failure	Force return semantics
inverter	Flip SUCCESS ↔ FAILURE
is_success / is_failure	Assert child result polarity
retry	Retry child up to N times on failure
retry_until_success / retry_until_failed	Retry until result flips
timeout_after	Return FAILURE if child exceeds deadline
cooldown	Skip child if called again before delay has elapsed
delay	Wait N seconds before running child

Async backend

3.0.0 uses anyio as the sole backend. Three runtimes supported:

Backend	Value
asyncio (default)	"asyncio"
trio	"trio"
asyncio + uvloop	"asyncio+uvloop"

Installation

pip install async-btree
# or
uv add async-btree

Optional extras for non-asyncio backends:

uv add trio           # trio backend
uv add uvloop         # asyncio+uvloop backend

Migrating from 2.x? See the migration guides in the documentation.

Usage

See API Reference documentation.

One-shot run

import async_btree as bt

result = bt.run(my_tree)                          # asyncio (default)
result = bt.run(my_tree, backend="trio")
result = bt.run(my_tree, backend="asyncio+uvloop")

Multiple runs in the same context

with bt.BTreeRunner(backend="asyncio") as runner:
    result1 = runner.run(tree_tick)
    result2 = runner.run(tree_tick)

Each runner.run() starts from the context snapshot captured at __enter__ — ContextVar mutations inside a tick do not carry over to the next tick.

Building trees

import async_btree as bt

b_tree = bt.sequence(children=[
    bt.always_success(child=bt.action(target=say_hello, name="John")),
    bt.action(target=check_battery),
    bt.always_success(child=bt.action(target=gripper.open)),
    bt.always_success(child=bt.action(target=approach_object, name="house")),
    bt.always_success(child=bt.action(target=gripper.close)),
])

bt.run(b_tree)

Concurrent execution

parallel_tree = bt.parallele(
    children=[sensor_a, sensor_b, sensor_c],
    success_threshold=2,   # succeed if at least 2 children succeed
)
result = bt.run(parallel_tree)

Exception handling

@bt.ignore_exception
async def unreliable_sensor() -> bool:
    raise IOError("disconnected")

# or apply dynamically at tree construction time
safe = bt.ignore_exception(unreliable_sensor)

Tree introspection

my_func = bt.alias(child=bt.repeat_while(child=bt.action(hello), condition=success_until_zero), name="btree_1")

abstract_tree = bt.analyze(my_func)
print(bt.stringify_analyze(abstract_tree))

 --> btree_1:
     --(child)--> repeat_while:
         --(condition)--> success_until_zero:
         --(child)--> action:
                      target: hello

Examples

• tutorial_1.py — basic actions, decorators, sequences, backend selection
• tutorial_2_decisions.py — decision trees and selectors with ContextVar
• tutorial_3_context.py — ContextVar isolation and propagation
• tutorial_4_exceptions.py — exception handling, ControlFlowException, parallele
• tutorial_5_switch.py — routing with switch, default branch, ContextVar-driven dispatch

See full API Reference and Tutorial.