| """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() |
|
|
