Spaces:
Sleeping
Sleeping
File size: 3,849 Bytes
28fff71 f7933d3 63ab561 e44659b 28fff71 63ab561 f7933d3 63ab561 28fff71 63ab561 28fff71 63ab561 e44659b 63ab561 e44659b a05579f 63ab561 28fff71 63ab561 a05579f 63ab561 a05579f 63ab561 28fff71 63ab561 a05579f 63ab561 28fff71 63ab561 28fff71 63ab561 28fff71 63ab561 f7933d3 63ab561 f7933d3 63ab561 f7933d3 63ab561 28fff71 f7933d3 28fff71 f7933d3 |
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 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
import concurrent.futures
import gradio as gr
from huggingface_hub import InferenceClient
import json
import traceback
import tempfile
import shutil
class HuggingChatExecutor:
def __init__(self):
self.temp_dir = tempfile.TemporaryDirectory()
self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
self.client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
def _execute_chat(self, message, system_prompt, functions=None):
messages = [{"role": "system", "content": system_prompt}]
messages.append({"role": "user", "content": message})
try:
if functions:
# Parse functions string as JSON array
functions_list = json.loads(functions)
response = self.client.chat_completion(
messages,
functions=functions_list,
stream=False
)
else:
response = self.client.chat_completion(
messages,
stream=False
)
# Format the output
output = {
"message": message,
"system_prompt": system_prompt,
"functions": functions if functions else "No functions provided",
"response": response.choices[0].message
}
return json.dumps(output, indent=2)
except Exception as e:
error_trace = traceback.format_exc()
return f"Error executing chat:\n{str(e)}\n\nTraceback:\n{error_trace}"
def execute(self, message, system_prompt, functions):
future = self.executor.submit(self._execute_chat, message, system_prompt, functions)
try:
output = future.result(timeout=30)
return output
except concurrent.futures.TimeoutError:
return "Error: Timed out waiting for chat response"
except Exception as e:
return f"Error: {str(e)}"
def __del__(self):
self.executor.shutdown(wait=False)
shutil.rmtree(self.temp_dir.name)
def wrapper_execute(message, system_prompt, functions):
executor = HuggingChatExecutor()
return executor.execute(message, system_prompt, functions)
def create_interface():
with gr.Blocks() as demo:
gr.Markdown("# HuggingChat Tool Executor")
gr.Markdown("Execute chat completions with function calling capabilities")
gr.Markdown("### Message")
message_input = gr.Textbox(
label="User Message",
placeholder="Enter your message here...",
lines=3
)
gr.Markdown("### System Prompt")
system_input = gr.Textbox(
label="System Prompt",
value="You are a helpful AI assistant.",
lines=2
)
gr.Markdown("### Functions")
functions_input = gr.Textbox(
label="Functions (JSON array)",
placeholder='''[
{
"name": "get_weather",
"description": "Get the weather in a location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA"
}
},
"required": ["location"]
}
}
]''',
lines=10
)
gr.Markdown("### Output")
output_text = gr.Textbox(label="Response", lines=20)
run_button = gr.Button("Run")
run_button.click(
wrapper_execute,
inputs=[message_input, system_input, functions_input],
outputs=output_text
)
return demo
if __name__ == "__main__":
demo = create_interface()
demo.launch()
|