feat(main): finish backend

app.py

@@ -0,0 +1,7 @@
+from open_cortex.ui.app import build_app
+
+demo = build_app()
+
+if __name__ == "__main__":
+    demo.queue()
+    demo.launch()

open_cortex/ui/app.py

@@ -0,0 +1,138 @@
+import gradio as gr
+from open_cortex.ui.gradio_history import history_to_chat_messages
+from open_cortex.runtime.client import stream_chat_events
+
+
+def format_runtime(event) -> str:
+    if event.kind == "request_started":
+        return "Phase: request started\nWaiting for first token..."
+
+    if event.kind == "first_token" and event.snapshot is not None:
+        snapshot = event.snapshot
+        context_tokens = (
+            snapshot.slot_context_tokens[0]
+            if snapshot.slot_context_tokens
+            else None
+        )
+
+        return "\n".join(
+            [
+                "Phase: first token",
+                f"TTFT: {event.ttft_ms:.1f} ms",
+                f"Context: {context_tokens} / {snapshot.slot_context_size}",
+                f"Token Stream: {snapshot.decode_tps} tok/s",
+                (
+                    "Engine: "
+                    f"processing={snapshot.requests_processing} "
+                    f"deferred={snapshot.requests_deferred}"
+                ),
+            ]
+        )
+
+    if event.kind == "request_completed":
+        return "\n".join(
+            [
+                "Phase: completed",
+                f"Prompt tokens: {event.prompt_tokens}",
+                f"Output tokens: {event.completion_tokens}",
+                f"Prefill: {event.prompt_tps:.1f} tok/s",
+                f"Decode: {event.decode_tps:.1f} tok/s",
+            ]
+        )
+
+    return "Phase: decoding"
+
+
+
+def user(user_message: str, history: list[dict]) -> tuple[str, list[dict]]:
+    if not user_message.strip():
+        return "", history
+
+    return "", history + [
+        {
+            "role": "user",
+            "content": user_message,
+        }
+    ]
+
+def bot(history: list):
+    messages = history_to_chat_messages(history)
+
+    history.append(
+        {
+            "role": "assistant",
+            "content": "",
+        }
+    )
+
+    runtime_text = "Phase: idle"
+
+    for event in stream_chat_events(messages):
+        runtime_text = format_runtime(event)
+
+        if event.text_delta:
+            history[-1]["content"] += event.text_delta
+
+        yield history, runtime_text
+
+
+
+def build_app() -> gr.Blocks:
+    with gr.Blocks() as demo:
+        gr.Markdown("# OpenCortex Minimal Chat")
+
+        with gr.Row():
+            with gr.Column(scale=2):
+                chatbot = gr.Chatbot(
+                    height=480,
+                )
+                msg = gr.Textbox(
+                    placeholder="Ask the local model...",
+                    show_label=False,
+                )
+                clear = gr.Button("Clear")
+
+            with gr.Column(scale=1):
+                runtime = gr.Textbox(
+                    label="Runtime",
+                    value="Phase: idle",
+                    lines=10,
+                    interactive=False,
+                )
+
+
+        msg.submit(
+            user,
+            [msg, chatbot],
+            [msg, chatbot],
+            queue=False,
+        ).then(
+            bot,
+            chatbot,
+            [chatbot, runtime],
+        )
+
+        clear.click(
+            lambda: ([], "Phase: idle"),
+            None,
+            [chatbot, runtime],
+            queue=False,
+        )
+    return demo
+
+
+def main() -> None:
+    demo = build_app()
+    demo.queue()
+    demo.launch()
+
+
+
+if __name__ == "__main__":
+    main()
+
+
+
+
+
+

pyproject.toml

@@ -21,3 +21,6 @@ dev = [
 [[tool.uv.index]]
 url = "https://pypi.tuna.tsinghua.edu.cn/simple"
 default = true
+
+[project.scripts]
+open-cortex = "open_cortex.ui.app:main"