When beginning the workflow from the terminal, it’s simple 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 may pause the workflow and look ahead to the consumer enter (See the human-in-the-loop instance on this official Llamaindex notebook). 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 option to ship streaming occasions to point the progress of the workflow, as proven within the pocket book 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"] = Discipline(
..., description="Kind of the occasion"
)
event_sender: str = Discipline(
..., description="Sender (workflow step identify) of the occasion"
)
event_content: Dict[str, Any] = Discipline(..., description="Content material of the occasion")
To allow sending streaming occasions, the workflow step must have entry to the shared context, which is completed by including @step(pass_context=True) decorator to the step definition. Then within the step definition, we are able to ship occasion messages concerning the progress by the context. For instance, within the tavily_query() step:
@step(pass_context=True)
async def tavily_query(self, ctx: Context, ev: StartEvent) -> TavilyResultsEvent:
ctx.knowledge["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 one other sort 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 offer 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 need to proceed to the following step after we’ve got acquired 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 following 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:
...# Watch for consumer enter
if not self.user_input_future.carried out():
user_response = await self.user_input_future
logger.data(f"gather_feedback_outline: Acquired consumer response: {user_response}")
# Course of user_response, which needs to be a JSON string
strive:
response_data = json.hundreds(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 operate run_workflow_endpoint() takes ResearchTopic as enter. Within the operate, an asynchronous generator event_generator() is outlined, which creates a process to run the workflow and streams the occasions to the shopper because the workflow progresses. When the workflow finishes, it’ll additionally stream the ultimate file outcomes to the shopper.
class ResearchTopic(BaseModel):
question: str = Discipline(..., instance="instance question")@app.submit("/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}")
strive:
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 shopper and dealing with file obtain requests. Since every workflow is assigned a singular workflow ID, we are able to map the consumer enter acquired from the shopper to the right workflow. By name the set_result() on the awaiting Future, the pending workflow can resume execution.
@app.submit("/submit_user_input")
async def submit_user_input(knowledge: dict = Physique(...)):
workflow_id = knowledge.get("workflow_id")
user_input = knowledge.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.carried out():
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 carried out")
return {"standing": "enter acquired"}
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
/ "last.pptx"
)
if file_path.exists():
return FileResponse(
path=file_path,
media_type="software/vnd.openxmlformats-officedocument.presentationml.presentation",
filename=f"last.pptx",
)
else:
increase HTTPException(status_code=404, element="File not discovered")
Within the frontend web page, after the consumer submits the analysis subject by 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):
strive:
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 major():
...
with st.sidebar:
with st.type(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 operating.")
The occasion knowledge streamed from the backend is fetched by httpx.AsyncClient and put right into a message queue for additional processing. Completely different info is extracted relying on the occasion varieties. For occasion sort “request_user_input”, the thread can 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 shopper:
async with shopper.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 knowledge..."))
data_json = None
async for knowledge in fetch_streaming_data(url, payload):
if knowledge:
strive:
data_json = json.hundreds(knowledge)
if "workflow_id" in data_json:
# Ship workflow_id to major thread
message_queue.put(("workflow_id", data_json["workflow_id"]))
proceed
elif "final_result" in data_json:
# Ship final_result to major 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 essential?
message_queue.put(("message", knowledge))
if data_json and "final_result" in data_json or "final_result" in str(knowledge):
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 knowledge.
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 part 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 sort “request_user_input”, we are going to show associated info in a separate container and collect consumer suggestions. As there may 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 are linked to every occasion to make sure the widgets don’t intervene with one another:
def gather_outline_feedback(placeholder):
container = placeholder.container()
with container:
if st.session_state.user_input_required:
knowledge = st.session_state.user_input_prompt
event_type = knowledge.get("event_type")
if event_type == "request_user_input":
abstract = knowledge.get("event_content").get("abstract")
define = knowledge.get("event_content").get("define")
prompt_message = knowledge.get("event_content").get(
"message", "Please evaluate the define."
)# show the content material for consumer enter
st.markdown("## Unique Abstract:")
st.text_area("Abstract", abstract, disabled=True, top=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 possibility.")
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
strive:
response = requests.submit(
"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"Did not submit consumer enter: {str(e)}")
return
...
In the long run, when the workflow run lastly finishes, the frontend shopper will get a response that comprises the trail to the ultimate generated information (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
strive:
# 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'<iframe src="knowledge:software/pdf;base64,{base64.b64encode(st.session_state.pdf_data).decode()}" width="100%" top="600px" sort="software/pdf"></iframe>',
unsafe_allow_html=True,
)
besides Exception as e:
st.error(f"Did not 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
strive:
# 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",
knowledge=pptx_data,
file_name="generated_slides.pptx",
mime="software/vnd.openxmlformats-officedocument.presentationml.presentation",
)
besides Exception as e:
st.error(f"Did not load the PPTX file: {str(e)}")
We are going to create a multi-service Docker software with docker-compose to run the frontend and backend apps.
model: '3.8'companies:
backend:
construct:
context: ./backend
args:
- --no-cache
ports:
- "8000:80"
networks:
- app-network
volumes:
- .env:/app/.env
- ./knowledge:/app/knowledge
- ./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 in the course of the execution, and show the ultimate consequence to the consumer.