Spaces:
Sleeping
Sleeping
File size: 2,306 Bytes
f69f259 b6769aa a04305a b6769aa | 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 | import gradio as gr
import asyncio
import functools
from concurrent.futures import ThreadPoolExecutor
import contextvars
import logging
from model.contextual_response.contextual_response import RAGManager, get_contextual_response
from model.multi_layer_operation_predictor.operation_predictor import get_operation_definition
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
rag_manager = RAGManager()
_executor = ThreadPoolExecutor(max_workers=3)
def to_thread(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
loop = asyncio.get_running_loop()
ctx = contextvars.copy_context()
func_call = functools.partial(ctx.run, func, *args, **kwargs)
return await loop.run_in_executor(_executor, func_call)
return wrapper
async def run_contextual_response(currentLine: str) -> str:
"""Run contextual response in a safe manner."""
try:
return await get_contextual_response(currentLine)
except Exception as e:
logger.error(f"Error in contextual response: {str(e)}")
return ""
def predict_operation(currentLine, language):
"""
Predicts the operation based on user input and selected language
"""
result = get_operation_definition(currentLine, language.lower())
return result if result else "No matching operation found"
iface_operation = gr.Interface(
fn=predict_operation,
inputs=[
gr.Textbox(label="Enter Operation Description"),
gr.Dropdown(choices=["python", "javascript", "typescript", "php", "java"], label="Select Programming Language")
],
outputs=gr.Textbox(label="Predicted Operation"),
title="Operation Prediction",
description="Enter an operation description and select a programming language to get the predicted operation.",
)
iface_contextual = gr.Interface(
fn=lambda question: asyncio.run(run_contextual_response(question)),
inputs=gr.Textbox(label="Enter your question"),
outputs=gr.Textbox(label="Contextual Response"),
title="Contextual Response",
description="Enter a question to get a contextual response.",
)
app = gr.TabbedInterface([iface_operation, iface_contextual], ["Operation Prediction", "Contextual Response"])
if __name__ == "__main__":
app.launch()
|