Actions¶

class burr.core.action.Action¶

Bases: Function, Reducer, ABC

__init__()¶: Represents an action in a state machine. This is the base class from which actions extend. Note that this class needs to have a name set after the fact.

get_source() → str¶: Returns the source code of the action. This will default to the source code of the class in which the action is implemented, but can be overwritten.” Override if you want debugging/tracking to display a different source

property inputs: list[str] | tuple[list[str], list[str]]¶

Represents inputs that are used for this to run. These correspond to the **run_kwargs in run above.

Note that this has two possible return values: 1. A list of strings – these are the keys that are required to run the function 2. A tuple of two lists of strings – the first list is the required keys, the second is the optional keys

Returns:: Either a list of strings (required inputs) or a tuple of two lists of strings (required and optional inputs)

is_async() → bool¶

Convenience method to check if the function is async or not. This can be used by the application to run it.

Returns:: True if the function is async, False otherwise

property name: str¶: Gives the name of this action. This should be unique across your application.

property optional_and_required_inputs: tuple[set[str], set[str]]¶

Returns a tuple of two lists of strings – the first list is the required keys, the second is the optional keys. This is internal and not meant to override.

Returns:: Tuple of required keys and optional keys

abstract property reads: list[str]¶

Returns the keys from the state that this function reads

Returns:: A list of keys

abstract run(state: State, **run_kwargs) → dict¶

Runs the function on the given state and returns the result. The result is just a key/value dictionary.

Parameters:

state – State to run the function on
run_kwargs – Additional arguments to the function passed at runtime.

Returns:

Result of the function

validate_inputs(inputs: Dict[str, Any] | None) → None¶

Validates the inputs to the function. This is a convenience method to allow for validation of inputs before running the function.

Parameters:: inputs – Inputs to validate
Raises:: ValueError – If the inputs are invalid

with_name(name: str) → Self¶

Returns a copy of the given action with the given name. Why do we need this? We instantiate actions without names, and then set them later. This is a way to make the API cleaner/consolidate it, and the ApplicationBuilder will end up handling it for you, in the with_actions(…) method, which is the only way to use actions.

Note they can also take in names in the constructor for testing, but otherwise this is not something users will ever have to think about.

Parameters:: name – Name to set
Returns:: A new action with the given name

abstract property writes: list[str]¶

Returns the keys from the state that this reducer writes.

Returns:: A list of keys

class burr.core.action.Result(*fields: str)¶

__init__(*fields: str)¶

Represents a result action. This is purely a convenience class to pull data from state and give it out to the result. It does nothing to the state itself.

Parameters:: fields – Fields to pull from the state and put into results

property reads: list[str]¶

Returns the keys from the state that this function reads

Returns:: A list of keys

run(state: State) → dict¶

Runs the function on the given state and returns the result. The result is just a key/value dictionary.

Parameters:

state – State to run the function on
run_kwargs – Additional arguments to the function passed at runtime.

Returns:

Result of the function

property writes: list[str]¶

Returns the keys from the state that this reducer writes.

Returns:: A list of keys

class burr.core.action.Input(*fields: str)¶

__init__(*fields: str)¶

Represents an input action – this reads something from an input then writes that directly to state. This is a convenience class for when you don’t need to process the input and just want to put it in state for later use.

Parameters:: fields – Fields to pull from the inputs and put into state

property inputs: list[str]¶

Represents inputs that are used for this to run. These correspond to the **run_kwargs in run above.

Returns:: Either a list of strings (required inputs) or a tuple of two lists of strings (required and optional inputs)

property reads: list[str]¶

Returns the keys from the state that this function reads

Returns:: A list of keys

run(state: State, **run_kwargs) → dict¶

Runs the function on the given state and returns the result. The result is just a key/value dictionary.

Parameters:

state – State to run the function on
run_kwargs – Additional arguments to the function passed at runtime.

Returns:

Result of the function

property writes: list[str]¶

Returns the keys from the state that this reducer writes.

Returns:: A list of keys

@burr.core.action.action( reads: List[str], writes: List[str], ) → Callable[[Callable], FunctionRepresentingAction]¶

Decorator to create a function-based action. This is user-facing. Note that, in the future, with typed state, we may not need this for all cases.

If parameters are not bound, they will be interpreted as inputs and must be passed in at runtime. If they have default values, they will be recorded as optional inputs. These can (optionally) be provided at runtime.

Parameters:

reads – Items to read from the state
writes – Items to write to the state

Returns:

The decorator to assign the function as an action

burr.core.action.bind(

self: FunctionRepresentingAction,

**kwargs: Any,

) → FunctionRepresentingAction¶

Binds an action to the given parameters. This is functionally equivalent to functools.partial, but is more explicit and is meant to be used in the API. This only works with the @action functional API and not with the class-based API.

@action(["x"], ["y"])
def my_action(state: State, z: int) -> tuple[dict, State]:
    return {"y": state.get("x") + z}, state

my_action.bind(z=2)

Parameters:

self – The decorated function
kwargs – The keyword arguments to bind

Returns:

The decorated function with the given parameters bound

class burr.core.action.Function¶

Interface to represent the ‘computing’ part of an action

property inputs: list[str] | tuple[list[str], list[str]]¶

Represents inputs that are used for this to run. These correspond to the **run_kwargs in run above.

Returns:: Either a list of strings (required inputs) or a tuple of two lists of strings (required and optional inputs)

is_async() → bool¶

Convenience method to check if the function is async or not. This can be used by the application to run it.

Returns:: True if the function is async, False otherwise

property optional_and_required_inputs: tuple[set[str], set[str]]¶

Returns a tuple of two lists of strings – the first list is the required keys, the second is the optional keys. This is internal and not meant to override.

Returns:: Tuple of required keys and optional keys

abstract property reads: list[str]¶

Returns the keys from the state that this function reads

Returns:: A list of keys

abstract run(state: State, **run_kwargs) → dict¶

Runs the function on the given state and returns the result. The result is just a key/value dictionary.

Parameters:

state – State to run the function on
run_kwargs – Additional arguments to the function passed at runtime.

Returns:

Result of the function

validate_inputs(inputs: Dict[str, Any] | None) → None¶

Validates the inputs to the function. This is a convenience method to allow for validation of inputs before running the function.

Parameters:: inputs – Inputs to validate
Raises:: ValueError – If the inputs are invalid

class burr.core.action.Reducer¶

Interface to represent the ‘updating’ part of an action

abstract property writes: list[str]¶

Returns the keys from the state that this reducer writes.

Returns:: A list of keys

class burr.core.action.StreamingResultContainer( streaming_result_generator: Generator[Tuple[dict, State | None], None, None], initial_state: State, process_result: Callable[[dict, State], tuple[dict, State]], callback: Callable[[dict | None, State, Exception | None], None], )¶

Container for a streaming result. This allows you to:

Iterate over the result as it comes in
Get the final result/state at the end

If you’re familiar with generators/iterators in python, this is effectively an iterator that caches the final result after calling it. This is meant to be used exclusively with the streaming action calls in Application. Note that you will never instantiate this class directly, but you will use it in the API when it is returned by stream_result. For reference, here’s how you would use it:

action_we_just_ran, streaming_result_container = application.stream_result(...)
print(f"getting streaming results for action={action_we_just_ran.name}")

for result_component in streaming_result_container:
    print(result_component['response']) # this assumes you have a response key in your result

final_result, final_state = streaming_result_container.get()

static pass_through( results: dict, final_state: State, ) → StreamingResultContainer¶: Instantiates a streaming result container that just passes through the given results This is to be used internally – it allows us to wrap non-streaming action results in a streaming result container.

class burr.core.action.AsyncStreamingResultContainer( streaming_result_generator: AsyncGenerator[Tuple[dict, State | None], None], initial_state: State, process_result: Callable[[dict, State], tuple[dict, State]], callback: Callable[[dict | None, State, Exception | None], Coroutine[None, None, None]], )¶

Container for an async streaming result. This allows you to: 1. Iterate over the result as it comes in 2. Await the final result/state at the end

If you’re familiar with generators/iterators in python, this is effectively an iterator that caches the final result after calling it. This is meant to be used exclusively with the streaming action calls in Application. Note that you will never instantiate this class directly, but you will use it in the API when it is returned by astream_result. For reference, here’s how you would use it:

action_we_just_ran, streaming_result_container = await application.stream_result(...)
print(f"getting streaming results for action={action_we_just_ran.name}")

async for result_component in streaming_result_container:
    print(result_component['response']) # this assumes you have a response key in your result

final_result, final_state = await streaming_result_container.get()

static pass_through( results: dict, final_state: State, ) → AsyncStreamingResultContainer¶: Creates a streaming result container that just passes through the given results. This is not a public facing API.

class burr.core.action.streaming_action(reads: List[str], writes: List[str])¶

Decorator to create a streaming function-based action. This is user-facing.

If parameters are not bound, they will be interpreted as inputs and must be passed in at runtime.

See the following example for how to use this decorator – this reads prompt from the state and writes response back out, yielding all intermediate chunks.

Note that this must return a value. If it does not, we will not know how to update the state, and we will error out.

@streaming_action(reads=["prompt"], writes=['response'])
def streaming_response(state: State) -> Generator[dict, None, tuple[dict, State]]:
    response = client.chat.completions.create(
        model='gpt-3.5-turbo',
        messages=[{
            'role': 'user',
            'content': state["prompt"]
            }],
        temperature=0,
    )
    buffer = []
    for chunk in response:
        delta = chunk.choices[0].delta.content
        buffer.append(delta)
        # yield partial results
        yield {'response': delta}
    full_response = ''.join(buffer)
    # return the final result
    return {'response': full_response}, state.update(response=full_response)

class burr.core.action.StreamingAction¶

Base class for Streaming action. These are “multi-step”, meaning that they run in multiple passes (run -> update)

run(state: State, **run_kwargs) → dict¶: Runs the streaming action through to completion.

abstract stream_run(

state: State,

**run_kwargs,

) → Generator[dict, None, None]¶

Streaming action stream_run is different than standard action run. It: 1. streams in an intermediate result (the dict output) 2. yields the final result at the end

Note that the user, in this case, is responsible for joining the result.

For instance, you could have:

def stream_run(state: State) -> Generator[dict, None, dict]:
    buffer = [] # you might want to be more efficient than simple strcat
    for token in query(state['prompt']):
        yield {'response' : token}
        buffer.append(token)
    yield {'response' : "".join(buffer)}

This would utilize a simple string buffer (implemented by a list) to store the results and then join them at the end. We return the final result.

Parameters:

state – State to run the action on
run_kwargs – parameters passed to the run function – these are specified by inputs

Returns:

A generator that streams in a result and returns the final result