Models reached AGREEMENT
-Confidence level: {confidence}
-Models could not reach agreement
-Review all solutions above for best answer
-Multi-Agent Problem Solving System
", show_label=False) - - with gr.Row(): - with gr.Column(scale=2): - gr.Markdown("### Problem Input") - problem_input = gr.Textbox(label="Problem", placeholder="e.g., What is 32 + 63?", lines=3) - rounds_slider = gr.Slider(2, 6, value=2, step=1, label="Maximum Rounds") - solve_button = gr.Button("Solve Problem", elem_classes=["primary-button"]) - - status_text = gr.Markdown("### Status: Ready", elem_classes=["status-bar"], visible=True) - - with gr.Column(): - initial_solutions = gr.Markdown(elem_classes=["step-section"], visible=False) - round_judgment_1 = gr.Markdown(elem_classes=["step-section"], visible=False) - revised_solutions_1 = gr.Markdown(elem_classes=["step-section"], visible=False) - round_judgment_2 = gr.Markdown(elem_classes=["step-section"], visible=False) - revised_solutions_2 = gr.Markdown(elem_classes=["step-section"], visible=False) - round_judgment_3 = gr.Markdown(elem_classes=["step-section"], visible=False) - revised_solutions_3 = gr.Markdown(elem_classes=["step-section"], visible=False) - final_report = gr.HTML(elem_classes=["final-report"], visible=False) - - solve_button.click( - fn=solve_problem, - inputs=[ - problem_input, - rounds_slider - ], - outputs=[ - initial_solutions, - round_judgment_1, - revised_solutions_1, - round_judgment_2, - revised_solutions_2, - round_judgment_3, - revised_solutions_3, - final_report, - status_text - ] - ) - - return demo.queue() - -if __name__ == "__main__": - demo = create_polythink_interface() +import os +import asyncio +import gradio as gr +import logging +from huggingface_hub import InferenceClient +import cohere +import google.generativeai as genai +from anthropic import Anthropic +import openai +from typing import List, Dict, Any, Optional +from datetime import datetime + +# Configure logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# --- Agent Class --- +class PolyThinkAgent: + def __init__(self, model_name: str, model_path: str, role: str = "solver", api_provider: str = None): + self.model_name = model_name + self.model_path = model_path + self.role = role + self.api_provider = api_provider + self.clients = {} + self.hf_token = None + self.inference = None + + def set_clients(self, clients: Dict[str, Any]): + """Set the API clients for this agent""" + self.clients = clients + if "huggingface" in clients: + self.hf_token = clients["huggingface"] + if self.hf_token: + self.inference = InferenceClient(token=self.hf_token) + + async def solve_problem(self, problem: str) -> Dict[str, Any]: + """Generate a solution to the given problem""" + try: + if self.api_provider == "cohere" and "cohere" in self.clients: + response = self.clients["cohere"].chat( + model=self.model_path, + message=f""" + PROBLEM: {problem} + INSTRUCTIONS: + - Provide a clear, concise solution in one sentence. + - Include brief reasoning in one additional sentence. + - Do not repeat the solution or add extraneous text. + """ + ) + solution = response.text.strip() + return {"solution": solution, "model_name": self.model_name} + + elif self.api_provider == "anthropic" and "anthropic" in self.clients: + response = self.clients["anthropic"].messages.create( + model=self.model_path, + messages=[{ + "role": "user", + "content": f""" + PROBLEM: {problem} + INSTRUCTIONS: + - Provide a clear, concise solution in one sentence. + - Include brief reasoning in one additional sentence. + - Do not repeat the solution or add extraneous text. + """ + }] + ) + solution = response.content[0].text.strip() + return {"solution": solution, "model_name": self.model_name} + + elif self.api_provider == "openai" and "openai" in self.clients: + response = self.clients["openai"].chat.completions.create( + model=self.model_path, + max_tokens=100, + messages=[{ + "role": "user", + "content": f""" + PROBLEM: {problem} + INSTRUCTIONS: + - Provide a clear, concise solution in one sentence. + - Include brief reasoning in one additional sentence. + - Do not repeat the solution or add extraneous text. + - Keep the response under 100 characters. + """ + }] + ) + solution = response.choices[0].message.content.strip() + return {"solution": solution, "model_name": self.model_name} + + elif self.api_provider == "huggingface" and self.inference: + prompt = f""" + PROBLEM: {problem} + INSTRUCTIONS: + - Provide a clear, concise solution in one sentence. + - Include brief reasoning in one additional sentence. + - Do not repeat the solution or add extraneous text. + - Keep the response under 100 characters. + SOLUTION AND REASONING: + """ + result = self.inference.text_generation( + prompt, model=self.model_path, max_new_tokens=5000, temperature=0.5 + ) + solution = result if isinstance(result, str) else result.generated_text + return {"solution": solution.strip(), "model_name": self.model_name} + + elif self.api_provider == "gemini" and "gemini" in self.clients: + model = self.clients["gemini"].GenerativeModel(self.model_path) + try: + response = model.generate_content( + f""" + PROBLEM: {problem} + INSTRUCTIONS: + - Provide a clear, concise solution in one sentence. + - Include brief reasoning in one additional sentence. + - Do not repeat the solution or add extraneous text. + - Keep the response under 100 characters. + """, + generation_config=genai.types.GenerationConfig( + temperature=0.5, + ) + ) + # Check response validity and handle different response structures + try: + # First try to access text directly if available + if hasattr(response, 'text'): + solution = response.text.strip() + # Otherwise check for candidates + elif hasattr(response, 'candidates') and response.candidates: + # Make sure we have candidates and parts before accessing + if hasattr(response.candidates[0], 'content') and hasattr(response.candidates[0].content, 'parts'): + solution = response.candidates[0].content.parts[0].text.strip() + else: + logger.warning(f"Gemini response has candidates but missing content structure: {response}") + solution = "Error parsing API response; incomplete response structure." + else: + # Fallback for when candidates is empty + logger.warning(f"Gemini API returned no candidates: {response}") + solution = "No solution generated; API returned empty response." + except Exception as e: + logger.error(f"Error extracting text from Gemini response: {e}, response: {response}") + solution = "Error parsing API response." + except Exception as e: + logger.error(f"Gemini API call failed: {e}") + solution = f"API error: {str(e)}" + return {"solution": solution, "model_name": self.model_name} + + else: + return {"solution": f"Error: Missing API configuration for {self.api_provider}", "model_name": self.model_name} + + except Exception as e: + logger.error(f"Error in {self.model_name}: {str(e)}") + return {"solution": f"Error: {str(e)}", "model_name": self.model_name} + async def evaluate_solutions(self, problem: str, solutions: List[Dict[str, Any]]) -> Dict[str, Any]: + """Evaluate solutions from solver agents""" + try: + prompt = f""" + PROBLEM: {problem} + SOLUTIONS: + 1. {solutions[0]['model_name']}: {solutions[0]['solution']} + 2. {solutions[1]['model_name']}: {solutions[1]['solution']} + INSTRUCTIONS: + - Extract the numerical final answer from each solution (e.g., 68 from '16 + 52 = 68'). + - Extract the key reasoning steps from each solution. + - Apply strict evaluation criteria: + * Numerical answers must match EXACTLY (including units and precision). + * Key reasoning steps must align in approach and logic. + - Output exactly: 'AGREEMENT: YES' if BOTH the numerical answers AND reasoning align perfectly. + - Output 'AGREEMENT: NO' followed by a one-sentence explanation if either the answers or reasoning differ in ANY way. + - Be conservative in declaring agreement - when in doubt, declare disagreement. + - Do not add scoring, commentary, or extraneous text. + EVALUATION: + """ + + if self.api_provider == "gemini" and "gemini" in self.clients: + # Instantiate the model for consistency and clarity + model = self.clients["gemini"].GenerativeModel(self.model_path) + # Use generate_content on the model instance + response = model.generate_content( + prompt, + generation_config=genai.types.GenerationConfig( + temperature=0.5, + ) + ) + + # Handle potential empty response or missing text attribute + try: + # First try to access text directly if available + if hasattr(response, 'text'): + judgment = response.text.strip() + # Otherwise check for candidates + elif hasattr(response, 'candidates') and response.candidates: + # Make sure we have candidates and parts before accessing + if hasattr(response.candidates[0], 'content') and hasattr(response.candidates[0].content, 'parts'): + judgment = response.candidates[0].content.parts[0].text.strip() + else: + logger.warning(f"Gemini response has candidates but missing content structure: {response}") + judgment = "AGREEMENT: NO - Unable to evaluate due to API response structure issue." + else: + # Fallback for when candidates is empty + logger.warning(f"Empty response from Gemini API: {response}") + judgment = "AGREEMENT: NO - Unable to evaluate due to API response issue." + except Exception as e: + logger.error(f"Error extracting text from Gemini response: {e}") + judgment = "AGREEMENT: NO - Unable to evaluate due to API response issue." + + return {"judgment": judgment, "reprompt_needed": "AGREEMENT: NO" in judgment.upper()} + + elif self.api_provider == "openai" and "openai" in self.clients: + response = self.clients["openai"].chat.completions.create( + model=self.model_path, + max_tokens=200, + messages=[{"role": "user", "content": prompt}] + ) + judgment = response.choices[0].message.content.strip() + return {"judgment": judgment, "reprompt_needed": "AGREEMENT: NO" in judgment.upper()} + + elif self.api_provider == "huggingface" and self.inference: + result = self.inference.text_generation( + prompt, model=self.model_path, max_new_tokens=200, temperature=0.5 + ) + judgment = result if isinstance(result, str) else result.generated_text + return {"judgment": judgment.strip(), "reprompt_needed": "AGREEMENT: NO" in judgment.upper()} + + else: + return {"judgment": f"Error: Missing API configuration for {self.api_provider}", "reprompt_needed": False} + + except Exception as e: + logger.error(f"Error in judge: {str(e)}") + return {"judgment": f"Error: {str(e)}", "reprompt_needed": False} + + async def reprompt_with_context(self, problem: str, solutions: List[Dict[str, Any]], judgment: str) -> Dict[str, Any]: + """Generate a revised solution based on previous solutions and judgment""" + try: + prompt = f""" + PROBLEM: {problem} + PREVIOUS SOLUTIONS: + 1. {solutions[0]['model_name']}: {solutions[0]['solution']} + 2. {solutions[1]['model_name']}: {solutions[1]['solution']} + JUDGE FEEDBACK: {judgment} + INSTRUCTIONS: + - Provide a revised, concise solution in one sentence. + - Include brief reasoning in one additional sentence. + - Address the judge's feedback. + """ + + if self.api_provider == "cohere" and "cohere" in self.clients: + response = self.clients["cohere"].chat( + model=self.model_path, + message=prompt + ) + solution = response.text.strip() + return {"solution": solution, "model_name": self.model_name} + + elif self.api_provider == "anthropic" and "anthropic" in self.clients: + response = self.clients["anthropic"].messages.create( + model=self.model_path, + max_tokens=100, + messages=[{"role": "user", "content": prompt}] + ) + solution = response.content[0].text.strip() + return {"solution": solution, "model_name": self.model_name} + + elif self.api_provider == "openai" and "openai" in self.clients: + response = self.clients["openai"].chat.completions.create( + model=self.model_path, + max_tokens=100, + messages=[{"role": "user", "content": prompt}] + ) + solution = response.choices[0].message.content.strip() + return {"solution": solution, "model_name": self.model_name} + + elif self.api_provider == "huggingface" and self.inference: + prompt += "\nREVISED SOLUTION AND REASONING:" + result = self.inference.text_generation( + prompt, model=self.model_path, max_new_tokens=500, temperature=0.5 + ) + solution = result if isinstance(result, str) else result.generated_text + return {"solution": solution.strip(), "model_name": self.model_name} + + elif self.api_provider == "gemini" and "gemini" in self.clients: + # Instantiate the model for consistency and clarity + model = self.clients["gemini"].GenerativeModel(self.model_path) + # Use generate_content + response = model.generate_content( + f""" + PROBLEM: {problem} + PREVIOUS SOLUTIONS: + 1. {solutions[0]['model_name']}: {solutions[0]['solution']} + 2. {solutions[1]['model_name']}: {solutions[1]['solution']} + JUDGE FEEDBACK: {judgment} + INSTRUCTIONS: + - Provide a revised, concise solution in one sentence. + - Include brief reasoning in one additional sentence. + - Address the judge's feedback. + """, + generation_config=genai.types.GenerationConfig( + temperature=0.5, + max_output_tokens=100 + ) + ) + # Handle potential empty response or missing text attribute + try: + # First try to access text directly if available + if hasattr(response, 'text'): + solution = response.text.strip() + # Otherwise check for candidates + elif hasattr(response, 'candidates') and response.candidates: + # Make sure we have candidates and parts before accessing + if hasattr(response.candidates[0], 'content') and hasattr(response.candidates[0].content, 'parts'): + solution = response.candidates[0].content.parts[0].text.strip() + else: + logger.warning(f"Gemini response has candidates but missing content structure: {response}") + solution = "Unable to generate a solution due to API response structure issue." + else: + # Fallback for when candidates is empty + logger.warning(f"Empty response from Gemini API: {response}") + solution = "Unable to generate a solution due to API response issue." + except Exception as e: + logger.error(f"Error extracting text from Gemini response: {e}") + solution = "Unable to generate a solution due to API response issue." + + return {"solution": solution, "model_name": self.model_name} + else: + return {"solution": f"Error: Missing API configuration for {self.api_provider}", "model_name": self.model_name} + + except Exception as e: + logger.error(f"Error in {self.model_name}: {str(e)}") + return {"solution": f"Error: {str(e)}", "model_name": self.model_name} + +# --- Model Registry --- +class ModelRegistry: + @staticmethod + def get_available_models(): + """Get the list of available models grouped by provider (original list)""" + return { + "Anthropic": [ + {"name": "Claude 3.5 Sonnet", "id": "claude-3-5-sonnet-20240620", "provider": "anthropic", "type": ["solver"], "icon": "📜"}, + {"name": "Claude 3.7 Sonnet", "id": "claude-3-7-sonnet-20250219", "provider": "anthropic", "type": ["solver"], "icon": "📜"}, + {"name": "Claude 3 Opus", "id": "claude-3-opus-20240229", "provider": "anthropic", "type": ["solver"], "icon": "📜"}, + {"name": "Claude 3 Haiku", "id": "claude-3-haiku-20240307", "provider": "anthropic", "type": ["solver"], "icon": "📜"} + ], + "OpenAI": [ + {"name": "GPT-4o", "id": "gpt-4o", "provider": "openai", "type": ["solver"], "icon": "🤖"}, + {"name": "GPT-4 Turbo", "id": "gpt-4-turbo", "provider": "openai", "type": ["solver"], "icon": "🤖"}, + {"name": "GPT-4", "id": "gpt-4", "provider": "openai", "type": ["solver"], "icon": "🤖"}, + {"name": "GPT-3.5 Turbo", "id": "gpt-3.5-turbo", "provider": "openai", "type": ["solver"], "icon": "🤖"}, + {"name": "OpenAI o1", "id": "o1", "provider": "openai", "type": ["solver", "judge"], "icon": "🤖"}, + {"name": "OpenAI o3", "id": "o3", "provider": "openai", "type": ["solver", "judge"], "icon": "🤖"} + ], + "Cohere": [ + {"name": "Cohere Command R", "id": "command-r-08-2024", "provider": "cohere", "type": ["solver"], "icon": "💬"}, + {"name": "Cohere Command R+", "id": "command-r-plus-08-2024", "provider": "cohere", "type": ["solver"], "icon": "💬"} + ], + "Google": [ + {"name": "Gemini 1.5 Pro", "id": "gemini-1.5-pro", "provider": "gemini", "type": ["solver"], "icon": "🌟"}, + {"name": "Gemini 2.0 Flash Thinking Experimental 01-21", "id": "gemini-2.0-flash-thinking-exp-01-21", "provider": "gemini", "type": ["solver", "judge"], "icon": "🌟"}, + {"name": "Gemini 2.5 Pro Experimental 03-25", "id": "gemini-2.5-pro-exp-03-25", "provider": "gemini", "type": ["solver", "judge"], "icon": "🌟"} + ], + "HuggingFace": [ + {"name": "Llama 3.3 70B Instruct", "id": "meta-llama/Llama-3.3-70B-Instruct", "provider": "huggingface", "type": ["solver"], "icon": "🔥"}, + {"name": "Llama 3.2 3B Instruct", "id": "meta-llama/Llama-3.2-3B-Instruct", "provider": "huggingface", "type": ["solver"], "icon": "🔥"}, + {"name": "Llama 3.1 70B Instruct", "id": "meta-llama/Llama-3.1-70B-Instruct", "provider": "huggingface", "type": ["solver"], "icon": "🔥"}, + {"name": "Mistral 7B Instruct v0.3", "id": "mistralai/Mistral-7B-Instruct-v0.3", "provider": "huggingface", "type": ["solver"], "icon": "🔥"}, + {"name": "DeepSeek R1 Distill Qwen 32B", "id": "deepseek-ai/DeepSeek-R1-Distill-Qwen-32B", "provider": "huggingface", "type": ["solver", "judge"], "icon": "🔥"}, + {"name": "DeepSeek Coder V2 Instruct", "id": "deepseek-ai/DeepSeek-Coder-V2-Instruct", "provider": "huggingface", "type": ["solver"], "icon": "🔥"}, + {"name": "Qwen 2.5 72B Instruct", "id": "Qwen/Qwen2.5-72B-Instruct", "provider": "huggingface", "type": ["solver"], "icon": "🔥"}, + {"name": "Qwen 2.5 Coder 32B Instruct", "id": "Qwen/Qwen2.5-Coder-32B-Instruct", "provider": "huggingface", "type": ["solver"], "icon": "🔥"}, + {"name": "Qwen 2.5 Math 1.5B Instruct", "id": "Qwen/Qwen2.5-Math-1.5B-Instruct", "provider": "huggingface", "type": ["solver"], "icon": "🔥"}, + {"name": "Gemma 3 27B Instruct", "id": "google/gemma-3-27b-it", "provider": "huggingface", "type": ["solver"], "icon": "🔥"}, + {"name": "Phi-3 Mini 4K Instruct", "id": "microsoft/Phi-3-mini-4k-instruct", "provider": "huggingface", "type": ["solver"], "icon": "🔥"} + ] + } + + @staticmethod + def get_solver_models(): + """Get models suitable for solver role with provider grouping""" + all_models = ModelRegistry.get_available_models() + solver_models = {} + + for provider, models in all_models.items(): + provider_models = [] + for model in models: + if "solver" in model["type"]: + provider_models.append({ + "name": f"{model['icon']} {model['name']} ({provider})", + "id": model["id"], + "provider": model["provider"] + }) + if provider_models: + solver_models[provider] = provider_models + + return solver_models + + @staticmethod + def get_judge_models(): + """Get only specific reasoning models suitable for judge role with provider grouping""" + all_models = ModelRegistry.get_available_models() + judge_models = {} + allowed_judge_models = [ + "Gemini 2.0 Flash Thinking Experimental 01-21 (Google)", + "DeepSeek R1 (HuggingFace)", + "Gemini 2.5 Pro Experimental 03-25 (Google)", + "OpenAI o1 (OpenAI)", + "OpenAI o3 (OpenAI)" + ] + + for provider, models in all_models.items(): + provider_models = [] + for model in models: + full_name = f"{model['name']} ({provider})" + if "judge" in model["type"] and full_name in allowed_judge_models: + provider_models.append({ + "name": f"{model['icon']} {model['name']} ({provider})", + "id": model["id"], + "provider": model["provider"] + }) + if provider_models: + judge_models[provider] = provider_models + + return judge_models + +# --- Orchestrator Class --- +class PolyThinkOrchestrator: + def __init__(self, solver1_config=None, solver2_config=None, judge_config=None, api_clients=None): + self.solvers = [] + self.judge = None + self.api_clients = api_clients or {} + + if solver1_config: + solver1 = PolyThinkAgent( + model_name=solver1_config["name"].split(" ", 1)[1].rsplit(" (", 1)[0] if " " in solver1_config["name"] else solver1_config["name"], + model_path=solver1_config["id"], + api_provider=solver1_config["provider"] + ) + solver1.set_clients(self.api_clients) + self.solvers.append(solver1) + + if solver2_config: + solver2 = PolyThinkAgent( + model_name=solver2_config["name"].split(" ", 1)[1].rsplit(" (", 1)[0] if " " in solver2_config["name"] else solver2_config["name"], + model_path=solver2_config["id"], + api_provider=solver2_config["provider"] + ) + solver2.set_clients(self.api_clients) + self.solvers.append(solver2) + + if judge_config: + self.judge = PolyThinkAgent( + model_name=judge_config["name"].split(" ", 1)[1].rsplit(" (", 1)[0] if " " in judge_config["name"] else judge_config["name"], + model_path=judge_config["id"], + role="judge", + api_provider=judge_config["provider"] + ) + self.judge.set_clients(self.api_clients) + + async def get_initial_solutions(self, problem: str) -> List[Dict[str, Any]]: + tasks = [solver.solve_problem(problem) for solver in self.solvers] + return await asyncio.gather(*tasks) + + async def get_judgment(self, problem: str, solutions: List[Dict[str, Any]]) -> Dict[str, Any]: + if self.judge: + return await self.judge.evaluate_solutions(problem, solutions) + return {"judgment": "No judge configured", "reprompt_needed": False} + + async def get_revised_solutions(self, problem: str, solutions: List[Dict[str, Any]], judgment: str) -> List[Dict[str, Any]]: + tasks = [solver.reprompt_with_context(problem, solutions, judgment) for solver in self.solvers] + return await asyncio.gather(*tasks) + + def generate_final_report(self, problem: str, history: List[Dict[str, Any]]) -> str: + report = f""" +Answer: {answer}
+Reasoning: {reasoning}
+Confidence level: {confidence}
+Models reached AGREEMENT
+Confidence level: {confidence}
+Models could not reach agreement
+Review all solutions above for best answer
+Multi-Agent Problem Solving System
", show_label=False) + + with gr.Row(elem_classes=["header-controls"]): + gr.Button("🌙 Theme", elem_classes=["theme-toggle"], show_label=False) + gr.Button("❓ Help", elem_classes=["help-button"], show_label=False) + + with gr.Row(): + with gr.Column(scale=2): + gr.Markdown("### Problem Input") + problem_input = gr.Textbox( + label="Problem", + placeholder="Enter your problem here...", + lines=5, + elem_classes=["problem-input"] + ) + + with gr.Row(): + max_rounds = gr.Slider( + minimum=1, + maximum=5, + value=3, + step=1, + label="Maximum Rounds", + elem_classes=["rounds-slider"] + ) + solve_btn = gr.Button("Solve", variant="primary", elem_classes=["solve-button"]) + + status_bar = gr.HTML( + value="", + elem_classes=["status-bar"], + visible=False + ) + + with gr.Column(scale=3): + with gr.Tabs() as output_tabs: + with gr.TabItem("Process", elem_classes=["process-tab"]): + initial_solutions = gr.Markdown(visible=False) + round_outputs = [gr.Markdown(visible=False) for _ in range(6)] + + with gr.TabItem("Final Report", elem_classes=["report-tab"]): + final_report = gr.Markdown(visible=False) + + with gr.Row(elem_classes=["action-buttons"], visible=False) as action_buttons: + export_btn = gr.Button("Export Report", elem_classes=["action-button"]) + share_btn = gr.Button("Share Results", elem_classes=["action-button"]) + + def toggle_theme(): + return """ + + """ + + def show_help(): + return """ + + """ + + solve_btn.click( + fn=show_loading_state, + outputs=[ + initial_solutions, + *round_outputs, + final_report, + status_bar, + action_buttons + ] + ).then( + fn=solve_problem, + inputs=[problem_input, max_rounds], + outputs=[ + initial_solutions, + *round_outputs, + final_report, + status_bar, + action_buttons + ] + ) + + export_btn.click( + fn=export_report, + inputs=[problem_input, final_report], + outputs=None + ) + + share_btn.click( + fn=share_results, + inputs=[problem_input, final_report], + outputs=None + ) + + return demo.queue() + +if __name__ == "__main__": + demo = create_polythink_interface() demo.launch(share=True) \ No newline at end of file