def collect_events(
        self, ev: Event, expected: List[Type[Event]], buffer_id: Optional[str] = None
    ) -> Optional[List[Event]]:
        """
        Collects events for buffering in workflows.

        This method adds the current event to the internal buffer and attempts to collect all
        expected event types. If all expected events are found, they will be returned in order.
        Otherwise, it returns None and restores any collected events back to the buffer.

        Args:
            ev (Event): The current event to add to the buffer.
            expected (List[Type[Event]]): List of expected event types to collect.
            buffer_id (str): A unique identifier for the events collected. Ideally this should be
            the step name, so to avoid any interference between different steps. If not provided,
            a stable identifier will be created using the list of expected events.

        Returns:
            Optional[List[Event]]: List of collected events in the order of expected types if all
                                  expected events are found; otherwise None.

        """
        buffer_id = buffer_id or self._get_event_buffer_id(expected)

        if buffer_id not in self._event_buffers:
            self._event_buffers[buffer_id] = defaultdict(list)

        event_type_path = self._get_full_path(type(ev))
        self._event_buffers[buffer_id][event_type_path].append(ev)

        retval: List[Event] = []
        for e_type in expected:
            e_type_path = self._get_full_path(e_type)
            e_instance_list = self._event_buffers[buffer_id].get(e_type_path, [])
            if e_instance_list:
                retval.append(e_instance_list.pop(0))
            else:
                # We already know we don't have all the events
                break

        if len(retval) == len(expected):
            return retval

        # put back the events if unable to collect all
        for i, ev_to_restore in enumerate(retval):
            e_type = type(retval[i])
            e_type_path = self._get_full_path(e_type)
            self._event_buffers[buffer_id][e_type_path].append(ev_to_restore)

        return None

        Collects events for buffering in workflows.

        This method adds the current event to the internal buffer and attempts to collect all
        expected event types. If all expected events are found, they will be returned in order.
        Otherwise, it returns None and restores any collected events back to the buffer.

        Args:
            ev (Event): The current event to add to the buffer.
            expected (List[Type[Event]]): List of expected event types to collect.
            buffer_id (str): A unique identifier for the events collected. Ideally this should be
            the step name, so to avoid any interference between different steps. If not provided,
            a stable identifier will be created using the list of expected events.

        Returns:
            Optional[List[Event]]: List of collected events in the order of expected types if all
                                  expected events are found; otherwise None.

        
llama-index-core/llama_index/core/workflow/context.py