| import os |
| import random |
| from typing import Dict, List |
|
|
| import google.generativeai as genai |
| import gradio as gr |
| import openai |
| from anthropic import Anthropic |
| from openai import OpenAI |
|
|
|
|
| def get_all_models(): |
| """Get all available models from the registries.""" |
| return [ |
| "SambaNova: Meta-Llama-3.2-1B-Instruct", |
| "SambaNova: Meta-Llama-3.2-3B-Instruct", |
| "SambaNova: Llama-3.2-11B-Vision-Instruct", |
| "SambaNova: Llama-3.2-90B-Vision-Instruct", |
| "SambaNova: Meta-Llama-3.1-8B-Instruct", |
| "SambaNova: Meta-Llama-3.1-70B-Instruct", |
| "SambaNova: Meta-Llama-3.1-405B-Instruct", |
| "Hyperbolic: Qwen/Qwen2.5-Coder-32B-Instruct", |
| "Hyperbolic: meta-llama/Llama-3.2-3B-Instruct", |
| "Hyperbolic: meta-llama/Meta-Llama-3.1-8B-Instruct", |
| "Hyperbolic: meta-llama/Meta-Llama-3.1-70B-Instruct", |
| "Hyperbolic: meta-llama/Meta-Llama-3-70B-Instruct", |
| "Hyperbolic: NousResearch/Hermes-3-Llama-3.1-70B", |
| "Hyperbolic: Qwen/Qwen2.5-72B-Instruct", |
| "Hyperbolic: deepseek-ai/DeepSeek-V2.5", |
| "Hyperbolic: meta-llama/Meta-Llama-3.1-405B-Instruct", |
| ] |
|
|
|
|
| def generate_discussion_prompt(original_question: str, previous_responses: List[str]) -> str: |
| """Generate a prompt for models to discuss and build upon previous |
| responses.""" |
| prompt = f"""You are participating in a multi-AI discussion about this question: "{original_question}" |
| |
| Previous responses from other AI models: |
| {chr(10).join(f"- {response}" for response in previous_responses)} |
| |
| Please provide your perspective while: |
| 1. Acknowledging key insights from previous responses |
| 2. Adding any missing important points |
| 3. Respectfully noting if you disagree with anything and explaining why |
| 4. Building towards a complete answer |
| |
| Keep your response focused and concise (max 3-4 paragraphs).""" |
| return prompt |
|
|
|
|
| def generate_consensus_prompt(original_question: str, discussion_history: List[str]) -> str: |
| """Generate a prompt for final consensus building.""" |
| return f"""Review this multi-AI discussion about: "{original_question}" |
| |
| Discussion history: |
| {chr(10).join(discussion_history)} |
| |
| As a final synthesizer, please: |
| 1. Identify the key points where all models agreed |
| 2. Explain how any disagreements were resolved |
| 3. Present a clear, unified answer that represents our collective best understanding |
| 4. Note any remaining uncertainties or caveats |
| |
| Keep the final consensus concise but complete.""" |
|
|
|
|
| def chat_with_openai(model: str, messages: List[Dict], api_key: str | None) -> str: |
| import openai |
|
|
| client = openai.OpenAI(api_key=api_key) |
| response = client.chat.completions.create(model=model, messages=messages) |
| return response.choices[0].message.content |
|
|
|
|
| def chat_with_anthropic(messages: List[Dict], api_key: str | None) -> str: |
| """Chat with Anthropic's Claude model.""" |
| client = Anthropic(api_key=api_key) |
| response = client.messages.create(model="claude-3-sonnet-20240229", messages=messages, max_tokens=1024) |
| return response.content[0].text |
|
|
|
|
| def chat_with_gemini(messages: List[Dict], api_key: str | None) -> str: |
| """Chat with Gemini Pro model.""" |
| genai.configure(api_key=api_key) |
| model = genai.GenerativeModel("gemini-pro") |
|
|
| |
| gemini_messages = [] |
| for msg in messages: |
| role = "user" if msg["role"] == "user" else "model" |
| gemini_messages.append({"role": role, "parts": [msg["content"]]}) |
|
|
| response = model.generate_content([m["parts"][0] for m in gemini_messages]) |
| return response.text |
|
|
|
|
| def chat_with_sambanova( |
| messages: List[Dict], api_key: str | None, model_name: str = "Llama-3.2-90B-Vision-Instruct" |
| ) -> str: |
| """Chat with SambaNova's models using their OpenAI-compatible API.""" |
| client = openai.OpenAI( |
| api_key=api_key, |
| base_url="https://api.sambanova.ai/v1", |
| ) |
|
|
| response = client.chat.completions.create( |
| model=model_name, messages=messages, temperature=0.1, top_p=0.1 |
| ) |
| return response.choices[0].message.content |
|
|
|
|
| def chat_with_hyperbolic( |
| messages: List[Dict], api_key: str | None, model_name: str = "Qwen/Qwen2.5-Coder-32B-Instruct" |
| ) -> str: |
| """Chat with Hyperbolic's models using their OpenAI-compatible API.""" |
| client = OpenAI(api_key=api_key, base_url="https://api.hyperbolic.xyz/v1") |
|
|
| |
| full_messages = [ |
| {"role": "system", "content": "You are a helpful assistant. Be descriptive and clear."}, |
| *messages, |
| ] |
|
|
| response = client.chat.completions.create( |
| model=model_name, |
| messages=full_messages, |
| temperature=0.7, |
| max_tokens=1024, |
| ) |
| return response.choices[0].message.content |
|
|
|
|
| def multi_model_consensus( |
| question: str, selected_models: List[str], rounds: int = 3, progress: gr.Progress = gr.Progress() |
| ) -> list[tuple[str, str]]: |
| if not selected_models: |
| raise gr.Error("Please select at least one model to chat with.") |
|
|
| chat_history = [] |
| discussion_history = [] |
|
|
| |
| progress(0, desc="Getting initial responses...") |
| initial_responses = [] |
| for i, model in enumerate(selected_models): |
| provider, model_name = model.split(": ", 1) |
|
|
| try: |
| if provider == "Anthropic": |
| api_key = os.getenv("ANTHROPIC_API_KEY") |
| response = chat_with_anthropic(messages=[{"role": "user", "content": question}], api_key=api_key) |
| elif provider == "SambaNova": |
| api_key = os.getenv("SAMBANOVA_API_KEY") |
| response = chat_with_sambanova( |
| messages=[ |
| {"role": "system", "content": "You are a helpful assistant"}, |
| {"role": "user", "content": question}, |
| ], |
| api_key=api_key, |
| ) |
| elif provider == "Hyperbolic": |
| api_key = os.getenv("HYPERBOLIC_API_KEY") |
| response = chat_with_hyperbolic(messages=[{"role": "user", "content": question}], api_key=api_key) |
| else: |
| api_key = os.getenv("GEMINI_API_KEY") |
| response = chat_with_gemini(messages=[{"role": "user", "content": question}], api_key=api_key) |
|
|
| initial_responses.append(f"{model}: {response}") |
| discussion_history.append(f"Initial response from {model}:\n{response}") |
| chat_history.append((f"Initial response from {model}", response)) |
| except Exception as e: |
| chat_history.append((f"Error from {model}", str(e))) |
|
|
| |
| for round_num in range(rounds): |
| progress((round_num + 1) / (rounds + 2), desc=f"Discussion round {round_num + 1}...") |
| round_responses = [] |
|
|
| random.shuffle(selected_models) |
| for model in selected_models: |
| provider, model_name = model.split(": ", 1) |
|
|
| try: |
| discussion_prompt = generate_discussion_prompt(question, discussion_history) |
| if provider == "Anthropic": |
| api_key = os.getenv("ANTHROPIC_API_KEY") |
| response = chat_with_anthropic( |
| messages=[{"role": "user", "content": discussion_prompt}], api_key=api_key |
| ) |
| elif provider == "SambaNova": |
| api_key = os.getenv("SAMBANOVA_API_KEY") |
| response = chat_with_sambanova( |
| messages=[ |
| {"role": "system", "content": "You are a helpful assistant"}, |
| {"role": "user", "content": discussion_prompt}, |
| ], |
| api_key=api_key, |
| ) |
| elif provider == "Hyperbolic": |
| api_key = os.getenv("HYPERBOLIC_API_KEY") |
| response = chat_with_hyperbolic( |
| messages=[{"role": "user", "content": discussion_prompt}], api_key=api_key |
| ) |
| else: |
| api_key = os.getenv("GEMINI_API_KEY") |
| response = chat_with_gemini( |
| messages=[{"role": "user", "content": discussion_prompt}], api_key=api_key |
| ) |
|
|
| round_responses.append(f"{model}: {response}") |
| discussion_history.append(f"Round {round_num + 1} - {model}:\n{response}") |
| chat_history.append((f"Round {round_num + 1} - {model}", response)) |
| except Exception as e: |
| chat_history.append((f"Error from {model} in round {round_num + 1}", str(e))) |
|
|
| |
| progress(0.9, desc="Building final consensus...") |
| model = selected_models[0] |
| provider, model_name = model.split(": ", 1) |
|
|
| try: |
| consensus_prompt = generate_consensus_prompt(question, discussion_history) |
| if provider == "Anthropic": |
| api_key = os.getenv("ANTHROPIC_API_KEY") |
| final_consensus = chat_with_anthropic( |
| messages=[{"role": "user", "content": consensus_prompt}], api_key=api_key |
| ) |
| elif provider == "SambaNova": |
| api_key = os.getenv("SAMBANOVA_API_KEY") |
| final_consensus = chat_with_sambanova( |
| messages=[ |
| {"role": "system", "content": "You are a helpful assistant"}, |
| {"role": "user", "content": consensus_prompt}, |
| ], |
| api_key=api_key, |
| ) |
| elif provider == "Hyperbolic": |
| api_key = os.getenv("HYPERBOLIC_API_KEY") |
| final_consensus = chat_with_hyperbolic( |
| messages=[{"role": "user", "content": consensus_prompt}], api_key=api_key |
| ) |
| else: |
| api_key = os.getenv("GEMINI_API_KEY") |
| final_consensus = chat_with_gemini( |
| messages=[{"role": "user", "content": consensus_prompt}], api_key=api_key |
| ) |
| except Exception as e: |
| final_consensus = f"Error getting consensus from {model}: {str(e)}" |
|
|
| chat_history.append(("Final Consensus", final_consensus)) |
|
|
| progress(1.0, desc="Done!") |
| return chat_history |
|
|
|
|
| with gr.Blocks() as demo: |
| gr.Markdown("# Experimental Multi-Model Consensus Chat") |
| gr.Markdown( |
| """Select multiple models to collaborate on answering your question. |
| The models will discuss with each other and attempt to reach a consensus. |
| Maximum 3 models can be selected at once.""" |
| ) |
|
|
| with gr.Row(): |
| with gr.Column(): |
| model_selector = gr.Dropdown( |
| choices=get_all_models(), |
| multiselect=True, |
| label="Select Models (max 3)", |
| info="Choose up to 3 models to participate in the discussion", |
| value=["SambaNova: Llama-3.2-90B-Vision-Instruct", "Hyperbolic: Qwen/Qwen2.5-Coder-32B-Instruct"], |
| max_choices=3, |
| ) |
| rounds_slider = gr.Slider( |
| minimum=1, |
| maximum=2, |
| value=1, |
| step=1, |
| label="Discussion Rounds", |
| info="Number of rounds of discussion between models", |
| ) |
|
|
| chatbot = gr.Chatbot(height=600, label="Multi-Model Discussion") |
| msg = gr.Textbox(label="Your Question", placeholder="Ask a question for the models to discuss...") |
|
|
| def respond(message, selected_models, rounds): |
| chat_history = multi_model_consensus(message, selected_models, rounds) |
| return chat_history |
|
|
| msg.submit(respond, [msg, model_selector, rounds_slider], [chatbot], api_name="consensus_chat") |
|
|
| for fn in demo.fns.values(): |
| fn.api_name = False |
|
|
| if __name__ == "__main__": |
| demo.launch() |
|
|