from typing import List, Literal, Tuple, TypedDict
import os
import gradio as gr
try:
from common import org_search_component as oss
from formatting import process_reasons, parse_pcs_descriptions, parse_geo_descriptions
from services import RfpRecommend, RfpFeedback
except ImportError:
from ..common import org_search_component as oss
from .formatting import process_reasons, parse_pcs_descriptions, parse_geo_descriptions
from .services import RfpRecommend, RfpFeedback
api = RfpRecommend()
reporting = RfpFeedback()
class LoggedComponents(TypedDict):
recommendations: gr.components.Component
ratings: List[gr.components.Component]
correctness: gr.components.Component
sufficiency: gr.components.Component
comments: gr.components.Component
email: gr.components.Component
def single_recommendation_response(
item_number: int,
rec_type: Literal["RFP"] = "RFP"
) -> gr.Radio:
"""Generates a radio button group to provide feedback for single recommendation indexed by `item_number`.
Since the index values start from `0` we add `1` to indicate the ordinal value in the info text.
Parameters
----------
item_number : int
Recommendation index starting from 0
Returns
-------
gr.Radio
"""
ordinal = str(item_number + 1)
suffix = "th"
if ordinal.endswith('1') and not ordinal.endswith('11'):
suffix = "st"
elif ordinal.endswith('2') and not ordinal.endswith('12'):
suffix = "nd"
elif ordinal.endswith('3') and not ordinal.endswith('13'):
suffix = "rd"
elem = gr.Radio(
choices=[
"Not relevant and not useful",
"Relevant but not useful",
"Relevant and useful"
],
label=f"Recommendation #{ordinal}",
info=f"Evaluate the {ordinal}{suffix} {rec_type} (if applicable)"
)
return elem
def recommend_invoke(recipient: gr.State):
response = api(candid_entity_id=recipient[0])
output = []
for rfp in (response.get("recommendations", []) or []):
output.append([
rfp["funder_id"],
rfp["funder_name"],
rfp["funder_address"],
rfp["amount"],
(
f""
f"{rfp['application_url']}"
),
rfp["deadline"],
rfp["description"],
parse_pcs_descriptions(rfp["taxonomy"]),
parse_geo_descriptions(rfp["area_served"])
])
if len(output) == 0:
raise gr.Error("No relevant RFPs were found, please try again in the future as new RFPs become available.")
return output, process_reasons(response.get("meta", {}) or {}), response
def build_recommender() -> Tuple[LoggedComponents, gr.Blocks]:
with gr.Blocks(theme=gr.themes.Soft(), title="RFP recommendations") as demo:
gr.Markdown(
"""
RFP recommendations
Receive recommendations for funding opportunities relevant to your work.
Please read the guide to get started.
"""
)
with gr.Row():
with gr.Column():
_, selected_org_state = oss.render()
with gr.Row():
recommend = gr.Button("Get recommendations", scale=5, variant="primary")
with gr.Row():
with gr.Accordion(label="Parameters used for recommendations", open=False):
reasons_output = gr.DataFrame(
col_count=3,
headers=["Reason category", "Reason value", "Reason description"],
interactive=False
)
rec_outputs = gr.DataFrame(
label="Recommended RFPs",
type="array",
headers=[
"Funder ID", "Name", "Address",
"Amount", "Application URL", "Deadline",
"Description", "About", "Where"
],
col_count=(9, "fixed"),
datatype=[
"number", "str", "str",
"str", "markdown", "date",
"str", "markdown", "markdown"
],
wrap=True,
max_height=1000,
column_widths=[
"5%", "10%", "20%",
"5", "15%", "5%",
"10%", "10%", "20%"
],
interactive=False
)
recommendations_json = gr.JSON(label="Recommended RFPs JSON", visible=False)
# pylint: disable=no-member
recommend.click(
fn=recommend_invoke,
inputs=[selected_org_state],
outputs=[rec_outputs, reasons_output, recommendations_json]
)
logged = LoggedComponents(
recommendations=recommendations_json
)
return logged, demo
def build_feedback(
components: LoggedComponents,
N: int = 5,
rec_type: Literal["RFP"] = "RFP",
) -> gr.Blocks:
def handle_feedback(*args):
try:
reporting(
recommendation_data=args[0],
ratings=list(args[1: (N + 1)]),
info_is_correct=args[N + 1],
info_is_sufficient=args[N + 2],
comments=args[N + 3],
email=args[N + 4]
)
gr.Info("Thank you for providing feedback!")
except Exception as ex:
if hasattr(ex, "response"):
error_msg = ex.response.json().get("response", {}).get("error")
raise gr.Error(f"Failed to submit feedback: {error_msg}")
raise gr.Error("Failed to submit feedback")
feedback_components = []
with gr.Blocks(theme=gr.themes.Soft(), title="Candid AI demo") as demo:
gr.Markdown("""
Help us improve this tool with your valuable feedback
Please provide feedback for the recommendations on the previous tab.
It is not required to provide feedback on all recommendations before submitting.
"""
)
with gr.Row():
with gr.Column():
with gr.Group():
for i in range(N):
f = single_recommendation_response(i, rec_type=rec_type)
feedback_components.append(f)
if "ratings" not in components:
components["ratings"] = [f]
else:
components["ratings"].append(f)
correctness = gr.Radio(
choices=["True", "False"],
label="Information is correct?",
info="Are the displayed RFP details correct?"
)
sufficiency = gr.Radio(
choices=["True", "False"],
label="Sufficient data?",
info="Is enough RFP data available to provide meaningful recommendations?"
)
comment = gr.Textbox(label="Additional comments (optional)", lines=4)
email = gr.Textbox(label="Your email (optional)", lines=1)
components["correctness"] = correctness
components["sufficiency"] = sufficiency
components["comments"] = comment
components["email"] = email
with gr.Row():
submit = gr.Button("Submit Feedback", variant='primary', scale=5)
gr.ClearButton(components=feedback_components, variant="stop")
# pylint: disable=no-member
submit.click(
fn=handle_feedback,
inputs=[comp for k, cl in components.items() for comp in (cl if isinstance(cl, list) else [cl])],
outputs=None,
show_api=False,
api_name=False,
preprocess=False,
)
return demo
def build_demo():
logger, recommender = build_recommender()
feedback = build_feedback(logger)
return gr.TabbedInterface(
interface_list=[recommender, feedback],
tab_names=["RFP recommendations", "Feedback"],
title="Candid's RFP recommendation engine",
theme=gr.themes.Soft()
)
if __name__ == '__main__':
app = build_demo()
app.queue(max_size=5).launch(
show_api=False,
auth=[
(os.getenv("APP_USERNAME"), os.getenv("APP_PASSWORD")),
(os.getenv("APP_PUBLIC_USERNAME"), os.getenv("APP_PUBLIC_PASSWORD")),
],
auth_message="Login to Candid's letter of intent demo",
ssr_mode=False
)