"""
Simple BigCodeArena - A simplified AI coding battle arena
Focuses on core functionality: two models, automatic code extraction, and execution
"""
import gradio as gr
from gradio_sandboxcomponent import SandboxComponent
import pandas as pd
import datetime
import os
import asyncio
import concurrent.futures
import time
from datasets import Dataset, load_dataset
# Import completion utilities
from completion import make_config, registered_api_completion
from sandbox.prompts import GENERAL_SANDBOX_INSTRUCTION
# Import code extraction utilities
from sandbox.code_analyzer import (
SandboxEnvironment,
extract_code_from_markdown,
)
# Import sandbox execution functions
from sandbox.code_runner import (
run_html_sandbox,
run_react_sandbox,
run_vue_sandbox,
run_pygame_sandbox,
run_gradio_sandbox,
run_streamlit_sandbox,
run_code_interpreter,
run_c_code,
run_cpp_code,
run_java_code,
run_golang_code,
run_rust_code,
mermaid_to_html,
javascript_to_html
)
# Import sandbox telemetry
from sandbox.sandbox_telemetry import log_sandbox_telemetry_gradio_fn
# Create a proper sandbox state structure
def create_sandbox_state() -> dict:
"""Create a new sandbox state for a model"""
return {
'enable_sandbox': True,
'enabled_round': 0,
'sandbox_run_round': 0,
'edit_round': 0,
'sandbox_environment': SandboxEnvironment.AUTO,
'auto_selected_sandbox_environment': None,
'sandbox_instruction': "Run the extracted code in the appropriate sandbox environment",
'code_to_execute': "",
'code_dependencies': ([], []),
'btn_list_length': 5,
'sandbox_id': None,
'chat_session_id': None,
'conv_id': None,
"sandbox_output": None,
"sandbox_error": None,
}
def reset_sandbox_state(state: dict) -> dict:
"""Reset the sandbox state"""
state['enabled_round'] = 0
state['sandbox_run_round'] = 0
state['edit_round'] = 0
state['auto_selected_sandbox_environment'] = None
state['code_to_execute'] = ""
state['code_dependencies'] = ([], [])
state['sandbox_error'] = None
state['sandbox_output'] = None
state['sandbox_id'] = None
state['conv_id'] = None
state['chat_session_id'] = None
return state
# Load API configuration
def load_api_config():
"""Load API configuration from yaml file"""
try:
config = make_config("api_config.yaml")
return config
except Exception as e:
return {}
# Global variables
api_config = load_api_config()
available_models = list(api_config.keys()) if api_config else []
# HuggingFace dataset configuration
HF_DATASET_NAME = os.getenv("HF_DATASET_NAME")
HF_TOKEN = os.getenv("HF_TOKEN")
# Global ranking data cache
ranking_data = None
ranking_last_updated = None
def get_random_models():
"""Get two random models from available models"""
if len(available_models) < 2:
return available_models[0] if available_models else None, available_models[0] if available_models else None
import random
models = random.sample(available_models, 2)
return models[0], models[1]
def create_chat_state(model_name: str) -> dict:
"""Create a new chat state for a model"""
return {
"model_name": model_name,
"messages": [],
"sandbox_state": create_sandbox_state(),
"has_output": False,
"generating": False, # Track if model is currently generating
"interactions": [], # Store user interactions
}
def generate_response_with_completion(state, temperature, max_tokens):
"""Generate response using the completion API system with full conversation history"""
if state is None:
return state, ""
# Get the last user message
user_message = None
for msg in reversed(state["messages"]):
if msg["role"] == "user":
user_message = msg["content"]
break
if not user_message:
return state, ""
# Prepare messages for API call - include full conversation history
messages = [{"role": "system", "content": GENERAL_SANDBOX_INSTRUCTION}]
for msg in state["messages"]:
if msg["role"] in ["user", "assistant"] and msg["content"] is not None:
messages.append({"role": msg["role"], "content": msg["content"]})
# Get model config
model_name = state["model_name"]
if model_name not in api_config:
return state, f"Error: Model {model_name} not configured"
model_config = api_config[model_name]
api_type = model_config.get("api_type", "openai")
# retrieve the api completion function from register
api_completion_func = registered_api_completion[api_type]
# build arguments for api completions
# Use the actual model identifier from config, not the display name
actual_model = model_config.get("model", model_name)
kwargs = {
"model": actual_model,
"temperature": temperature,
"max_tokens": max_tokens,
"api_dict": model_config.get("endpoints", [{}])[0] if model_config.get("endpoints") else None,
"messages": messages,
}
output = api_completion_func(**kwargs)
# Extract the answer from the response
if isinstance(output, dict) and "answer" in output:
response_text = output["answer"]
# Return response as dict with content and interaction keys
response_dict = {
"content": response_text,
"interaction": state.get("interactions", [])
}
return state, response_dict
else:
error_msg = f"Error: Invalid response format from {api_type}"
# Return error as dict with content and interaction keys
error_dict = {
"content": error_msg,
"interaction": state.get("interactions", [])
}
return state, error_dict
def generate_response_async(state, temperature, max_tokens):
"""Async wrapper for generate_response_with_completion"""
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
# Run the synchronous function in a thread pool
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(generate_response_with_completion, state, temperature, max_tokens)
return future.result()
finally:
loop.close()
async def generate_responses_parallel(state0, state1, temperature, max_tokens):
"""Generate responses for both models in parallel with error handling"""
loop = asyncio.get_event_loop()
# Run both model generations in parallel
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
future0 = loop.run_in_executor(executor, generate_response_with_completion, state0, temperature, max_tokens)
future1 = loop.run_in_executor(executor, generate_response_with_completion, state1, temperature, max_tokens)
# Wait for both to complete with error handling
try:
result0, result1 = await asyncio.gather(future0, future1, return_exceptions=True)
# Handle exceptions
if isinstance(result0, Exception):
result0 = (state0, {"content": f"Error: {str(result0)}", "interaction": []})
if isinstance(result1, Exception):
result1 = (state1, {"content": f"Error: {str(result1)}", "interaction": []})
except Exception as e:
# Fallback to sequential processing
result0 = generate_response_with_completion(state0, temperature, max_tokens)
result1 = generate_response_with_completion(state1, temperature, max_tokens)
return result0, result1
def extract_and_execute_code(message, sandbox_state):
"""Extract code from message and prepare for execution"""
if not message:
return sandbox_state, "", ""
# Extract code using the same logic as code_runner.py
extract_result = extract_code_from_markdown(
message=message,
enable_auto_env=True
)
if extract_result is None:
return sandbox_state, "", ""
code, code_language, env_selection, install_command = extract_result
# Update sandbox state (now a dictionary)
sandbox_state['code_to_execute'] = code
sandbox_state['install_command'] = install_command
sandbox_state['auto_selected_sandbox_environment'] = env_selection
return sandbox_state, code, str(env_selection)
def add_text_and_generate(state0, state1, text, temperature, max_tokens, model_a, model_b):
"""Add text and generate responses for both models"""
if not text.strip():
return state0, state1, "", "", "", "", "", "", "", "", "", "", "", ""
# Initialize states if needed
if state0 is None or state1 is None:
if state0 is None:
state0 = create_chat_state(model_a)
if state1 is None:
state1 = create_chat_state(model_b)
# Add user message to both states
state0["messages"].append({"role": "user", "content": text})
state1["messages"].append({"role": "user", "content": text})
# Mark that generation is starting - this will be used to hide vote buttons
state0["generating"] = True
state1["generating"] = True
# Generate responses in parallel
start_time = time.time()
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result0, result1 = loop.run_until_complete(
generate_responses_parallel(state0, state1, temperature, max_tokens)
)
state0, response0 = result0
state1, response1 = result1
generation_time = time.time() - start_time
except Exception as e:
# Fallback to sequential processing
state0, response0 = generate_response_with_completion(state0, temperature, max_tokens)
state1, response1 = generate_response_with_completion(state1, temperature, max_tokens)
generation_time = time.time() - start_time
finally:
loop.close()
# Add the assistant responses to the message history
state0["messages"].append({"role": "assistant", "content": response0["content"]})
state1["messages"].append({"role": "assistant", "content": response1["content"]})
# Format chat history for display
chat0 = format_chat_history(state0["messages"])
chat1 = format_chat_history(state1["messages"])
# Extract code from responses for sandbox
sandbox_state0 = (
state0.get("sandbox_state", create_sandbox_state())
if state0
else create_sandbox_state()
)
sandbox_state1 = (
state1.get("sandbox_state", create_sandbox_state())
if state1
else create_sandbox_state()
)
sandbox_state0, code0, env0 = extract_and_execute_code(response0["content"], sandbox_state0)
sandbox_state1, code1, env1 = extract_and_execute_code(response1["content"], sandbox_state1)
# Update sandbox states in the main states
if state0 is not None:
state0["sandbox_state"] = sandbox_state0
state0["has_output"] = True
state0["generating"] = False # Mark generation as complete
if state1 is not None:
state1["sandbox_state"] = sandbox_state1
state1["has_output"] = True
state1["generating"] = False # Mark generation as complete
# Clear previous sandbox outputs when new message is sent
sandbox_output0 = ""
sandbox_output1 = ""
# Force clear sandbox components to ensure refresh
sandbox_component_update0 = gr.update(value=("", False, []), visible=False)
sandbox_component_update1 = gr.update(value=("", False, []), visible=False)
# Also clear the sandbox view components to show fresh results
sandbox_view_a = ""
sandbox_view_b = ""
# Run sandbox executions in parallel if both models have code
if code0.strip() or code1.strip():
sandbox_start_time = time.time()
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
# Prepare sandbox execution parameters
install_command0 = sandbox_state0.get('install_command', "") if code0.strip() else ""
install_command1 = sandbox_state1.get('install_command', "") if code1.strip() else ""
# Run both sandbox executions in parallel
result0, result1 = loop.run_until_complete(
run_sandboxes_parallel(
sandbox_state0, code0, install_command0,
sandbox_state1, code1, install_command1
)
)
sandbox_time = time.time() - sandbox_start_time
# Process results for model A
if code0.strip():
sandbox_url0, sandbox_output0, sandbox_error0 = result0
# Check if this is a web-based environment that should use SandboxComponent
env_type = sandbox_state0.get('auto_selected_sandbox_environment') or sandbox_state0.get('sandbox_environment')
# Use the URL directly from the function return
if sandbox_url0:
# Force refresh by using a unique key and clearing first
sandbox_component_update0 = gr.update(
value=(sandbox_url0, True, []),
visible=True,
key=f"sandbox_a_{int(time.time() * 1000)}" # Unique key to force refresh
)
# Update sandbox view with output and errors
if sandbox_output0:
sandbox_view_a += sandbox_output0
if sandbox_error0:
sandbox_view_a = f"🚨 Errors/Warnings
\n\n```\n{sandbox_error0}\n```\n\n🚨 Errors/Warnings
\n\n```\n{sandbox_error1}\n```\n\n🚨 Errors/Warnings
\n\n```\n{sandbox_error0}\n```\n\n🚨 Errors/Warnings
\n\n```\n{sandbox_error1}\n```\n\n🚨 Errors/Warnings
\n\n```\n{sandbox_error0.strip()}\n```\n\n🚨 Errors/Warnings
\n\n```\n{sandbox_error1.strip()}\n```\n\n