Spaces:
Sleeping
Sleeping
File size: 3,020 Bytes
42173a6 d8d4f18 26ad448 b1944f5 26ad448 d8d4f18 26ad448 42173a6 d8d4f18 42173a6 d8d4f18 42173a6 d8d4f18 26ad448 d8d4f18 26ad448 d8d4f18 26ad448 d8d4f18 42173a6 d8d4f18 42173a6 26ad448 d8d4f18 3fcda94 42173a6 26ad448 d8d4f18 42173a6 d8d4f18 26ad448 d8d4f18 42173a6 26ad448 d8d4f18 42173a6 d8d4f18 26ad448 42173a6 d8d4f18 42173a6 d8d4f18 42173a6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | import os
import gradio as gr
import aiohttp
import asyncio
LICENSING_SERVICE_BASE = os.getenv(
"LICENSING_SERVICE_BASE",
"https://gbdlicensing01containerapp.delightfulocean-00db0bf9.westeurope.azurecontainerapps.io"
).rstrip("/")
# Map from human topic index name
INDEX_MAP = {
"Licensing Fundamentals": "licensing-fundamentals",
# add more as you onboard them
}
async def call_licensing_api(
endpoint: str,
api_key: str,
payload: dict,
) -> dict:
"""Generic POST to /create_index, /answer_question or /classify_topic."""
url = f"{LICENSING_SERVICE_BASE}/{endpoint}"
headers = {"X-API-Key": api_key, "Content-Type": "application/json"}
timeout = aiohttp.ClientTimeout(total=60)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(url, json=payload, headers=headers) as resp:
resp.raise_for_status()
return await resp.json()
async def classify_topic(api_key: str, question: str) -> str:
"""Ask your service to pick exactly one topic from INDEX_MAP.keys()."""
payload = {
"question": question,
"topics": list(INDEX_MAP.keys())
}
resp = await call_licensing_api("classify_topic", api_key, payload)
return resp.get("topic", "")
async def get_answer(api_key: str, selected_topic: str, question: str):
"""Get summary + answer, auto-classifying if needed."""
if not selected_topic or selected_topic not in INDEX_MAP:
selected_topic = await classify_topic(api_key, question)
index_name = INDEX_MAP[selected_topic]
payload = {"index_name": index_name, "question": question}
resp = await call_licensing_api("answer_question", api_key, payload)
return resp.get("summary", ""), resp.get("answer", "")
async def main():
host = os.getenv("HOST", "0.0.0.0")
port = int(os.getenv("PORT", 7860))
with gr.Blocks() as demo:
api_key_input = gr.Textbox(
label="Licensing Service API Key",
placeholder="Paste your API key here",
type="password"
)
topic = gr.Dropdown(
choices=list(INDEX_MAP.keys()),
label="Topic (optional)",
value=None # no placeholder—None means “no selection”
)
question = gr.Textbox(
lines=3,
placeholder="Type your question here…",
label="Your Question"
)
summary = gr.Textbox(
lines=2,
interactive=False,
label="Summary"
)
answer = gr.Textbox(
lines=5,
interactive=False,
label="Answer"
)
submit = gr.Button("Submit")
submit.click(
fn=get_answer,
inputs=[api_key_input, topic, question],
outputs=[summary, answer]
)
demo.launch(
server_name=host,
server_port=port,
share=True
)
if __name__ == "__main__":
asyncio.run(main())
|