File size: 2,564 Bytes
5ba1bb4 | 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 | """Gradio chat with code formatting and no image inputs."""
from functools import lru_cache
import os
import threading
from .runtime import CONFIG, answer, gpu_available, load_engine
import gradio as gr
LOCK = threading.Lock()
PREVIEW = os.environ.get("POCKETCHAT_PREVIEW") == "1" or not gpu_available()
ADAPTER = os.environ.get("POCKETCHAT_ADAPTER")
@lru_cache(maxsize=1)
def engine():
return load_engine(ADAPTER)
def respond(message, history):
if not message or not message.strip():
raise gr.Error("Enter a message first.")
if PREVIEW:
raise gr.Error("The cloud model is not connected yet.")
try:
with LOCK:
tokenizer, model = engine()
reply = answer(tokenizer, model, message, history)
except (ValueError, RuntimeError) as error:
raise gr.Error(str(error)) from error
return "", (history or []) + [
{"role": "user", "content": message}, {"role": "assistant", "content": reply}]
with gr.Blocks(title=CONFIG["name"]) as demo:
gr.Markdown("# PocketVision · Chat & Code\nA place to talk, learn, write code, and work through bugs.")
if PREVIEW:
gr.Markdown("**Interface preview** · The cloud model is not connected yet.")
elif ADAPTER:
gr.Markdown("Custom chat and programming model · Review generated code before using it.")
else:
gr.Markdown("Pretrained model · Custom chat and programming training is pending.")
chatbot = gr.Chatbot(type="messages", label="Conversation", height=460, allow_tags=False)
message = gr.Textbox(label="Your message", lines=3, max_lines=14,
placeholder="Ask a question, paste code, or describe a bug…", interactive=not PREVIEW)
with gr.Row():
send = gr.Button("Send", variant="primary", interactive=not PREVIEW)
clear = gr.Button("New chat")
gr.Examples([
"Write a Python function that removes duplicates while preserving their order.",
"Explain JavaScript async and await with a small example.",
"Help me debug my code. I’ll paste it in the next message.",
], inputs=message)
for event in [send.click, message.submit]:
event(respond, [message, chatbot], [message, chatbot])
clear.click(lambda: ("", []), outputs=[message, chatbot])
def launch():
demo.queue(default_concurrency_limit=1).launch(
server_name="0.0.0.0" if os.environ.get("SPACE_ID") else "127.0.0.1",
server_port=int(os.environ.get("PORT", "7860")), share=False)
if __name__ == "__main__":
launch()
|