01_basic_workflow.py

from llama_index.core.workflow import Workflow, step, Event, Context
from llama_index.core.workflow.events import (
    StartEvent,
    StopEvent,
    InputRequiredEvent,
    HumanResponseEvent,
)
from llama_index.core.workflow.handler import WorkflowHandler

from llama_index.llms.ollama import Ollama
from llama_index.core.llms import ChatMessage


model_name = "qwen2.5:0.5b"
llm = Ollama(model=model_name, request_timeout=120.0)


class ProgressEvent(Event):
    def __init__(self, msg: str):
        super().__init__()
        self.msg = msg


class RetryEvent(Event):
    pass


class FinalResultEvent(Event):
    pass


class HumanInteractionWorkflow(Workflow):
    @step
    async def ask_question(
        self, ctx: Context, ev: StartEvent | RetryEvent
    ) -> InputRequiredEvent:
        if isinstance(ev, StartEvent):
            ctx.write_event_to_stream(
                ProgressEvent(msg=f"I am doing query on '{ev.query}'")
            )
        else:
            ctx.write_event_to_stream(
                ProgressEvent(msg=f"I am doing query again on '{ev.query}'")
            )
        await ctx.set("original_query", ev.query)
        payload = llm.chat([ChatMessage(role="user", content=ev.query)])
        return InputRequiredEvent(prefix="", query=ev.query, payload=payload)

    @step
    async def human_review(
        self, ctx: Context, ev: HumanResponseEvent
    ) -> RetryEvent | FinalResultEvent:
        ctx.write_event_to_stream(
            ProgressEvent(msg=f"The human has responded {ev.response}")
        )
        original_query = await ctx.get("original_query")
        if ev.response == "yes":
            return FinalResultEvent(result=f"Result of {original_query}")
        else:
            return RetryEvent(query=original_query)

    @step
    async def final_result(self, ctx: Context, ev: FinalResultEvent) -> StopEvent:
        ctx.write_event_to_stream(ProgressEvent(msg="final result is preparing"))
        original_query = await ctx.get("original_query")
        print(original_query, ev.result)
        return StopEvent(result=ev.result)


async def executor(query: str):
    workflow = HumanInteractionWorkflow(timeout=None, verbose=False)
    try:
        handler: WorkflowHandler = workflow.run(query=query)

        async for event in handler.stream_events():
            if isinstance(event, InputRequiredEvent):
                # query = event.query
                payload = event.payload
                print("[Result]:", payload)
                print(
                    "[Assessment]: If you are satisfied type 'yes' for retry type 'no'"
                )
                user_input = input("[Review]:")
                handler.ctx.send_event(HumanResponseEvent(response=user_input))
            elif isinstance(event, ProgressEvent):
                print("[Progress]", event.msg)

        final_result = await handler
        print("[Final Result]", final_result)
    except Exception as e:
        print("[Error]", e)


import asyncio


def main():
    query = input("[Input]: ")
    asyncio.run(executor(query=query))


if __name__ == "__main__":
    main()


# Run : python hitl-workflow/01_basic_workflow.py
# Say something, and then based on the response provide your feedback