When beginning the workflow from the terminal, it’s easy to see which step it’s executing and the logging we put in these steps.
We are able to additionally allow the human-in-the-loop interplay by merely utilizing user_feedback = enter()
within the workflow. This can pause the workflow and await the consumer enter (See the human-in-the-loop instance on this official Llamaindex pocket book). Nonetheless, to have the ability to obtain the identical performance in a user-friendly interface, we’d like extra modifications to the unique workflow.
Workflow can take a very long time to execute, so for a greater consumer expertise, Llamaindex offered a solution to ship streaming occasions to point the progress of the workflow, as proven within the pocket book right here. In my workflow, I outline a WorkflowStreamingEvent
class to incorporate helpful details about the occasion message, resembling the kind of the occasion, and from which step it’s despatched:
class WorkflowStreamingEvent(BaseModel):
event_type: Literal["server_message", "request_user_input"] = Area(
..., description="Kind of the occasion"
)
event_sender: str = Area(
..., description="Sender (workflow step title) of the occasion"
)
event_content: Dict[str, Any] = Area(..., description="Content material of the occasion")
To allow sending streaming occasions, the workflow step must have entry to the shared context, which is finished by including @step(pass_context=True)
decorator to the step definition. Then within the step definition, we will ship occasion messages concerning the progress via the context. For instance, within the tavily_query()
step:
@step(pass_context=True)
async def tavily_query(self, ctx: Context, ev: StartEvent) -> TavilyResultsEvent:
ctx.information["research_topic"] = ev.user_query
question = f"arxiv papers concerning the state-of-the-art of {ev.user_query}"
ctx.write_event_to_stream(
Occasion(
msg=WorkflowStreamingEvent(
event_type="server_message",
event_sender=examine.currentframe().f_code.co_name,
event_content={"message": f"Querying Tavily with: '{question}'"},
).model_dump()
)
)
On this instance, we set the event_type
to be “server_message”
. It signifies that it’s an replace message and no consumer motion is required. We’ve got one other kind of occasion "request_user_input"
that signifies a consumer enter is required. For instance, within the gather_feedback_outline()
step within the workflow, after producing the slide textual content outlines from the unique paper abstract, a message is shipped to immediate the consumer to supply approval and suggestions on the define textual content:
@step(pass_context=True)
async def gather_feedback_outline(
self, ctx: Context, ev: OutlineEvent
) -> OutlineFeedbackEvent | OutlineOkEvent:
"""Current consumer the unique paper abstract and the outlines generated, collect suggestions from consumer"""
...# Ship a particular occasion indicating that consumer enter is required
ctx.write_event_to_stream(
Occasion(
msg=json.dumps(
{
"event_type": "request_user_input",
"event_sender": examine.currentframe().f_code.co_name,
"event_content": {
"abstract": ev.abstract,
"define": ev.define.dict(),
"message": "Do you approve this define? If not, please present suggestions.",
},
}
)
)
)
...
These occasions are dealt with otherwise within the backend API and the frontend logic, which I’ll describe intimately within the later sections of this text.
When sending a "request_user_input"
occasion to the consumer, we solely wish to proceed to the following step after now we have obtained the consumer enter. As proven within the workflow diagram above, it both proceeds to the outlines_with_layout()
step if the consumer approves the define, or to the summary2outline()
step once more if the consumer doesn’t approve.
That is achieved utilizing the Future()
object from Python’s asyncio
library. Within the SlideGenerationWorkflow
class, we set an attribute self.user_input_future = asyncio.Future()
that may be waited on within the gather_feedback_outline()
step. The next execution of the workflow is conditioned on the content material of the consumer suggestions:
@step(pass_context=True)
async def gather_feedback_outline(
self, ctx: Context, ev: OutlineEvent
) -> OutlineFeedbackEvent | OutlineOkEvent:
...# Await consumer enter
if not self.user_input_future.performed():
user_response = await self.user_input_future
logger.data(f"gather_feedback_outline: Received consumer response: {user_response}")
# Course of user_response, which must be a JSON string
attempt:
response_data = json.masses(user_response)
approval = response_data.get("approval", "").decrease().strip()
suggestions = response_data.get("suggestions", "").strip()
besides json.JSONDecodeError:
# Deal with invalid JSON
logger.error("Invalid consumer response format")
increase Exception("Invalid consumer response format")
if approval == ":materials/thumb_up:":
return OutlineOkEvent(abstract=ev.abstract, define=ev.define)
else:
return OutlineFeedbackEvent(
abstract=ev.abstract, define=ev.define, suggestions=suggestions
)
We arrange the backend utilizing fastAPI, expose a POST endpoint to deal with requests, and provoke the workflow run. The asynchronous perform run_workflow_endpoint()
takes ResearchTopic
as enter. Within the perform, an asynchronous generator event_generator()
is outlined, which creates a process to run the workflow and streams the occasions to the consumer because the workflow progresses. When the workflow finishes, it’ll additionally stream the ultimate file outcomes to the consumer.
class ResearchTopic(BaseModel):
question: str = Area(..., instance="instance question")@app.put up("/run-slide-gen")
async def run_workflow_endpoint(subject: ResearchTopic):
workflow_id = str(uuid.uuid4())
wf = SummaryAndSlideGenerationWorkflow(wid=workflow_id, timeout=2000, verbose=True)
wf.add_workflows(
summary_gen_wf=SummaryGenerationWorkflow(
wid=workflow_id, timeout=800, verbose=True
)
)
wf.add_workflows(
slide_gen_wf=SlideGenerationWorkflow(
wid=workflow_id, timeout=1200, verbose=True
)
)
async def event_generator():
loop = asyncio.get_running_loop()
logger.debug(f"event_generator: loop id {id(loop)}")
yield f"{json.dumps({'workflow_id': workflow_id})}nn"
process = asyncio.create_task(wf.run(user_query=subject.question))
logger.debug(f"event_generator: Created process {process}")
attempt:
async for ev in wf.stream_events():
logger.data(f"Sending message to frontend: {ev.msg}")
yield f"{ev.msg}nn"
await asyncio.sleep(0.1) # Small sleep to make sure correct chunking
final_result = await process
# Assemble the obtain URL
download_pptx_url = f"http://backend:80/download_pptx/{workflow_id}"
download_pdf_url = f"http://backend:80/download_pdf/{workflow_id}"
final_result_with_url = {
"consequence": final_result,
"download_pptx_url": download_pptx_url,
"download_pdf_url": download_pdf_url,
}
yield f"{json.dumps({'final_result': final_result_with_url})}nn"
besides Exception as e:
error_message = f"Error in workflow: {str(e)}"
logger.error(error_message)
yield f"{json.dumps({'occasion': 'error', 'message': error_message})}nn"
lastly:
# Clear up
workflows.pop(workflow_id, None)
return StreamingResponse(event_generator(), media_type="textual content/event-stream")
Along with this endpoint, there are endpoints for receiving consumer enter from the consumer and dealing with file obtain requests. Since every workflow is assigned a singular workflow ID, we will map the consumer enter obtained from the consumer to the right workflow. By name the set_result()
on the awaiting Future
, the pending workflow can resume execution.
@app.put up("/submit_user_input")
async def submit_user_input(information: dict = Physique(...)):
workflow_id = information.get("workflow_id")
user_input = information.get("user_input")
wf = workflows.get(workflow_id)
if wf and wf.user_input_future:
loop = wf.user_input_future.get_loop() # Get the loop from the long run
logger.data(f"submit_user_input: wf.user_input_future loop id {id(loop)}")
if not wf.user_input_future.performed():
loop.call_soon_threadsafe(wf.user_input_future.set_result, user_input)
logger.data("submit_user_input: set_result known as")
else:
logger.data("submit_user_input: future already performed")
return {"standing": "enter obtained"}
else:
increase HTTPException(
status_code=404, element="Workflow not discovered or future not initialized"
)
The obtain endpoint additionally identifies the place the ultimate file is positioned based mostly on the workflow ID.
@app.get("/download_pptx/{workflow_id}")
async def download_pptx(workflow_id: str):
file_path = (
Path(settings.WORKFLOW_ARTIFACTS_PATH)
/ "SlideGenerationWorkflow"
/ workflow_id
/ "remaining.pptx"
)
if file_path.exists():
return FileResponse(
path=file_path,
media_type="utility/vnd.openxmlformats-officedocument.presentationml.presentation",
filename=f"remaining.pptx",
)
else:
increase HTTPException(status_code=404, element="File not discovered")
Within the frontend web page, after the consumer submits the analysis subject via st.text_input()
, a long-running course of is began in a background thread in a brand new occasion loop for receiving the streamed occasions from the backend, with out interfering with the remainder of the web page:
def start_long_running_task(url, payload, message_queue, user_input_event):
attempt:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(
get_stream_data(url, payload, message_queue, user_input_event)
)
loop.shut()
besides Exception as e:
message_queue.put(("error", f"Exception in background thread: {str(e)}"))...
def predominant():
...
with st.sidebar:
with st.kind(key="slide_gen_form"):
question = st.text_input(
"Enter the subject of your analysis:",
)
submit_button = st.form_submit_button(label="Submit")
if submit_button:
# Reset the workflow_complete flag for a brand new workflow
st.session_state.workflow_complete = False
# Begin the long-running process in a separate thread
if (
st.session_state.workflow_thread is None
or not st.session_state.workflow_thread.is_alive()
):
st.write("Beginning the background thread...")
st.session_state.workflow_thread = threading.Thread(
goal=start_long_running_task,
args=(
"http://backend:80/run-slide-gen",
{"question": question},
st.session_state.message_queue,
st.session_state.user_input_event,
),
)
st.session_state.workflow_thread.begin()
st.session_state.received_lines = []
else:
st.write("Background thread is already working.")
The occasion information streamed from the backend is fetched by httpx.AsyncClient
and put right into a message queue for additional processing. Totally different info is extracted relying on the occasion varieties. For occasion kind “request_user_input”
, the thread can also be paused till the consumer enter is offered.
async def fetch_streaming_data(url: str, payload: dict = None):
async with httpx.AsyncClient(timeout=1200.0) as consumer:
async with consumer.stream("POST", url=url, json=payload) as response:
async for line in response.aiter_lines():
if line:
yield lineasync def get_stream_data(url, payload, message_queue, user_input_event):
# message_queue.put(("message", "Beginning to fetch streaming information..."))
data_json = None
async for information in fetch_streaming_data(url, payload):
if information:
attempt:
data_json = json.masses(information)
if "workflow_id" in data_json:
# Ship workflow_id to predominant thread
message_queue.put(("workflow_id", data_json["workflow_id"]))
proceed
elif "final_result" in data_json:
# Ship final_result to predominant thread
message_queue.put(("final_result", data_json["final_result"]))
proceed
event_type = data_json.get("event_type")
event_sender = data_json.get("event_sender")
event_content = data_json.get("event_content")
if event_type in ["request_user_input"]:
# Ship the message to the principle thread
message_queue.put(("user_input_required", data_json))
# Wait till consumer enter is offered
user_input_event.wait()
user_input_event.clear()
proceed
else:
# Ship the road to the principle thread
message_queue.put(("message", format_workflow_info(data_json)))
besides json.JSONDecodeError: # todo: is that this needed?
message_queue.put(("message", information))
if data_json and "final_result" in data_json or "final_result" in str(information):
break # Cease processing after receiving the ultimate consequence
We retailer the messages within the st.session_state
and use a st.expander()
to show and replace these streamed information.
if st.session_state.received_lines:
with expander_placeholder.container():
# Create or replace the expander with the newest truncated line
expander = st.expander(st.session_state.expander_label)
for line in st.session_state.received_lines:
expander.write(line)
expander.divider()
To make sure the UI stays responsive and shows the occasion messages when they’re being processed in a background thread, we use a customed autorefresh element to refresh the web page at a set interval:
if not st.session_state.workflow_complete:
st_autorefresh(interval=2000, restrict=None, key="data_refresh")
When the streamed occasion is of kind “request_user_input”
, we are going to show associated info in a separate container and collect consumer suggestions. As there will be a number of occasions that require consumer enter from one workflow run, we put them in a message queue and ensure to assign a singular key to the st.suggestions()
, st.text_area()
and st.button()
which might be linked to every occasion to make sure the widgets don’t intrude with one another:
def gather_outline_feedback(placeholder):
container = placeholder.container()
with container:
if st.session_state.user_input_required:
information = st.session_state.user_input_prompt
event_type = information.get("event_type")
if event_type == "request_user_input":
abstract = information.get("event_content").get("abstract")
define = information.get("event_content").get("define")
prompt_message = information.get("event_content").get(
"message", "Please overview the define."
)# show the content material for consumer enter
st.markdown("## Unique Abstract:")
st.text_area("Abstract", abstract, disabled=True, peak=400)
st.divider()
st.markdown("## Generated Slide Define:")
st.json(define)
st.write(prompt_message)
# Outline distinctive keys for widgets
current_prompt = st.session_state.prompt_counter
approval_key = f"approval_state_{current_prompt}"
feedback_key = f"user_feedback_{current_prompt}"
# Show the approval suggestions widget
approval = st.suggestions("thumbs", key=approval_key)
st.write(f"Present Approval state is: {approval}")
logging.data(f"Present Approval state is: {approval}")
# Show the suggestions textual content space
suggestions = st.text_area(
"Please present suggestions if in case you have any:", key=feedback_key
)
# Deal with the submission of consumer response
if st.button(
"Submit Suggestions", key=f"submit_response_{current_prompt}"
):
if not st.session_state.user_response_submitted:
# Retrieve approval and suggestions utilizing distinctive keys
approval_state = st.session_state.get(approval_key)
user_feedback = st.session_state.get(feedback_key, "")
# Guarantee approval_state is legitimate
if approval_state not in [0, 1]:
st.error("Please choose an approval choice.")
return
user_response = {
"approval": (
":materials/thumb_down:"
if approval_state == 0
else ":materials/thumb_up:"
),
"suggestions": user_feedback,
}
# Ship the consumer's response to the backend
attempt:
response = requests.put up(
"http://backend:80/submit_user_input",
json={
"workflow_id": st.session_state.workflow_id,
"user_input": json.dumps(user_response),
},
)
response.raise_for_status()
logging.data(
f"Backend response for submitting approval: {response.status_code}"
)
besides requests.RequestException as e:
st.error(f"Didn't submit consumer enter: {str(e)}")
return
...
Ultimately, when the workflow run lastly finishes, the frontend consumer will get a response that incorporates the trail to the ultimate generated recordsdata (similar slide deck in pdf format for rendering within the UI and pptx format for downloading as the ultimate consequence). We show the pdf file and create a button for downloading the pptx file:
if "download_url_pdf" in st.session_state and st.session_state.download_url_pdf:
download_url_pdf = st.session_state.download_url_pdf
attempt:
# Fetch the PDF content material
pdf_response = requests.get(download_url_pdf)
pdf_response.raise_for_status()
st.session_state.pdf_data = pdf_response.content materialst.markdown("### Generated Slide Deck:")
# Show the PDF utilizing an iframe
st.markdown(
f'',
unsafe_allow_html=True,
)
besides Exception as e:
st.error(f"Didn't load the PDF file: {str(e)}")
# Present the obtain button for PPTX if accessible
if (
"download_url_pptx" in st.session_state
and st.session_state.download_url_pptx
):
download_url_pptx = st.session_state.download_url_pptx
attempt:
# Fetch the PPTX content material
pptx_response = requests.get(download_url_pptx)
pptx_response.raise_for_status()
pptx_data = pptx_response.content material
st.download_button(
label="Obtain Generated PPTX",
information=pptx_data,
file_name="generated_slides.pptx",
mime="utility/vnd.openxmlformats-officedocument.presentationml.presentation",
)
besides Exception as e:
st.error(f"Didn't load the PPTX file: {str(e)}")
We are going to create a multi-service Docker utility with docker-compose
to run the frontend and backend apps.
model: '3.8'providers:
backend:
construct:
context: ./backend
args:
- --no-cache
ports:
- "8000:80"
networks:
- app-network
volumes:
- .env:/app/.env
- ./information:/app/information
- ./workflow_artifacts:/app/workflow_artifacts
- ~/.azure:/root/.azure
frontend:
construct:
context: ./frontend
args:
- --no-cache
ports:
- "8501:8501"
networks:
- app-network
networks:
app-network:
That’s it! Simply run docker-compose up
, and we now have an app that may run a analysis workflow based mostly on the consumer’s enter question, immediate the consumer for suggestions throughout the execution, and show the ultimate consequence to the consumer.