概述

因为等待事件是一种常见的模式，上下文Context对象有一个方便的函数，collect_events()。它将捕获事件并存储它们，返回 None 直到收集到它需要的所有事件。这些事件将按照指定的顺序附加到collect_events 的输出中。

实现逻辑

1.定义几个事件，事件紧紧简单的保存字符串内容

2.定义处理步骤，这几个处理函数，都从StartEvent后开始执行

3.定义run_query处理步骤，该步骤要等到以上三个步骤执行完成后才开始执行。此时，就需要使用collect_events函数来协助把多个事件集中起来。

实现代码

from llama_index.core.workflow import (
    Event,
    StartEvent,
    StopEvent,
    Workflow,
    step,
)
​
from llama_index.core.memory import VectorMemory
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, Settings
from llama_index.llms.ollama import Ollama
from llama_index.core.workflow import Context
​
import asyncio
import random
​
Settings.llm = Ollama(model="llama3.2", request_timeout=360)
​
class InputEvent(Event):
    input: str
​
class SetupEvent(Event):
    error: bool
​
class QueryEvent(Event):
    query: str
​
#定义事件流步骤处理函数
class CollectExampleFlow(Workflow):
    @step
    async def setup(self, ctx: Context, ev: StartEvent) -> SetupEvent:
        # generically start everything up
        if not hasattr(self, "setup") or not self.setup:
            self.setup = True
            print("I got set up")
        return SetupEvent(error=False)
​
    @step
    async def collect_input(self, ev: StartEvent) -> InputEvent:
        if hasattr(ev, "input"):
            # perhaps validate the input
            print("I got some input")
            return InputEvent(input=ev.input)
​
    @step
    async def parse_query(self, ev: StartEvent) -> QueryEvent:
        if hasattr(ev, "query"):
            # parse the query in some way
            print("I got a query")
            return QueryEvent(query=ev.query)
​
    @step
    async def run_query(
        self, ctx: Context, ev: InputEvent | SetupEvent | QueryEvent
    ) -> StopEvent | None:
        ready = ctx.collect_events(ev, [QueryEvent, InputEvent, SetupEvent])
        # 等待所有事件都完成后才开始执行
        if ready is None:
            print("Not enough events yet")
            return None
​
        # run the query
        print("Now I have all the events")
        print(ready)
​
        result = f"Ran query '{ready[0].query}' on input '{ready[1].input}'"
        return StopEvent(result=result)
​
# 画出事件依赖图
from llama_index.utils.workflow import draw_all_possible_flows
draw_all_possible_flows(CollectExampleFlow, filename="CollectExampleFlow.html")
​
# 主函数
async def main():
    c = CollectExampleFlow()
    result = await c.run(input="Here's some input", query="Here's my question")
    print(result)
​
# run main
if __name__ == '__main__':
    asyncio.run(main())

输出

运行后的输出如下：

<class 'NoneType'>
<class '__main__.InputEvent'>
<class '__main__.QueryEvent'>
<class 'llama_index.core.workflow.events.StopEvent'>
<class '__main__.SetupEvent'>
CollectExampleFlow.html
I got some input
I got a query
Not enough events yet
Not enough events yet
Now I have all the events
[QueryEvent(query="Here's my question"), InputEvent(input="Here's some input"), SetupEvent(error=False)]
Ran query 'Here's my question' on input 'Here's some input'

从以上输出可以看到，每个事件都被触发，并且收集事件重复返回 None ，直到足够的事件到达。

事件依赖图

小结

通过Context的ctx.collect_events函数，可以实现等待多个分支都成功执行后再执行指定步骤。