Building an Asynchronous Event-Driven Service Using Context Variables

In this lesson, we'll dive into building our agent, specifically focusing on its architecture or backend. Let's explore how our chat service uses asynchronous server-side events for real-time communication and how we can improve its implementation.

Our current setup uses a generator to manage asynchronous tasks, employing the 

 method to yield server-sent events every time our large-scale model generates a new chunk. This method is straightforward and works well in practice, but it's not without its flaws. Here are a few challenges we face:

async def process_messages(messages, chat):
    async for patch in chat.astream_log(messages):
        for op in patch.ops:
            if op["op"] == "add" and op["path"] == "/streamed_output/-":
                content = (
                    op["value"] if isinstance(op["value"], str) else op["value"].content
                )
                json_dict = {"type": "llm_chunk", "content": content}
                json_str = json.dumps(json_dict)
                yield f"data: {json_str}\n\n"

Imagine an implementation as simple and powerful as the 

 function, which logs events without needing to yield or return. Our goal is to achieve something similar for emitting events in our chat application.

 to handle asynchronous events efficiently. This method would allow us to pass event data straightforwardly, like an instance of a 

 class we've created. The beauty of this approach is its simplicity and the ability to use it anywhere in our code without blocking the server.

await emit_event(ChunkEvent(content=content))

---
title:  Building an Asynchronous Event-Driven Service Using Context Variables
isPublicLesson: true
privateVideoUrl: https://stream.mux.com/VgABSjKMiQYwIg2RBnzw00F9bASsb5j02yeTT7tI5XRQE.m3u8
---

# Building an Asynchronous Event-Driven Service Using Context Variables

In this lesson, we'll dive into building our agent, specifically focusing on its architecture or backend. Let's explore how our chat service uses asynchronous server-side events for real-time communication and how we can improve its implementation.

### Understanding the Current Implementation

Our current setup uses a generator to manage asynchronous tasks, employing the `astream_log` method to yield server-sent events every time our large-scale model generates a new chunk. This method is straightforward and works well in practice, but it's not without its flaws. Here are a few challenges we face:

1.  **Coupling of Network and Domain Logic**: As our chat logic becomes more intricate, intertwining it with network operations complicates maintenance and scalability.
2.  **Difficulty in Subdividing Code**: The monolithic nature of our generator makes it hard to break down into smaller, more manageable functions. For instance, creating subfunctions within our setup could lead to redundant yields in each child function call, bloating the code unnecessarily.

```python
async def process_messages(messages, chat):
    async for patch in chat.astream_log(messages):
        for op in patch.ops:
            if op["op"] == "add" and op["path"] == "/streamed_output/-":
                content = (
                    op["value"] if isinstance(op["value"], str) else op["value"].content
                )
                json_dict = {"type": "llm_chunk", "content": content}
                json_str = json.dumps(json_dict)
                yield f"data: {json_str}\n\n"
```

### Moving Towards an Ideal Implementation

Imagine an implementation as simple and powerful as the `print` function, which logs events without needing to yield or return. Our goal is to achieve something similar for emitting events in our chat application.

We envision a method called `emit_event` that uses `await` to handle asynchronous events efficiently. This method would allow us to pass event data straightforwardly, like an instance of a `chunk event` class we've created. The beauty of this approach is its simplicity and the ability to use it anywhere in our code without blocking the server.

```python
await emit_event(ChunkEvent(content=content))
```

##### Managing Shared State with `contextvar`

To implement `emit_event` effectively, we need a way to manage state across asynchronous operations without resorting to global or local variables, which either risk data leaks between requests or restrict flexibility. Enter `contextvar`, a native Python module perfect for this task. It allows us to share variables within a specific request's context securely.

-   **Set Up**: Initialize a queue in the context when a request starts.
-   **Usage**: Retrieve and use this queue from anywhere in the code to add or process events.

Our chat service will use two main methods:

-   **`emit_event`**: Adds an event to the queue.
-   **`close`**: Signals the closure of the event queue.

Louis Sanna

Welcome back. In this lesson, we will start building our agent focusing on the

or backend. Here you can see the code we built when we created our chat. It's a

we are leveraging the "astrim-lock method" or "long-chain" to yield some recent

chunk is being generated by our larger model. The simple notation is simple and

However, there are several limitations. The first is that we are coupling

logic. And as domain logic grows more complicated, it becomes really a pain.

this code is hard to speak into sub-function. Why is that? Imagine this code

hundreds or hundreds of thousands of lines, so you want to create a sub-method

say. Now in this function, anything may happen and you want to emit an event to

updated. The problem is that if you yield in the chat function, you will be

iterate over the generator and yield the second time. And so on. If you have a

five functions, at every level, you will be forced to yield again and so you'll

a code which is very valuable. Now let's think our dream implementation. Why

Well, there is a method which looks very close to what we would like. Which is

Let's take a look at print. It looks very simple, just print a little word. And

it's quite magical. Why is that? Because we did not return, we do not yield,

anywhere. And anytime you use print, it will add a new event to the logs. Now

great user experience and this is why this print method is so often used. So

like as an experience. And our second property of our implementation, we would

a second use behavior as we do not want to block our server anytime an event is

Now let's look at our dream implementation. It's a simple method called emit

We need to use a word as an iconus and then we simply pass it an event, your

chunk event we created. And that's all. You can create from anywhere in your

How can we make this implementation possible? How can we make this

We want to emit those events from anywhere. So we need to share state. It

state as we do not want to mix everyone between request. It cannot be a local

emit from anywhere in the call stack. How can we do that? We can do that with

which is the native Python module, which less does create viable with such

between a given request. Then once we create the context bar, let me show you

You simply need to set what you want in your context.

Here we are setting our emit event function and then from anywhere in this code

get here it is. And from get you get your emit event helper. Now what do we set

We will set a queue. A queue is a data structure implementing a first in first

And here we are leveraging the asynchronous module to use the asynchronous

create two methods, emit event, which simply add an event to the queue and

add a unique object to send your other closure over a queue. And that's all. So

logic, we simply need to call emit event and emit an event and it will be put

all. And as you can see, we've created some very simple class to add some

Now let's look at how we are going to build our semi-cent event from this queue

bit more complication, but it's completely separated from the domain code. Let

core logic. We simply loop over the queue and we say as long as the queue is

Anytime we find an event in the queue, if it's a chat event, we yield a service

And once the queue is empty and the task is done, we always run nothing to do

and we are finished our job. So that was a bit of setup. And as you'll see, you

evidence making the rest of record much simpler and cleaner. See you soon.

Learn

The newline Guide to Building Your First GraphQL Server with Node and TypeScript

Teach

Amelia Wattenberger

Author of Fullstack D3

Community

Masterclasses

Tutorials

Fullstack React with TypeScript