Spaces:
Running
Running
File size: 6,028 Bytes
dad171a 81f854b dad171a 476db6b dad171a 476db6b dad171a 188c8df dad171a d6d7069 81f854b dad171a f5fd9fa dad171a |
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 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
import gradio as gr
import requests
import json
import os
import time
from collections import defaultdict
BASE_URL = "https://api.jigsawstack.com/v1"
headers = {
"x-api-key": os.getenv("JIGSAWSTACK_API_KEY")
}
# Rate limiting configuration
request_times = defaultdict(list)
MAX_REQUESTS = 20 # Maximum requests per time window
TIME_WINDOW = 3600 # Time window in seconds (1 hour)
def get_real_ip(request: gr.Request):
"""Extract real IP address using x-forwarded-for header or fallback"""
if not request:
return "unknown"
forwarded = request.headers.get("x-forwarded-for")
if forwarded:
ip = forwarded.split(",")[0].strip() # First IP in the list is the client's
else:
ip = request.client.host # fallback
return ip
def check_rate_limit(request: gr.Request):
"""Check if the current request exceeds rate limits"""
if not request:
return True, "Rate limit check failed - no request info"
ip = get_real_ip(request)
now = time.time()
request_times[ip] = [t for t in request_times[ip] if now - t < TIME_WINDOW]
# Check if rate limit exceeded
if len(request_times[ip]) >= MAX_REQUESTS:
time_remaining = int(TIME_WINDOW - (now - request_times[ip][0]))
time_remaining_minutes = round(time_remaining / 60, 1)
time_window_minutes = round(TIME_WINDOW / 60, 1)
return False, f"Rate limit exceeded. You can make {MAX_REQUESTS} requests per {time_window_minutes} minutes. Try again in {time_remaining_minutes} minutes."
# Add current request timestamp
request_times[ip].append(now)
return True, ""
def generate_embedding(input_type, text_content, url, content_type, token_overflow_mode, request: gr.Request):
"""Generate embeddings using JigsawStack Embedding API with rate limiting"""
# Check rate limit first
rate_limit_ok, rate_limit_msg = check_rate_limit(request)
if not rate_limit_ok:
return rate_limit_msg, ""
# Validate inputs
if input_type == "Text" and not text_content:
return "Error: Please provide text content.", ""
if input_type == "URL" and not url:
return "Error: Please provide a URL.", ""
try:
payload = {
"type": content_type,
"token_overflow_mode": token_overflow_mode
}
if input_type == "Text":
payload["text"] = text_content.strip()
elif input_type == "URL":
payload["url"] = url.strip()
response = requests.post(
f"{BASE_URL}/embedding",
headers=headers,
json=payload
)
response.raise_for_status()
result = response.json()
if not result.get("success"):
error_msg = f"Error: API call failed - {result.get('message', 'Unknown error')}"
return error_msg, ""
embedding = result.get("embeddings", [])
embedding_str = json.dumps(embedding, indent=2)
return "Embedding generated successfully!", embedding_str
except requests.exceptions.RequestException as e:
return f"Request failed: {str(e)}", ""
except Exception as e:
return f"An unexpected error occurred: {str(e)}", ""
with gr.Blocks() as demo:
gr.Markdown("""
<div style='text-align: center; margin-bottom: 24px;'>
<h1 style='font-size:2.2em; margin-bottom: 0.2em;'>🧩 Generate vector embeddings</h1>
<p style='font-size:1.2em; margin-top: 0;'>Generate vector embeddings from various content types.</p>
<p style='font-size:1em; margin-top: 0.5em;'>For more details and API usage, see the <a href='https://jigsawstack.com/docs/api-reference/ai/embedding?slug=docs&slug=api-reference&slug=ai&slug=embedding' target='_blank'>documentation</a>.</p>
</div>
""")
with gr.Row():
with gr.Column():
gr.Markdown("#### Content Input")
input_type = gr.Radio([
"Text",
"URL"
], value="Text", label="Select Input Type")
text_content = gr.Textbox(
label="Text Content",
placeholder="Enter the text content to generate embeddings for...",
visible=True,
lines=5
)
url = gr.Textbox(
label="URL",
placeholder="Enter the URL of the resource...",
visible=False
)
content_type = gr.Dropdown(
choices=["text", "text-other", "image", "audio", "pdf"],
value="text",
label="Content Type",
info="Select the type of content being processed"
)
token_overflow_mode = gr.Radio(
choices=["error", "truncate"],
value="error",
label="Token Overflow Mode",
info="How to handle input that exceeds token limits"
)
generate_btn = gr.Button("Generate Embedding", variant="primary")
with gr.Column():
gr.Markdown("#### Embedding Result")
status_message = gr.Textbox(label="Status", interactive=False)
embedding_result = gr.Textbox(
label="Vector Embedding",
interactive=False,
lines=15,
max_lines=25
)
def toggle_inputs(selected):
if selected == "Text":
return gr.update(visible=True), gr.update(visible=False)
else: # URL
return gr.update(visible=False), gr.update(visible=True)
input_type.change(toggle_inputs, inputs=[input_type], outputs=[text_content, url])
generate_btn.click(
generate_embedding,
inputs=[input_type, text_content, url, content_type, token_overflow_mode],
outputs=[status_message, embedding_result]
)
demo.launch()
|