Serializable context when using llama_deploy

[hristogg]

Hello,
Thanks for the support in advance, I am trying to use llama deploy to deploy a workflow, the workflow runs fine however I reach this error:
ERROR:llama_deploy.services.workflow - Encountered error in task 0546b1fe-2041-4040-8bba-36bebc083379! keys must be str, int, float, bool or None, not ModelMetaclass

Which seems to stem from the attempt to serialize the context of the handler from set_workflow_state.
At first I though to serialize everything in json before passing it ot the context and then deserialize once collecting it from the context hoping that this would solve my issue.

However now, my events is a string and I still have the same issue...
Here is how the event is defined:

class OutputDoc(Event):
    data: str

class OutputQna(Event):
    data: str

and the serialization fails on:
context_dict['events_buffer']
which shows has:
{<class '__main__.OutputDoc'>: [], <class '__main__.OutputQna'>: []}

As trying to understand it better, I tried constructing a dummy code:

from llama_index.core.workflow import Event
import json

class OutputQna(Event):
    data: str

a = OutputQna(data='test')
json.dumps(a)

Which I had the impression is what is supposed to happen, however this fails as well
although a.dict is present: {'data': 'dsadsadsa'}

Can you support me in understanding what I am doing wrong? It seems pretty straightforward to me that I'd like to pass List[NodesWithScore] around but even when serializing them I seem to be suffering :)

[dosubot[bot]]

The error you're encountering is due to the fact that the keys in the events_buffer dictionary are class types, which are not directly serializable using JSON. JSON serialization requires keys to be of type str, int, float, bool, or None, but not class types like ModelMetaclass.

To resolve this issue, you can modify the way the events_buffer is serialized by converting the class types to strings. Here's a possible approach:

1. Serialization: Convert the class type keys to strings when serializing the events_buffer:

   serialized_events_buffer = {
       str(k): [serializer.serialize(ev) for ev in v]
       for k, v in context_dict['events_buffer'].items()
   }

2. Deserialization: Convert the string keys back to class types when deserializing:

   deserialized_events_buffer = {
       globals()[k]: [serializer.deserialize(ev) for ev in v]
       for k, v in serialized_data['events_buffer'].items()
   }

This approach ensures that the keys are in a JSON-compatible format during serialization and are correctly converted back to class types during deserialization.

Additionally, for the OutputDoc and OutputQna classes, ensure that they are serializable by implementing a custom serialization method if necessary. The Event class in llama_deploy uses a custom model dump method to include additional attributes in the serialization process, which you might need to adapt for your custom event classes ^[1].

By following these steps, you should be able to serialize and deserialize your context without encountering the error.

To continue talking to Dosu, mention @dosu.

[hristogg]

@dosu I think there should be out of the box solution for passing nodes between events, without modifying the event handler so I am guessing that I am doing something wrong. How can I construct the events so that they are serializable?

[dosubot[bot]]

To construct events in llama_deploy that are serializable without modifying the event handler, you can use the JsonSerializer class from the llama_index.core.workflow.context_serializers module. This class allows you to serialize and deserialize events easily. Here's an example:

from llama_index.core.workflow.events import Event
from llama_index.core.workflow.context_serializers import JsonSerializer

class MyEvent(Event):
    param: str

# Create an instance of the event
event_instance = MyEvent(param="example")

# Use JsonSerializer to serialize the event
serializer = JsonSerializer()
serialized_event = serializer.serialize(event_instance)

# Deserialize the event back to an object
deserialized_event = serializer.deserialize(serialized_event)

# Verify that the deserialized event matches the original
assert event_instance.param == deserialized_event.param
assert event_instance._data == deserialized_event._data

This approach ensures that your events are serializable without needing to modify the event handler itself ^[1].

To continue talking to Dosu, mention @dosu.

[hristogg]

@dosu thank and how to incorporate these events into the workflow? I trying to combined two events to a single one through the context in the following way, here is a sample code:

class OutputQna(Event):
    data: str

class OutputDoc(Event):
    data: str
	
class RAGWorkflow(Workflow):
   

   
    @step
    async def qna(self, ctx: Context, ev: QnaQuery) -> OutputQna:
      #retrieval here
	  #make sure that we pass json to context
      json_data = json.dumps([node_with_score.__dict__ for node_with_score in qna_output], default=lambda o: o.__dict__, ensure_ascii=False)
   
      ctx.send_event(OutputQna(data=json_data))
      return None
    
    @step
    async def retriev(self, ctx: Context, ev: DocQuery) -> OutputDoc:
        #retrieval here
		#make sure that we pass json to context
		json_data = json.dumps([node_with_score.__dict__ for node_with_score in doc_output], default=lambda o: o.__dict__, ensure_ascii=False)
        ctx.send_event(OutputDoc(data=json_data))
        return None
    
       
    @step()
    async def rerank(self, ctx: Context, ev:  OutputDoc | OutputQna  ) -> AnswerEvent:
		#combine the two events here 
        combine = ctx.collect_events(ev, [OutputDoc, OutputQna] )
        if combine is None:
            return None
        event_a, event_b = combine

        return AnswerEvent(new_nodes=json_data)

[dosubot[bot]]

To incorporate these events into your workflow and combine them into a single event, you can follow the approach you've outlined with some adjustments. Here's how you can structure your workflow to handle the serialization and combination of events:

1. Define Your Events: Ensure your event classes are properly defined and can be serialized using the JsonSerializer.

2. Send Events: In each step, serialize your data to JSON and send it as an event.

3. Combine Events: Use the collect_events method to wait for both OutputDoc and OutputQna events before proceeding to combine them.

Here's a refined version of your code:

from llama_index.core.workflow import Workflow, Context, Event
from llama_index.core.workflow.context_serializers import JsonSerializer

class OutputQna(Event):
    data: str

class OutputDoc(Event):
    data: str

class AnswerEvent(Event):
    new_nodes: str

class RAGWorkflow(Workflow):

    @step
    async def qna(self, ctx: Context, ev: QnaQuery) -> None:
        # Retrieval logic here
        json_data = json.dumps([node_with_score.__dict__ for node_with_score in qna_output], ensure_ascii=False)
        ctx.send_event(OutputQna(data=json_data))

    @step
    async def retriev(self, ctx: Context, ev: DocQuery) -> None:
        # Retrieval logic here
        json_data = json.dumps([node_with_score.__dict__ for node_with_score in doc_output], ensure_ascii=False)
        ctx.send_event(OutputDoc(data=json_data))

    @step
    async def rerank(self, ctx: Context, ev: OutputDoc | OutputQna) -> AnswerEvent | None:
        # Combine the two events here
        combined_events = ctx.collect_events(ev, [OutputDoc, OutputQna])
        if combined_events is None:
            return None

        event_a, event_b = combined_events
        # Combine the data from both events
        combined_data = f"{event_a.data} {event_b.data}"
        return AnswerEvent(new_nodes=combined_data)

Key Points:

• Serialization: Ensure that the data you send in events is serialized to JSON, which you've done using json.dumps.
• Event Collection: Use ctx.collect_events to wait for both OutputDoc and OutputQna events. This method will return None until both events are available.
• Combining Data: Once both events are collected, you can combine their data as needed and return a new AnswerEvent.

This setup should help you manage the serialization and combination of events within your workflow effectively.

To continue talking to Dosu, mention @dosu.

[hristogg]

@dosu can you show me the difference and what you have changed, it seems like the code is the same, but with this code I have the issue with serialization from the rerank

[dosubot[bot]]

The main focus of the previous guidance was on ensuring that your events are serialized correctly and that the collect_events method is used properly. However, if you're still encountering serialization issues in the rerank step, let's address the potential causes and solutions more explicitly:

1. Ensure Proper Serialization: Make sure that the JsonSerializer is used consistently for both serialization and deserialization. This ensures that any custom serialization logic is applied.

2. Check Event Data: Verify that the data being serialized in each event is JSON-compatible. This means ensuring that all objects are converted to dictionaries or other JSON-serializable formats before serialization.

3. Debugging Serialization: If serialization fails, it might be due to non-serializable objects in your data. You can add logging to inspect the data being serialized:

   import logging
   
   logging.basicConfig(level=logging.DEBUG)
   
   # In your qna and retriev steps
   logging.debug(f"Serialized QnA data: {json_data}")
   logging.debug(f"Serialized Doc data: {json_data}")

4. Combine Events Safely: When combining events in the rerank step, ensure that the data from both events is correctly deserialized before combining:

   @step
   async def rerank(self, ctx: Context, ev: OutputDoc | OutputQna) -> AnswerEvent | None:
       combined_events = ctx.collect_events(ev, [OutputDoc, OutputQna])
       if combined_events is None:
           return None
   
       event_a, event_b = combined_events
       # Deserialize data if necessary
       data_a = json.loads(event_a.data)
       data_b = json.loads(event_b.data)
   
       # Combine the data from both events
       combined_data = json.dumps(data_a + data_b, ensure_ascii=False)
       return AnswerEvent(new_nodes=combined_data)

Key Adjustments:

• Deserialization: Ensure that you deserialize the event data before combining it. This is crucial if the data was serialized to JSON strings.
• Logging: Use logging to debug and verify the data at each step, which can help identify where serialization might be failing.

These adjustments should help address serialization issues in the rerank step. If problems persist, further inspection of the specific data being serialized might be necessary.

To continue talking to Dosu, mention @dosu.

[hristogg]

@dosu , I have tried to create a dummy workflow to see where the issue is and it still fails in the same way, I am starting to think that the issue is in sending two events which creates a dictionary of the two classes, here is my sample code, any suggestions would be welcomed. I think I am followingthe documentation on handling context:

import asyncio
from llama_index.core.prompts.prompt_type import PromptType
from llama_index.core.workflow import (
    Context,
    Event,
    StartEvent,
    StopEvent,
    Workflow,
    step,
)

class StepA(Event):
    data: str

class StepB(Event):
    data: str

class StepA1(Event):
    data: str

class StepB1(Event):
    data: str

class RAGWorkflow(Workflow):

    @step
    async def StartingStep(self, ctx: Context, ev: StartEvent) -> StepA | StepB:
        print('Going into starting step')
        ctx.send_event(StepA(data='Start -> StepA'))
        ctx.send_event(StepB(data='Start -> StepB'))
        return None
    
    @step
    async def stepa(self, ctx: Context, ev: StepA) -> StepA1:
        data = ev.data
        print(f'Reading data in stepA: {data}')
        data = data + ' + StepA append'
        print(f'Data after something from stepA {data}')
        ctx.send_event(StepA1(data=data))
        return None
    
    @step
    async def stepb(self, ctx: Context, ev: StepB) -> StepB1:
        data = ev.data
        print(f'Reading data in stepB: {data}')
        data = data + '+ stepB Append'
        print(f'Data after something from stepB: {data}')
        ctx.send_event(StepB1(data=data))
        return None
    
    @step
    async def combine(self, ctx: Context, ev: StepA1 | StepB1 ) -> StopEvent:
        print('Going into step combine')
        all_events = ctx.collect_events(ev, [StepA1, StepB1])
        if all_events is None:
            return None
        event_a, event_b = all_events
        #print(f'From Combine event_a is {event_a}')
        #print(f'From Combine event_b is {event_b}')
        combined = event_a.data + event_b.data
        print(f'Combined is {combined}')
        return StopEvent(result=combined)

#async def main():
#    w = RAGWorkflow(timeout=60, verbose=True)
#    result = await w.run()
#    print(result)
#    return(result)
async def main():
    from llama_deploy import ControlPlaneConfig, WorkflowServiceConfig, deploy_workflow

    print('start deployment')
    await deploy_workflow(
        workflow=RAGWorkflow(timeout=100,verbose=True),
        workflow_config=WorkflowServiceConfig(
            host="127.0.0.1",
            port=8002,
            service_name="my_workflow",  # This will make it accessible to all interfaces on the host
        ),
        control_plane_config=ControlPlaneConfig(),
    )
    print("Created workflow successfully")


if __name__ == "__main__":
    asyncio.run(main())

This still fails on:
ERROR:llama_deploy.services.workflow - Encountered error in task 34cc3f7e-6242-4cd5-a3eb-0f650c7813f9! keys must be str, int, float, bool or None, not ModelMetaclass
Due to:

{<class '__main__.StepA1'>: [], <class '__main__.StepB1'>: []}