Streaming Support for Intermediate Nodes

[meeran03]

Streaming Support For Intermediate Nodes

First of all, thank you for developing such an amazing framework! I’m currently working on a complex system using Burr, where actions (nodes) and their transitions (edges) are dynamically generated. This setup was working flawlessly when the system was a simple chatbot, but I’m now transitioning to a Voice Bot using VAPI, and I’m running into some challenges.

Context:

The goal is to have OpenAI-compatible event streams, as low latency is a priority for me. The current flow works like this:

• A relevant node completes its task, generates a response, and hands control back to the process_user_query node to collect input from the user.
• At that point, I send the API call response back to the user.

Here's a simplified version of the initial code:

while True:
    _action, result, state = await app.arun(
        halt_before=["process_user_query"], halt_after=[], inputs=inputs
    )
    if _action and _action.name == "process_user_query":
        break

return result, state

Transition to Streaming:

When I implemented streaming, I modified the function as follows:

async def get_response_stream(self, user: User, chat_id, query):
    """
    Get response for a query
    """
    app = await self.build_or_get_app(user, chat_id, sequence_number=None, stream=True)
    inputs = {"query": query}
    streaming_container = None

    while True:
        action, streaming_container = await app.astream_result(
            halt_before=["process_user_query"], halt_after=[], inputs=inputs
        )

        async for result in streaming_container:
            yield result, None

        if action and action.name == "process_user_query":
            break

    result, state = await streaming_container.get()
    yield result, state

The Problem:

Although I’ve switched my actions to streaming_actions, no results are being streamed back to the container in real-time. It seems streaming only begins once control is returned to the process_user_query node, which defeats the purpose of real-time feedback and increases latency.

---

Question:

Is there a way to allow results to be streamed continuously while still processing actions before control is passed back to the process_user_query node? Any guidance or suggestions on how to improve this would be greatly appreciated!

[faisalrehmankhattak]

I'm facing same issue and its stopping me to use burr for call applications

[elijahbenizzy]

Hey @faisalrehmankhattak and @meeran03 -- thanks for the issue! I think there are a few issues here, but this is a supported use-case and we'll make sure we get to a solution.

Two cases:

(1) you want to use halt_before + astream_result(...)

This assumes that the only node you want to stream back is process_user_query -- see (2) otherwise -- @meeran03 I think this is describing your case, but I'm not 100% sure.

This is a bit of unclear documentation, but I think there is an easy fix. halt_before doesn't actually stream (to transition between actions, we need full state to complete so we know where to transition to) -- if process_user_query is the node that streams, it'll stop before that, which means that the next time you run it it'll execute that (start + run through the stream). I wonder if it would be as simple as swtiching that to halt_after.

The code would look something like this:

async def get_response_stream(self, user: User, chat_id, query):
    """
    Get response for a query
    """
    app = await self.build_or_get_app(user, chat_id, sequence_number=None, stream=True)
    inputs = {"query": query}
    streaming_container = None

    while True:
        action, streaming_container = await app.astream_result(
            halt_after=["process_user_query"], inputs=inputs
        )

        async for result in streaming_container:
            yield result, None
        # no need to do anything here, the framework takes care of it when it's done

(2) you want to stream intermediate results.

E.G. there are multiple nodes in a row that you want to stream back. @faisalrehmankhattak my guess is that this is what you want (although just a guess, would love to see your use-case more).

While there is not a single API that does this (and this is a reasonable feature, we'd need a app.aiterate_stream_result(...), which would be simple enough to build, it's possible to get done with a while loop.

while True:
    action, streaming_container = await app.astream_result(
        halt_after=[action.name for action in app.graph.actions if action.streaming], # adjust according to what you want to actually halt after...
        inputs={...}
   )
   # do something with streaming_container results
   if action.name in some_set_of_terminal_nodes:
       break

The trick here is that we just halt after every streaming one, in a while loop. Then we stream it back, and continue after we're done streaming. Does this make sense/help?

[skrawcz]

To clarify: we currently don't support halt_before and streaming because to transition between actions, we need state. State is only available after an entire results has been returned. You'd need to use halt_after instead to get the streamed container back as Elijah mentions.

Though if there is no conditional edge transition, and it's always going to go to that action afterwards, it's conceivable we could pass that through somehow and enable halt_before to work...

[meeran03]

Using Streaming Nodes with Burr for Enhanced Sequential Flow

Thank you @elijahbenizzy for your input! The second use case definitely resonates with me, and I will give it a try.

Use Case Context:

In our system, almost all nodes are streaming nodes, except for process_user_query. Here's a brief overview of our use case:

Previously, we developed LLM applications that followed a sequential flow, but we quickly realized that LLMs struggle with executing step-by-step instructions in sequence. This limitation made it difficult to maintain the desired control over transitions between steps.

Our Solution:

To tackle this, we employed a state machine approach, where each step is modeled as a node. The key idea is to keep the flow locked within a node until a specific condition—reflected by the state—is met. Only then does the system transition to the next node, ensuring a reliable sequential flow.

We’ve essentially built something similar to BlandAI using Burr, but with even more power and flexibility. One of the highlights is our use of LLM function calling to manage conditional transitions, which can handle both natural language and Burr expressions.

Adding an edge (i.e., a transition) in the state machine has become incredibly simple with this setup. Here’s a quick example:

agent.add_edge(
    "order_inquiry_node",
    "current_node == 'order_inquiry_node'",
    "expression",
    """Use this tool when the user inquires about their previous orders or wants to repeat a previous order."""
)

I will try your proposed solution & I’ll report back with any insights or updates!

[meeran03]

Success with Streaming Solution and a Potential Burr Improvement

I finally managed to get the streaming functionality working using the second proposed solution. However, I ran into an issue where the machine wasn’t halting execution as expected. This was because I had implicit (default) transitions to the process_user_query node after the execution of every node.

Current Solution:

For now, I’ve resolved the issue by checking whether the streaming_container contains a complete result, and if so, breaking the loop. Here’s the code I’m currently using:

async def get_response_stream(self, user: User, chat_id, query):
    """
    Get response for a query
    """
    app = await self.build_or_get_app(user, chat_id, sequence_number=None, stream=True)
    inputs = {"query": query}
    streaming_container = None
    
    while True:
        action, streaming_container = await app.astream_result(
            halt_after=[
                action.name for action in app.graph.actions if action.streaming
            ],
            halt_before=["process_user_query"],
            inputs=inputs,
        )
        
        async for result in streaming_container:
            yield result, None

        result, state = await streaming_container.get()
        
        if action and action.name == "process_user_query" or result.get('content') is not None:
            break

    result, state = await streaming_container.get()
    yield result, state

Possible Improvement in Burr:

While this workaround solves the issue for now, I feel that Burr could potentially handle this more elegantly. Perhaps Burr could introduce a more explicit mechanism to stop execution when a complete result is received, eliminating the need for manual checks like this.

I’m happy with the current progress, but I’d love to hear any thoughts or suggestions on how Burr might handle this more natively!

[elijahbenizzy]

First, glad this worked! I think this is a good solution, given the current constraints. That said, this is a reasonable way of approaching it, and I think something that wouldn't be too hard! #360.

Would implementing this make your life easier? We have gotten a few qs like this in the past.

[meeran03]

I think it falls in the category of "nice to have(s)". So, for now, it's fine with me. I am ultimately going to make my product multi-vendor where users can themselves design their nodes through a GUI, so maybe it may cause issues there as then a specific stopping condition response.get('content') is not None may not work.

For now, it works fine. Thanks for the prompt support! Really appreciate it.

[elijahbenizzy]

I think it falls in the category of "nice to have(s)". So, for now, it's fine with me. I am ultimately going to make my product multi-vendor where users can themselves design their nodes through a GUI, so maybe it may cause issues there as then a specific stopping condition response.get('content') is not None may not work.

For now, it works fine. Thanks for the prompt support! Really appreciate it.

Great, makes sense. Going to leave that open to get user input, so comment if it's something you'll need/will materially make your life easier.

[skrawcz]

FYI @meeran03 I sent you an email about Burr cloud -- let me know if that went to spam.