| """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 |
| from .articles import MODES, POLICY_URL, article_messages, draft_article |
| 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}] |
|
|
|
|
| def write_article(topic, rows, mode): |
| try: |
| article_messages(topic, rows, mode) |
| if PREVIEW: |
| raise ValueError("The cloud model is not connected yet.") |
| with LOCK: |
| tokenizer, model = engine() |
| rendered, _ = draft_article(tokenizer, model, topic, rows, mode) |
| return rendered, rendered |
| except (ValueError, RuntimeError) as error: |
| raise gr.Error(str(error)) from error |
|
|
|
|
| with gr.Blocks(title=CONFIG["name"]) as demo: |
| gr.Markdown("# PocketVision · Chat, Code & Writing\nTalk, work through bugs, and turn source notes into clear reference writing.") |
| 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.") |
| with gr.Tab("Chat & code"): |
| chatbot = gr.Chatbot(type="messages", label="Conversation", height=360, 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]) |
| with gr.Tab("Articles & research"): |
| gr.Markdown("Draft a short encyclopedia-style article from sources you have read, or prepare an outline for a human editor. " |
| "Paste relevant excerpts; this app does not open URLs or search the web.") |
| topic = gr.Textbox(label="Topic", placeholder="The subject you want to explain") |
| mode = gr.Radio(choices=list(MODES), value=MODES[0], label="Writing task") |
| gr.Markdown(f"For English Wikipedia, use the outline task: its [current guideline]({POLICY_URL}) " |
| "restricts AI-generated article prose. Reference articles here are for other writing contexts.") |
| sources = gr.Dataframe(headers=["Source title", "URL", "Relevant excerpt"], |
| datatype=["str", "str", "str"], row_count=(2, "dynamic"), |
| col_count=(3, "fixed"), type="array", wrap=True, |
| label="Sources", value=[["", "", ""], ["", "", ""]]) |
| write = gr.Button("Prepare draft or outline", variant="primary", interactive=not PREVIEW) |
| draft = gr.Markdown() |
| copy = gr.Textbox(label="Markdown to copy", lines=8, show_copy_button=True, interactive=False) |
| write.click(write_article, [topic, sources, mode], [draft, copy]) |
|
|
|
|
| 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() |
|
|
