import os
import gradio as gr
from checks.status_check import is_endpoint_healthy
from checks.endpoint_utils import wake_endpoint
class ContentAgentUI:
"""
Gradio UI that:
- shows a minimal control panel first (status + Start button),
- auto-initializes the agent on load if the endpoint is already healthy,
- otherwise lets the user 'Start Agent' (wake -> health -> init),
- reveals the main chat panel (with header, guidance, examples, footer) after init.
"""
def __init__(self, endpoint_uri: str, is_healthy: bool, health_message: str, agent_initializer, agent_type:str, compute:str ):
self.endpoint_uri = endpoint_uri
self.is_healthy = bool(is_healthy)
self.health_message = health_message or ""
self.agent_initializer = agent_initializer # callable: (uri) -> CodeAgent
self.agent_type = agent_type or ""
self.compute = compute or ""
# set in build()
self.app: gr.Blocks | None = None
self.status_box = None
self.control_panel = None
self.main_panel = None
self.prompt = None
self.reply = None
self.agent_state = None
self.examples_radio = None
self.clear = None
# ---------- helpers ----------
def _create_user_guidance(self):
gr.Markdown(f"""
Content Agent determines whether content is polite and uses the following classification:
- polite
- somewhat polite
- neutral
- impolite
The 0–1 number is the confidence for the current classification label (e.g., P(impolite) or P(polite)).
Technology
- App is running {self.agent_type} text generation model.
- Agent uses Intel's Polite Guard NLP library tool
- Compute: {self.compute}
- Content Agent's LLM runs on-demand and will drop into sleep mode to save resources
""")
def _initial_status_text(self) -> str:
# neutral; on_load will set real status and maybe auto-init
return "Checking endpoint status…"
def _get_example(self):
example_root = os.path.join(os.path.dirname(__file__), "..", "examples")
examples = []
if os.path.exists(example_root):
example_files = [os.path.join(example_root, f) for f in os.listdir(example_root) if f.endswith(".txt")]
for file_path in example_files:
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
examples.append(f.read())
return examples
def _create_examples(self):
examples = self._get_example()
if examples:
gr.Markdown("### Examples")
example_radio = gr.Radio(choices=examples, label="Select one of these examples to load")
# fill the prompt when an example is picked
example_radio.change(fn=lambda ex: ex, inputs=example_radio, outputs=self.prompt)
else:
gr.Markdown("*No examples found.*")
def _get_css(self):
# Optional: Adjust path if needed for hosted environments
css_content = ""
css_path = os.path.join(os.getcwd(), "ui", "styles.css")
if ( os.path.exists(css_path)):
# Also check if it's readable and has content
with open(css_path, 'r') as f:
css_content = f.read()
return css_content
# ---------- agent call ----------
@staticmethod
def _call_agent(text: str, agent) -> str:
try:
if agent is None:
return "Content Agent's end-point is sleeping and will need to be started. Click 'Start Agent' and wait a few minutes."
return str(agent.run(text)) # smolagents.CodeAgent API
except Exception as e:
return f"Error: {e}"
# ---------- UI build ----------
def build(self) -> gr.Blocks:
if self.app is not None:
return self.app
with gr.Blocks(css=self._get_css()) as demo:
# global header (always visible)
gr.Markdown("# Content Agent")
# Control panel (shown first; may auto-hide on load)
with gr.Group(visible=True, elem_id="control_panel" ) as self.control_panel:
self.status_box = gr.Textbox(
label="Status",
value=self._initial_status_text(),
lines=8,
interactive=False,
)
start_btn = gr.Button("Start Agent")
gr.HTML("""
It may take up to 5 minutes to wake up the agent.
""")
# Main panel (hidden until agent is initialized)
with gr.Group(visible=False, elem_id="main_panel" ) as self.main_panel:
# English only
strInput="Content Input"
strPlaceholder="Copy and paste your content for evaluation here..."
strSubmit="Submit"
strOutput="Label and Score"
strClear="Clear"
# Guidance / about
self._create_user_guidance()
# Chat controls
self.agent_state = gr.State(None)
self.prompt = gr.Textbox(label=strInput, placeholder=strPlaceholder, lines=10)
with gr.Row(equal_height=True):
submit_btn = gr.Button(strSubmit)
self.clear = gr.Button(strClear)
self.reply = gr.Textbox(label=strOutput, interactive=False, lines=12, max_lines=20)
# Use bound methods to submit content
submit_btn.click(self._call_agent, inputs=[self.prompt, self.agent_state], outputs=self.reply)
self.prompt.submit(self._call_agent, inputs=[self.prompt, self.agent_state], outputs=self.reply)
self.clear.click(fn=lambda: "", outputs=self.prompt)
# Examples (optional)
self._create_examples()
# Footer
gr.HTML("")
# --- AUTO INIT ON LOAD IF HEALTHY ---
def on_load():
healthy, msg = is_endpoint_healthy(self.endpoint_uri)
if healthy:
try:
agent = self.agent_initializer(self.endpoint_uri)
return (
f"Endpoint healthy {msg}. Agent initialized.",
gr.update(visible=False), # hide control panel
gr.update(visible=True), # show main panel
agent,
)
except Exception as e:
return (
f"Agent init failed: {e}",
gr.update(visible=True),
gr.update(visible=False),
None,
)
# not healthy → keep Start button path
return (
f"The AI LLM is sleeping due to inactivity: {msg}\nClick 'Start Agent' to wake and initialize.",
gr.update(visible=True),
gr.update(visible=False),
None,
)
demo.load(
on_load,
inputs=None,
outputs=[self.status_box, self.control_panel, self.main_panel, self.agent_state],
)
# --- MANUAL START (wake → health → init) ---
def on_start():
lines: list[str] = []
def push(s: str):
lines.append(s)
return ("\n".join(lines), gr.update(), gr.update(), None)
# Wake with progress
yield push("Waking endpoint… this can take several minutes for a cold start.")
ok, err = wake_endpoint(self.endpoint_uri, max_wait=600, poll_every=5.0, log=lines.append)
yield ("\n".join(lines), gr.update(), gr.update(), None) # flush all logs
if not ok:
yield push(f"[Server message] {err or 'wake failed'}")
return
# Health → init
yield push("Endpoint awake ✅. Checking health…")
healthy, msg = is_endpoint_healthy(self.endpoint_uri)
if not healthy:
yield push(f"[Server message] {msg}")
return
yield push("Initializing agent…")
try:
agent = self.agent_initializer(self.endpoint_uri)
except Exception as e:
yield push(f"Agent init failed: {e}")
return
yield ("Agent initialized ✅", gr.update(visible=False), gr.update(visible=True), agent)
start_btn.click(
on_start,
inputs=None,
outputs=[self.status_box, self.control_panel, self.main_panel, self.agent_state],
)
self.app = demo
return self.app
# ---------- public API ----------
def launch(self, **kwargs):
return self.build().launch(**kwargs)