Release

@step
    async def get_modules(self, ctx: Context, ev: StartEvent) -> GetModulesEvent:
        """Get modules step."""
        # get input data, store llm into ctx
        task = ev.get("task")

        if task is None:
            raise ValueError("'task' argument is required.")

        # format prompt and get result from LLM
        prompt = SELECT_PRMOPT_TEMPLATE.format(
            task=task, reasoning_modules=_REASONING_MODULES
        )

        print(f"==== begin get modules =====>> {self.get_elapsed_time()}")

        ctx.write_event_to_stream(
            ChatEvent(
                message=MessageFactory.text_progress_message("正在生成思考模块..."),
            ),
        )

        print(f"==== begin get modules =====>> {self.get_elapsed_time()}")

        result = await self.llm.acomplete(prompt)

        print(f"==== end get modules =====>> {self.get_elapsed_time()}")

        ctx.write_event_to_stream(
            ChatEvent(
                message=MessageFactory.text_progress_message("思考模块生成完成"),
            ),
        )

        print(f"==== end get modules =====>> {self.get_elapsed_time()}")

        return GetModulesEvent(task=task, modules=str(result))

@router.post("/api/chat/demo")
async def demo(
    request: Request,
):
    body = await request.json()
    query = body.get("query") or ""

    async def generate_events():
        workflow = SelfDiscoverWorkflow(
            llm=get_llm(),
            timeout=None,
        )
        handler = workflow.run(
            task=query,
        )
        async for ev in handler.stream_events():
            if isinstance(ev, ChatEvent):
                yield SSEResponse.format_string(
                    SSEResponse.EventType.DATA,
                    ev.message,
                )

        await handler

    return SSEResponse.send(generate_events())

import asyncio
from llama_index.core.workflow import Context, Workflow, step, StartEvent, StopEvent, Event
from llama_index.llms.openai import OpenAI

class StreamingWorkflow(Workflow):
    @step
    async def stream(self, ctx: Context, ev: StartEvent) -> StopEvent:
        llm = OpenAI(model="gpt-4o-mini")
        
        ctx.write_event_to_stream(Event(msg="Starting..."))
        resp = await llm.acomplete("Hello, world!")
        ctx.write_event_to_stream(Event(msg=resp.text))

        ctx.write_event_to_stream(Event(msg="Stopping..."))
        resp = await llm.acomplete("What is the weather in Tokyo?")
        ctx.write_event_to_stream(Event(msg=resp.text))

        return StopEvent(result="Done!")
    
async def main():
    w = StreamingWorkflow()
    handler = w.run()
    async for ev in handler.stream_events():
        if hasattr(ev, "msg"):
            print(ev.msg)
    result = await handler
    print(result)

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

@app.get("/")
async def root():
    w = StreamingWorkflow()
    handler = w.run()

    async def stream():
        async for ev in handler.stream_events():
            if hasattr(ev, "msg"):
                yield ev.msg + "\n"

    return StreamingResponse(stream(), media_type="text/plain")