Apache Burr

The following examples demonstrate various features and use cases of the Burr library:

• simple-chatbot-intro: A basic chatbot implementation; recommended starting point for understanding Burr.
• conversational-rag: Demonstrates conversational Retrieval-Augmented Generation (RAG) using state and prior knowledge to augment LLM calls.
• hello-world-counter: A minimal state machine example used in documentation.
• llm-adventure-game: A text-based adventure game demonstrating progression through hidden states and component reuse.
• ml-training: A simple ML training pipeline demonstrating how to track model training.
• multi-agent-collaboration: Demonstrates multi-agent collaboration patterns.
• multi-modal-chatbot: Shows how to use a model to conditionally delegate tasks to other models.
• streaming-overview: Demonstrates using the streaming API for faster user responses.
• integrations/bedrock: Minimal graphs using Amazon Bedrock via BedrockAction and BedrockStreamingAction.
• tracing-and-spans: Demonstrates how to use Burr's tracing functionality for increased visibility.
• web-server: Shows how to integrate Burr into a web server for user interaction.

When scaling a Burr-powered web service horizontally, consider these two layers:

Database Layer

Scalability depends on your chosen database and schema. Burr facilitates efficient querying by partitioning data using application_id and a partition_key (such as a user_id). Indexing your state table on these keys allows for efficient retrieval of specific application states.

Compute Layer

To scale the server instances, you must handle state synchronization and locking to prevent race conditions if multiple servers attempt to run the same application simultaneously. Strategies include:

• Database Locking: Use a locking mechanism in the database to ensure only one server processes a specific application at a time.
• Sticky Sessions/Sharding: Route a specific user to the same server instance to minimize synchronization issues.
• Custom Persistence Logic: Implement custom logic at the persistence layer to handle forking or resolution of state.

Streaming actions allow you to yield intermediate results (like LLM tokens) to the user as they are produced, reducing time-to-first-token.

Function-based Streaming Actions

Use the @streaming_action decorator. The function must be a generator that yields a tuple: (result_dict, state_update, None, None).

• Intermediate results: Yield (result_dict, None) to provide data without updating the state.
• Final result: The last yield must include the final result and the state update (e.g., yield final_result_dict, state.append(...)).

Class-based Streaming Actions

Inherit from StreamingAction. This approach separates the execution logic from the state update logic:

• stream_run(...): A generator that yields only the result_dict. The final yield in this method is passed to the update method.
• update(result, state): Receives the final result from stream_run and returns the updated State.
• reads and writes: Properties defining the state keys used.

Async Streaming Actions

Both function-based and class-based patterns support async implementations using AsyncGenerator and async for loops.

To execute a full Cartesian product of all provided actions and all provided states (e.g., running every model in a list against every prompt in a list), implement the MapActionsAndStates class. This is the base class for both MapStates and MapActions.

Implementation requirements:

• .actions(): Generator yielding the actions to run.
• .states(): Generator yielding the state variations.
• .reduce(): Merges the results of all combinations into the final state.

from burr.core import action, state
from burr.core.parallelism import MapActionsAndStates, RunnableGraph
from typing import Callable, Generator, List

@action(reads=["prompt", "model"], writes=["llm_output"])
def query_llm(state: State, model: str) -> State:
    return state.update(llm_output=_query_my_llm(prompt=state["prompt"], model=model))

class TestModelsOverPrompts(MapActionsAndStates):

    def actions(self, state: State, context: ApplicationContext, inputs: Dict[str, Any]) -> Generator[Action | Callable | RunnableGraph, None, None]:
        for action in [
            query_llm.bind(model="gpt-4").with_name("gpt_4_answer"),
            query_llm.bind(model="o1").with_name("o1_answer"),
            query_llm.bind(model="claude").with_name("claude_answer"),
        ]:
            yield action

    def states(self, state: State, context: ApplicationContext, inputs: Dict[str, Any]) -> Generator[State, None, None]:
        for prompt in [
            "What is the meaning of life?",
            "What is the airspeed velocity of an unladen swallow?",
            "What is the best way to cook a steak?",
        ]:
            yield state.update(prompt=prompt)

    def reduce(self, state: State, states: Generator[State, None, None]) -> State:
        all_llm_outputs = []
        for sub_state in states:
            all_llm_outputs.append(
                {
                    "output" : sub_state["llm_output"],
                    "model" : sub_state["model"],
                    "prompt" : sub_state["prompt"],
                }
            )
        return state.update(all_llm_outputs=all_llm_outputs)

    @property
    def reads(self) -> List[str]:
        return ["prompts"]

    @property
    def writes(self) -> List[str]:
        return ["all_llm_outputs"]