Spaces:
Runtime error
Runtime error
File size: 3,556 Bytes
959296e c92c3bf d93dbdd 9b5b26a 4bfe53e 9b5b26a 4bfe53e df15652 0f61d7b 2764a1c 5d06296 1c61c68 4c2a00d d4578f1 4db54a2 e55e264 67c8f73 7686479 ed36d14 95de6d7 2945906 5d06296 95c49b7 5d06296 0e921e3 0f61d7b 5d06296 4999ab5 5d06296 a22a330 5d06296 4999ab5 f01a417 4999ab5 351e87d 5d06296 4999ab5 5d06296 df15652 5d06296 dac4869 6621a08 dac4869 a6ef072 dac4869 a6ef072 dac4869 5d06296 df15652 5d06296 b356cef df15652 5d06296 dac4869 a3ba8f9 dac4869 5d06296 a3ba8f9 5d06296 b356cef 5d06296 4999ab5 1c61c68 5d06296 |
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 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,tool
import sys # Add this import to fix the error
import datetime
import time
import requests
from checks.check_gui import StatusUI
from checks.failed_check import create_failed_gradio_ui
from checks.endpoint_check import check_public_endpoint
from checks.health_check import check_model_endpoint, should_launch_ui
import os
import pytz # Had to give it permission in Code agent
from tools.tools import load_tools
from agents.model import load_huggingface_model
from agents.prompts import load_prompts
from ContentGradio import ContentAgentUI
def initialize_agent(endpoint_uri: str):
# Load prompts
combined_prompts = load_prompts()
combined_tools = load_tools() ## from the tools folder (don't remove final answer)
"""Initialize and configure the CodeAgent"""
model = HfApiModel(
max_tokens=2096,
temperature=0.5,
model_id=endpoint_uri,
custom_role_conversions=None,
)
return CodeAgent(
model=model,
tools=combined_tools,
max_steps=6,
verbosity_level=3,
name="content_agent",
description="Evaluates whether text is polite or impolite.",
prompt_templates=combined_prompts(),
additional_authorized_imports=["pytz"]
)
def wake_up_endpoint(endpoint_uri, max_wait=30):
"""Poll the endpoint until it responds OK or timeout."""
# Get token from environment variable or secrets store
hf_token = os.environ.get("HF_TOKEN") # Or replace with your secret loading logic
headers = {"Authorization": f"Bearer {hf_token}"} if hf_token else {}
payload = {"inputs": "ping"} # Adjust to minimal valid input
start = time.time()
while time.time() - start < max_wait:
try:
print("Pinging endpoint...")
print(endpoint_uri)
response = requests.post(endpoint_uri, json=payload, timeout=5)
if response.ok:
print("✅ Endpoint is awake.")
return True
# If it's a 503 or 504, it's likely still warming up
print(f"Response: {response.status_code}, retrying...")
except requests.exceptions.RequestException as e:
print(f"Exception: {e}, retrying...")
time.sleep(3) # fixed short backoff; can be exponential if needed
print("❌ Timed out waiting for endpoint to wake up.")
return False
def main():
ui = StatusUI("🚦 Endpoint Logger")
ui.launch()
# Elsewhere in your code:
#ui.append("Checking endpoint…")
#ui.append("✅ Model responded OK.")
#ui.shutdown()
# Load endpoint and check health
# Prechecks
endpoint_uri = load_huggingface_model() # Get the URI for the endpoint
# Wake it up before health check
wake_up_successful = wake_up_endpoint(endpoint_uri)
if not wake_up_successful:
print("Warning: Could not wake up the endpoint. Exiting.")
exit(0)
is_healthy, status_info = check_model_endpoint(endpoint_uri) # Test the endpoint
if not is_healthy:
interface = create_failed_gradio_ui(status_info)
interface.launch(show_error=True, share=True)
return
# Initialize and run the agent
agent = initialize_agent(endpoint_uri)
# Create an instance of the ContentAgentUI class
ui = ContentAgentUI()
ui.pass_through_agent(agent) # Pass through the agent
if __name__ == "__main__":
main() |