Worklfow errors in collect_events

[sumitbindra]

When running the workflow using collect event, I am getting the follow linting error and an execution error:

I have checked to make sure both these events are generated accurately

Code:

@step
    async def generate_sql(self, ctx: Context,
                     ev: GetRelevantQuesEvent | RowSampleRetrieveEvent) -> TextToSQLEvent:
        """Generate SQL statement."""

        print(f"\n\nInside generate sql function\n\n")
        
        ready = ctx.collect_events(ev, [GetRelevantQuesEvent | RowSampleRetrieveEvent])
        
        if ready is None:
            print("Not enough events yet")
            return None

        print(f"\n\nReady events: {ready}\n\n")
        
        sql_manager = SQLProcessor()

linting error:

Argument of type "list[type[GetRelevantQuesEvent] | type[RowSampleRetrieveEvent]]" cannot be assigned to parameter "expected" of type "List[type[Event]]" in function "collect_events"
Type "UnionType" cannot be assigned to type "type[Event]" (pyright-extended)

Error:

Step get_relevant_ques_step produced event GetRelevantQuesEvent
Running step retrieve_row_samples

Context from the table:
Here are some relevant example rows (values in the columns above)

Step retrieve_row_samples produced event RowSampleRetrieveEvent
Running step generate_sql

Inside generate sql function

Not enough events yet
Step generate_sql produced no event
Running step generate_sql

Inside generate sql function

Not enough events yet
Step generate_sql produced no event
ERROR:asyncio:Exception in callback Dispatcher.span..wrapper..handle_future_result(span_id='Workflow.run...-2d61ff0a6afe', bound_args=<BoundArgumen..._responses'})>, instance=<llama_run_qu...x7fbb3c36fa90>, context=<_contextvars...x7fbb3bd8fb80>)() at /home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/llama_index/core/instrumentation/dispatcher.py:273
handle: <Handle Dispatcher.span..wrapper..handle_future_result(span_id='Workflow.run...-2d61ff0a6afe', bound_args=<BoundArgumen..._responses'})>, instance=<llama_run_qu...x7fbb3c36fa90>, context=<_contextvars...x7fbb3bd8fb80>)() at /home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/llama_index/core/instrumentation/dispatcher.py:273>
Traceback (most recent call last):
File "/nix/store/clx0mcir7qw8zk36zbr4jra789g3knf6-python3-3.11.10/lib/python3.11/asyncio/events.py", line 84, in _run
self._context.run(self._callback, *self._args)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/llama_index/core/instrumentation/dispatcher.py", line 283, in handle_future_result
exception = future.exception()
^^^^^^^^^^^^^^^^^^
asyncio.exceptions.CancelledError
ERROR:asyncio:Exception in callback Dispatcher.span..wrapper..handle_future_result(span_id='Workflow.run...-50e9300625b7', bound_args=<BoundArgumen...employers?'})>, instance=<llama_workfl...x7fbb3bd108d0>, context=<_contextvars...x7fbb3bd8eb40>)(<WorkflowHand...0.0 seconds')>) at /home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/llama_index/core/instrumentation/dispatcher.py:273
handle: <Handle Dispatcher.span..wrapper..handle_future_result(span_id='Workflow.run...-50e9300625b7', bound_args=<BoundArgumen...employers?'})>, instance=<llama_workfl...x7fbb3bd108d0>, context=<_contextvars...x7fbb3bd8eb40>)(<WorkflowHand...0.0 seconds')>) at /home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/llama_index/core/instrumentation/dispatcher.py:273>
Traceback (most recent call last):
File "/nix/store/clx0mcir7qw8zk36zbr4jra789g3knf6-python3-3.11.10/lib/python3.11/asyncio/events.py", line 84, in _run
self._context.run(self._callback, *self._args)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/llama_index/core/instrumentation/dispatcher.py", line 285, in handle_future_result
raise exception
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/llama_index/core/workflow/workflow.py", line 452, in _run_workflow
raise WorkflowTimeoutError(msg)
llama_index.core.workflow.errors.WorkflowTimeoutError: Operation timed out after 10.0 seconds
INFO: 172.31.196.63:54098 - "POST /analyze HTTP/1.1" 500 Internal Server Error
ERROR: Exception in ASGI application
Traceback (most recent call last):
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/uvicorn/protocols/http/h11_impl.py", line 406, in run_asgi
result = await app( # type: ignore[func-returns-value]
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/uvicorn/middleware/proxy_headers.py", line 70, in call
return await self.app(scope, receive, send)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/fastapi/applications.py", line 1054, in call
await super().call(scope, receive, send)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/applications.py", line 112, in call
await self.middleware_stack(scope, receive, send)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/middleware/errors.py", line 187, in call
raise exc
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/middleware/errors.py", line 165, in call
await self.app(scope, receive, _send)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/middleware/cors.py", line 93, in call
await self.simple_response(scope, receive, send, request_headers=headers)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/middleware/cors.py", line 144, in simple_response
await self.app(scope, receive, send)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 62, in call
await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
raise exc
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
await app(scope, receive, sender)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/routing.py", line 715, in call
await self.middleware_stack(scope, receive, send)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/routing.py", line 735, in app
await route.handle(scope, receive, send)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/routing.py", line 288, in handle
await self.app(scope, receive, send)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/routing.py", line 76, in app
await wrap_app_handling_exceptions(app, request)(scope, receive, send)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
raise exc
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
await app(scope, receive, sender)
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/starlette/routing.py", line 73, in app
response = await f(request)
^^^^^^^^^^^^^^^^
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/fastapi/routing.py", line 301, in app
raw_response = await run_endpoint_function(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/fastapi/routing.py", line 212, in run_endpoint_function
return await dependant.call(**values)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/runner/workspace/main.py", line 89, in search_to_sql
response = await workflow_run(tablename, request_data.query)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/runner/workspace/llama_workflows.py", line 224, in workflow_run
response = await workflow.run(table_name = table_name, query_str = query_str)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/llama_index/core/workflow/workflow.py", line 452, in _run_workflow
raise WorkflowTimeoutError(msg)
llama_index.core.workflow.errors.WorkflowTimeoutError: Operation timed out after 10.0 seconds
ERROR:asyncio:Task exception was never retrieved
future: <Task finished name='Task-74' coro=<Workflow.run.._run_workflow() done, defined at /home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/llama_index/core/workflow/workflow.py:399> exception=InvalidStateError('invalid state')>
Traceback (most recent call last):
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/llama_index/core/workflow/workflow.py", line 452, in _run_workflow
raise WorkflowTimeoutError(msg)
llama_index.core.workflow.errors.WorkflowTimeoutError: Operation timed out after 10.0 seconds

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/home/runner/workspace/.pythonlibs/lib/python3.11/site-packages/llama_index/core/workflow/workflow.py", line 456, in _run_workflow
result.set_exception(e)
asyncio.exceptions.InvalidStateError: invalid state

[sumitbindra]

there was an error in collect_events formtting

[dosubot[bot]]

The issue you're encountering with the collect_events method is due to the incorrect use of the union type (|) within the list of event types. Instead of using the union operator, you should specify each event type separately in the list. Here's how you can correct the code:

@step
async def generate_sql(self, ctx: Context,
                 ev: GetRelevantQuesEvent | RowSampleRetrieveEvent) -> TextToSQLEvent:
    """Generate SQL statement."""

    print(f"\n\nInside generate sql function\n\n")
    
    # Corrected line: specify each event type separately in the list
    ready = ctx.collect_events(ev, [GetRelevantQuesEvent, RowSampleRetrieveEvent])
    
    if ready is None:
        print("Not enough events yet")
        return None

    print(f"\n\nReady events: {ready}\n\n")
    
    sql_manager = SQLProcessor()

By listing GetRelevantQuesEvent and RowSampleRetrieveEvent separately in the list passed to collect_events, you should resolve the linting error. This change ensures that collect_events correctly interprets the expected event types ^[1].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Other}

[battogtokh14]

hi

[InAnYan]

Accidentally discovered your post.

You can improve the chances of being answered by:

• Better formatting your text (e.g.: wrap logs with three backticks).
• Trying to compose an MWE (minimal working example).