{para.strip()}
' for para in paragraphs if para.strip()]) else: logging.error(f"Unsupported format: {format_type}") return f"Unsupported format: {format_type}" def split_into_snippets(text: str, context_size: int) -> List[str]: """Split text into manageable snippets based on context size.""" sentences = re.split(r'(?<=[.!?]) +', text) snippets = [] current_snippet = "" for sentence in sentences: if len(current_snippet) + len(sentence) + 1 > context_size: if current_snippet: snippets.append(current_snippet.strip()) current_snippet = sentence + " " else: snippets.append(sentence.strip()) current_snippet = "" else: current_snippet += sentence + " " if current_snippet.strip(): snippets.append(current_snippet.strip()) return snippets def build_prompts(snippets: List[str], prompt_instruction: str, custom_prompt: Optional[str], snippet_num: Optional[int] = None) -> str: """Build formatted prompts from text snippets.""" if snippet_num is not None: if 1 <= snippet_num <= len(snippets): selected_snippets = [snippets[snippet_num - 1]] else: return f"Error: Invalid snippet number. Please choose between 1 and {len(snippets)}." else: selected_snippets = snippets prompts = [] base_prompt = custom_prompt if custom_prompt else prompt_instruction for idx, snippet in enumerate(selected_snippets, start=1): if len(selected_snippets) > 1: prompt_header = f"{base_prompt} Part {idx} of {len(selected_snippets)}: ---\n" else: prompt_header = f"{base_prompt} ---\n" framed_prompt = f"{prompt_header}{snippet}\n---" prompts.append(framed_prompt) return "\n\n".join(prompts) def send_to_model(prompt, model_selection, hf_model_choice, hf_custom_model, hf_api_key, groq_model_choice, groq_api_key, openai_api_key, openai_model_choice, cohere_api_key=None, cohere_model=None, glhf_api_key=None, glhf_model=None, glhf_custom_model=None): """Primary wrapper for model interactions with error handling.""" logging.info("send to model starting...") if not prompt or not prompt.strip(): return gr.HTML(""), "Error: No prompt provided", None try: logging.info("sending to model preparation.") # Basic input validation valid_selections = ["Clipboard only", "HuggingFace Inference", "Groq API", "OpenAI ChatGPT", "Cohere API", "GLHF API"] if model_selection not in valid_selections: return gr.HTML(""), "Error: Invalid model selection", None # Check environment API keys env_api_keys = { "GROQ_API_KEY": os.getenv('GROQ_API_KEY'), "OPENAI_API_KEY": os.getenv('OPENAI_API_KEY'), "COHERE_API_KEY": os.getenv('COHERE_API_KEY'), "GLHF_API_KEY": os.getenv('GLHF_API_KEY') } for key_name, key_value in env_api_keys.items(): if not key_value: logging.warning(f"No {key_name} found in environment") # Model-specific validation - check only required keys if model_selection == "Groq API" and not groq_api_key: groq_api_key = env_api_keys.get("GROQ_API_KEY") if not groq_api_key: return gr.HTML(""), "Error: Groq API key required", None elif model_selection == "OpenAI ChatGPT" and not openai_api_key: openai_api_key = env_api_keys.get("OPENAI_API_KEY") if not openai_api_key: return gr.HTML(""), "Error: OpenAI API key required", None elif model_selection == "GLHF API" and not glhf_api_key: glhf_api_key = env_api_keys.get("GLHF_API_KEY") if not glhf_api_key: return gr.HTML(""), "Error: GLHF API key required", None # Call the implementation function clipboard_status, summary, download_file = send_to_model_impl( prompt=prompt.strip(), model_selection=model_selection, hf_model_choice=hf_model_choice, hf_custom_model=hf_custom_model, hf_api_key=hf_api_key, groq_model_choice=groq_model_choice, groq_api_key=groq_api_key, openai_api_key=openai_api_key, openai_model_choice=openai_model_choice, cohere_api_key=cohere_api_key or env_api_keys.get("COHERE_API_KEY"), cohere_model=cohere_model, glhf_api_key=glhf_api_key, glhf_model=glhf_model, glhf_custom_model=glhf_custom_model, use_rate_limits=False # Adjust based on your needs ) return clipboard_status, summary, download_file except Exception as e: error_msg = str(e) or "Unknown error occurred" logging.error(f"Error in send_to_model: {error_msg}") return gr.HTML(f"Error: {error_msg}"), f"Error: {error_msg}", None finally: logging.info("send to model completed.") def send_to_model_impl(prompt, model_selection, hf_model_choice, hf_custom_model, hf_api_key, groq_model_choice, groq_api_key, openai_api_key, openai_model_choice, cohere_api_key=None, cohere_model=None, glhf_api_key=None, glhf_model=None, glhf_custom_model=None, use_rate_limits=False): """Implementation of model sending with all providers.""" logging.info("send to model impl commencing...") try: if model_selection == "Clipboard only": # Escape the prompt for JavaScript escaped_prompt = prompt.replace('"', '\\"').replace("'", "\\'").replace('\n', '\\n') # Create temporary file for download with tempfile.NamedTemporaryFile(delete=False, mode='w', suffix='.txt') as f: f.write(prompt) download_file = f.name # Create HTML with JavaScript using fallback methods html_template = f''' ''' # Return all three expected outputs: # 1. HTML component for clipboard action # 2. A success message for summary output # 3. The download file return gr.HTML(html_template), "Use Copy Text to Clipboard button below, then paste where you like.", download_file # Get the summary based on model selection if model_selection == "HuggingFace Inference": # Use the selected model ID directly model_id = hf_custom_model if hf_model_choice == "Custom Model" else hf_model_choice # Always try without API key first summary = send_to_hf_inference(prompt, model_id) if summary.startswith("Error: This model requires authentication") and hf_api_key: # Only try with API key if the model specifically requires it summary = send_to_hf_inference(prompt, model_id, hf_api_key, use_rate_limits) elif model_selection == "Groq API": if not groq_api_key: return gr.HTML(""), "Error: Groq API key required", None summary = send_to_groq(prompt, groq_model_choice, groq_api_key, use_rate_limits) elif model_selection == "OpenAI ChatGPT": if not openai_api_key: return "Error: OpenAI API key required", None summary = send_to_openai(prompt, openai_api_key, model=openai_model_choice, use_rate_limit=use_rate_limits) elif model_selection == "Cohere API": summary = send_to_cohere(prompt, cohere_api_key, cohere_model, use_rate_limits) elif model_selection == "GLHF API": if not glhf_api_key: return "Error: GLHF API key required", None # Handle model selection if glhf_model == "Custom Model": model_id = f"hf:{glhf_custom_model}" else: model_id = f"hf:{glhf_model}" summary = send_to_glhf(prompt, glhf_api_key, model_id, use_rate_limits) else: return "Error: Invalid model selection", None # Validate response if not summary: return gr.HTML(""), "Error: No response from model", None if not isinstance(summary, str): return gr.HTML(""), "Error: Invalid response type from model", None # Create download file for valid responses if not summary.startswith("Error"): with tempfile.NamedTemporaryFile(delete=False, mode='w', suffix='.txt') as f: f.write(summary) return gr.HTML(""), summary, f.name return gr.HTML(""), summary, None except Exception as e: error_msg = str(e) if not error_msg: error_msg = "Unknown error occurred" logging.error(f"Error in send_to_model_impl: {error_msg}") # FIX: Return all three values even in error case return gr.HTML(""), f"Error: {error_msg}", None def send_to_qwq(prompt: str): """Send prompt to QwQ API.""" try: from gradio_client import Client client = Client("Qwen/QwQ-32B-preview") # Call the add_text endpoint result = client.predict( _input={"files":[], "text": prompt}, _chatbot=[], api_name="/add_text" ) # Call the agent_run endpoint response = client.predict( _chatbot=result[1], # This is correct api_name="/agent_run" ) if isinstance(response, list) and len(response) > 0: # Extract text from first message in chat history if isinstance(response[0], list) and len(response[0]) > 0: if isinstance(response[0][1], dict): return response[0][1].get('text', 'No response text from QwQ') elif isinstance(response[0][1], str): return response[0][1] return 'No valid response from QwQ' return 'No response from QwQ' except Exception as e: logging.error(f"QwQ API error: {e}") return f"Error with QwQ API: {str(e)}" def send_to_hf_inference(prompt: str, model_name: str, api_key: str = None, use_rate_limit: bool = False) -> str: """Send prompt to HuggingFace Inference API.""" # Special handling for QwQ if model_name == "Qwen/QwQ-32B-preview": return send_to_qwq(prompt) def _send(): # Check token limits first is_within_limits, error_msg = check_token_limits(prompt, model_name) if not is_within_limits: return error_msg try: client = InferenceClient(token=api_key) if api_key else InferenceClient() response = client.text_generation( prompt, model=model_name, max_new_tokens=500, temperature=0.7, top_p=0.95, repetition_penalty=1.1 ) return str(response) except Exception as e: logging.error(f"HuggingFace inference error: {e}") return f"Error with HuggingFace inference: {str(e)}" return apply_rate_limit(_send, 16) if use_rate_limit else _send() def send_to_glhf(prompt: str, api_key: str, model_id: str, use_rate_limit: bool = False) -> str: """Send prompt to GLHF API.""" def _send(): try: import openai client = openai.OpenAI( api_key=api_key, base_url="https://glhf.chat/api/openai/v1", ) # For GLHF, always use streaming for reliability completion = client.chat.completions.create( stream=True, model=model_id, messages=[ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": prompt} ], ) response_text = [] for chunk in completion: if chunk.choices[0].delta.content is not None: response_text.append(chunk.choices[0].delta.content) return "".join(response_text) except Exception as e: logging.error(f"GLHF API error: {e}") return f"Error with GLHF API: {str(e)}" return apply_rate_limit(_send, 384) if use_rate_limit else _send() def send_to_openai(prompt: str, api_key: str, model: str = "gpt-3.5-turbo", use_rate_limit: bool = False) -> str: """Send prompt to OpenAI API.""" def _send(): try: from openai import OpenAI client = OpenAI(api_key=api_key) response = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": "You are a helpful assistant that provides detailed responses."}, {"role": "user", "content": prompt} ], temperature=0.7, max_tokens=500, top_p=0.95 ) if response.choices and len(response.choices) > 0: return response.choices[0].message.content return "Error: No response generated" except ImportError: return "Error: Please install the latest version of openai package" except Exception as e: logging.error(f"OpenAI API error: {e}") return f"Error with OpenAI API: {str(e)}" return apply_rate_limit(_send, 3000/60) if use_rate_limit else _send() def send_to_cohere(prompt: str, api_key: str = None, model: str = None, use_rate_limit: bool = False) -> str: """Send prompt to Cohere API with V2 and V1 fallback.""" def _send(): try: import cohere # Try V2 first try: client = cohere.ClientV2(api_key) if api_key else cohere.ClientV2() response = client.chat( model=model or "command-r-plus-08-2024", messages=[{ "role": "user", "content": prompt }], temperature=0.7, ) return response.message.content[0].text except Exception as v2_error: logging.warning(f"Cohere V2 failed, trying V1: {v2_error}") # Fallback to V1 client = cohere.Client(api_key) if api_key else cohere.Client() response = client.chat( message=prompt, model=model or "command-r-plus-08-2024", temperature=0.7, max_tokens=500, ) return response.text except Exception as e: logging.error(f"Cohere API error: {e}") return f"Error with Cohere API: {str(e)}" return apply_rate_limit(_send, 16) if use_rate_limit else _send() def send_to_groq(prompt: str, model_name: str, api_key: str, use_rate_limit: bool = False) -> str: """Send prompt to Groq API.""" def _send(): try: client = Groq(api_key=api_key) response = client.chat.completions.create( model=model_name, messages=[{ "role": "user", "content": prompt }], temperature=0.7, max_tokens=500, top_p=0.95 ) return response.choices[0].message.content except Exception as e: logging.error(f"Groq API error: {e}") return f"Error with Groq API: {str(e)}" return apply_rate_limit(_send, 4) if use_rate_limit else _send() def estimate_tokens(text: str) -> int: """Rough token estimation: ~4 characters per token on average""" return len(text) // 4 def check_token_limits(prompt: str, model_name: str) -> tuple[bool, str]: """Check if prompt might exceed model's token limits.""" token_limited_models = { "openai-community/gpt2": 1500, # 2048 - buffer "microsoft/phi-2": 1500, "TinyLlama/TinyLlama-1.1B-Chat-v1.0": 1500 } if model_name in token_limited_models: estimated_tokens = estimate_tokens(prompt) max_tokens = token_limited_models[model_name] if estimated_tokens > max_tokens: return False, f"Prompt too long (estimated {estimated_tokens} tokens). This model supports max {max_tokens} tokens." return True, "" def copy_to_clipboard(text): return gr.HTML(f""" """) def handle_model_selection(choice): """Handle model selection and update UI""" ctx_size = MODEL_CONTEXT_SIZES.get(choice, {}) if isinstance(ctx_size, dict): first_model = list(ctx_size.keys())[0] ctx_size = ctx_size[first_model] if choice == "OpenAI ChatGPT": model_choices = list(MODEL_CONTEXT_SIZES["OpenAI ChatGPT"].keys()) return [ gr.update(visible=False), # hf_options gr.update(visible=False), # groq_options gr.update(visible=True), # openai_options gr.update(visible=False), # cohere_options gr.update(visible=False), # glhf_options gr.update(value=ctx_size), # context_size gr.update(interactive=True), # send_model_btn gr.Dropdown(choices=model_choices, value=first_model), # openai_model gr.update(visible=False) # hf_custom_model visibility ] elif choice == "HuggingFace Inference": model_choices = list(MODEL_CONTEXT_SIZES["HuggingFace Inference"].keys()) return [ gr.update(visible=True), # hf_options gr.update(visible=False), # groq_options gr.update(visible=False), # openai_options gr.update(visible=False), # cohere_options gr.update(visible=False), # glhf_options gr.update(value=ctx_size), # context_size gr.update(interactive=True), # send_model_btn gr.Dropdown(choices=model_choices, value="mistralai/Mistral-7B-Instruct-v0.3"), gr.update(visible=False) # hf_custom_model initially hidden ] elif choice == "Groq API": model_choices = list(model_registry.groq_models.keys()) return [ gr.update(visible=False), # hf_options gr.update(visible=True), # groq_options gr.update(visible=False), # openai_options gr.update(visible=False), # cohere_options gr.update(visible=False), # glhf_options gr.update(value=ctx_size), # context_size gr.update(interactive=True), # send_model_btn gr.Dropdown(choices=model_choices, value=model_choices[0] if model_choices else None), gr.update(visible=False) # hf_custom_model visibility ] elif choice == "Cohere API": return [ gr.update(visible=False), # hf_options gr.update(visible=False), # groq_options gr.update(visible=False), # openai_options gr.update(visible=True), # cohere_options gr.update(visible=False), # glhf_options gr.update(value=ctx_size), # context_size gr.update(interactive=True), # send_model_btn gr.Dropdown(choices=[]), # not used gr.update(visible=False) # hf_custom_model visibility ] elif choice == "GLHF API": model_choices = list(MODEL_CONTEXT_SIZES["GLHF API"].keys()) return [ gr.update(visible=False), # hf_options gr.update(visible=False), # groq_options gr.update(visible=False), # openai_options gr.update(visible=False), # cohere_options gr.update(visible=True), # glhf_options gr.update(value=ctx_size), # context_size gr.update(interactive=True), # send_model_btn gr.Dropdown(choices=[]), # not used gr.update(visible=False) # hf_custom_model visibility ] # Default return for "Clipboard only" or other options return [ gr.update(visible=False), # hf_options gr.update(visible=False), # groq_options gr.update(visible=False), # openai_options gr.update(visible=False), # cohere_options gr.update(visible=False), # glhf_options gr.update(value=4096), # context_size gr.update(interactive=False), # send_model_btn gr.Dropdown(choices=[]), # not used gr.update(visible=False) # hf_custom_model visibility ] def copy_text_js(element_id: str) -> str: return f"""function() {{ let textarea = document.getElementById('{element_id}'); if (!textarea) return 'Element not found'; textarea.select(); try {{ document.execCommand('copy'); return 'Copied to clipboard!'; }} catch(err) {{ return 'Failed to copy: ' + err; }} }}""" def process_pdf(pdf, fmt, ctx_size): """Process PDF and return text and snippets""" try: if not pdf: return "Please upload a PDF file.", "", [], None # Extract text text = extract_text_from_pdf(pdf.name) if text.startswith("Error"): return text, "", [], None # Format content formatted_text = format_content(text, fmt) # Split into snippets snippets = split_into_snippets(formatted_text, ctx_size) # Save full text for download with tempfile.NamedTemporaryFile(delete=False, mode='w', suffix='.txt') as text_file: text_file.write(formatted_text) snippet_choices = [f"Snippet {i+1} of {len(snippets)}" for i in range(len(snippets))] return ( "PDF processed successfully!", formatted_text, snippets, snippet_choices, [text_file.name] ) except Exception as e: logging.error(f"Error processing PDF: {e}") return f"Error processing PDF: {str(e)}", "", [], None def generate_prompt(text, template, snippet_idx=None): """Generate prompt from text or selected snippet""" try: if not text: return "No text available.", "", None default_prompt = "Summarize the following text:" prompt_template = template if template else default_prompt if isinstance(text, list): # If text is list of snippets if snippet_idx is not None: if 0 <= snippet_idx < len(text): content = text[snippet_idx] else: return "Invalid snippet index.", "", None else: content = "\n\n".join(text) else: content = text prompt = f"{prompt_template}\n---\n{content}\n---" # Save prompt for download with tempfile.NamedTemporaryFile(delete=False, mode='w', suffix='.txt') as prompt_file: prompt_file.write(prompt) return "Prompt generated!", prompt, [prompt_file.name] except Exception as e: logging.error(f"Error generating prompt: {e}") return f"Error generating prompt: {str(e)}", "", None # Main Interface with gr.Blocks(css=""" .gradio-container {max-width: 90%; margin: 0 auto;} @media (max-width: 768px) {.gradio-container {max-width: 98%; padding: 10px;} .gr-row {flex-direction: column;} .gr-col {width: 100%; margin-bottom: 10px;}} """) as demo: # State variables pdf_content = gr.State("") snippets = gr.State([]) # Header gr.Markdown("# 📄 Smart PDF Summarizer") gr.Markdown("Upload a PDF document and get AI-powered summaries using various AI models.") with gr.Tabs() as tabs: # Tab 1: PDF Processing with gr.Tab("1️⃣ PDF Processing"): with gr.Row(): with gr.Column(scale=1): pdf_input = gr.File( label="📁 Upload PDF", file_types=[".pdf"] ) format_type = gr.Radio( choices=["txt", "md"], value="txt", label="📝 Output Format" ) context_size = gr.Slider( minimum=1000, maximum=200000, step=1000, value=4096, label="Context Size" ) gr.Markdown("### Context Size") with gr.Row(): for size_name, size_value in CONTEXT_SIZES.items(): gr.Button( size_name, size="sm", scale=1 ).click( lambda v=size_value: gr.update(value=v), None, context_size ) process_button = gr.Button("🔍 Process PDF", variant="primary") with gr.Column(scale=1): progress_status = gr.Textbox( label="Status", interactive=False, show_label=True, visible=True # Ensure error messages are always visible ) processed_text = gr.Textbox( label="Processed Text", lines=10, max_lines=50, show_copy_button=True ) download_full_text = gr.File(label="📥 Download Full Text") # Tab 2: Snippet Selection with gr.Tab("2️⃣ Snippet Selection"): with gr.Row(): with gr.Column(scale=1): snippet_selector = gr.Dropdown( label="Select Snippet", choices=[], interactive=True ) custom_prompt = gr.Textbox( label="✍️ Custom Prompt Template", placeholder="Enter your custom prompt here...", lines=2 ) generate_prompt_btn = gr.Button("Generate Prompt", variant="primary") with gr.Column(scale=1): generated_prompt = gr.Textbox( label="📋 Generated Prompt", lines=10, max_lines=50, show_copy_button=True, elem_id="generated_prompt", elem_classes="generated_prompt" ) with gr.Row(): download_prompt = gr.File(label="📥 Download Prompt") download_snippet = gr.File(label="📥 Download Selected Snippet") # Tab 3: Model Processing with gr.Tab("3️⃣ Model Processing"): with gr.Row(): with gr.Column(scale=1): model_choice = gr.Radio( choices=list(MODEL_CONTEXT_SIZES.keys()), value="Clipboard only", label="🤖 Provider Selection" ) # Model-specific option containers with gr.Column(visible=False) as openai_options: openai_model = gr.Dropdown( choices=list(MODEL_CONTEXT_SIZES["OpenAI ChatGPT"].keys()), value="gpt-3.5-turbo", label="OpenAI Model" ) openai_api_key = gr.Textbox( label="🔑 OpenAI API Key", type="password" ) with gr.Column(visible=False) as hf_options: hf_model = gr.Dropdown( choices=list(MODEL_CONTEXT_SIZES["HuggingFace Inference"].keys()), label="🔧 HuggingFace Model", value="mistralai/Mistral-7B-Instruct-v0.3", allow_custom_value=True ) hf_custom_model = gr.Textbox( label="Custom Model ID", placeholder="Enter custom model ID...", visible=False ) hf_api_key = gr.Textbox( label="🔑 HuggingFace API Key", type="password" ) with gr.Column(visible=False) as groq_options: groq_model = gr.Dropdown( choices=list(model_registry.groq_models.keys()), value=list(model_registry.groq_models.keys())[0] if model_registry.groq_models else None, label="Groq Model" ) groq_api_key = gr.Textbox( label="🔑 Groq API Key", type="password" ) groq_refresh_btn = gr.Button("🔄 Refresh Groq Models") with gr.Column(visible=False) as glhf_options: glhf_api_key = gr.Textbox( label="🔑 GLHF API Key", type="password" ) glhf_model = gr.Dropdown( choices=list(MODEL_CONTEXT_SIZES["GLHF API"].keys()), value="mistralai/Mistral-7B-Instruct-v0.3", label="Model Selection" ) glhf_custom_model = gr.Textbox( label="Custom Model ID", placeholder="Enter custom model ID...", visible=False ) with gr.Column(visible=False) as cohere_options: cohere_api_key = gr.Textbox( label="🔑 Cohere API Key", type="password" ) cohere_model = gr.Dropdown( choices=list(MODEL_CONTEXT_SIZES["Cohere API"].keys()), value="command-r-plus-08-2024", label="Cohere Model" ) # Action Buttons Row with gr.Row(): # Copy to Clipboard button with robust fallbacks copy_button = gr.HTML("""