PyCafe - Solara - GPT-4 Chatbot with OpenAI Streaming

app.py
requirements.txt
app.py
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
121
122
# app.py

import os
# ─── 1) FORCE PRODUCTION MODE ─────────────────────────────────────────────────
# Must come *before* importing solara so that Solara skips its dev‐mode service-workers.
os.environ["SOLARA_MODE"] = "production"

import os
from typing import List, cast
from openai import AsyncOpenAI
from openai.types.chat import ChatCompletionMessageParam
from typing_extensions import TypedDict

import solara
from solara.lab import task, ChatBox, ChatMessage, ChatInput

# ─── 2) MESSAGE TYPE & STATE ───────────────────────────────────────────────────

class MessageDict(TypedDict):
    role: str    # "user" or "assistant"
    content: str

messages: solara.Reactive[List[MessageDict]] = solara.reactive([])

# ─── 3) SECRET LOOKUP ─────────────────────────────────────────────────────────

# Start from any env var you’ve set
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

# Then override with pycafe secret in the server process
try:
    import pycafe
    secret = pycafe.get_secret(
        "OPENAI_API_KEY",
        "Go to https://platform.openai.com/account/api-keys to create one."
    )
    if secret:
        OPENAI_API_KEY = secret
except Exception:
    # if it fails (e.g. in a non-pycafe env), just stick with the env var
    pass

# ─── 4) STREAMING TASK ─────────────────────────────────────────────────────────

# We’ll lazily build the OpenAI client inside the server thread
openai_client: AsyncOpenAI | None = None

def no_api_key_message():
    messages.value = [
        {
            "role": "assistant",
            "content": "⚠️ No OpenAI API key found. "
                       "Please set the OPENAI_API_KEY env var or add it as a Py.cafe secret.",
        },
    ]

@task
async def promt_ai(message: str):
    global openai_client

    # If we haven’t yet, initialize the OpenAI client in this server thread
    if openai_client is None:
        if not OPENAI_API_KEY:
            no_api_key_message()
            return
        openai_client = AsyncOpenAI(api_key=OPENAI_API_KEY)

    # Add the user message
    messages.value = [*messages.value, {"role": "user", "content": message}]

    # Call GPT-4 with streaming
    response = await openai_client.chat.completions.create(
        model="gpt-4-1106-preview",
        messages=cast(List[ChatCompletionMessageParam], messages.value),
        stream=True,
    )

    # Start an empty assistant message so the UI shows “thinking…”
    messages.value = [*messages.value, {"role": "assistant", "content": ""}]

    # Stream in chunks
    async for chunk in response:
        if chunk.choices[0].finish_reason == "stop":  # type: ignore
            break
        delta = chunk.choices[0].delta.content
        if delta:
            updated = {
                "role": "assistant",
                "content": messages.value[-1]["content"] + delta,
            }
            messages.value = [*messages.value[:-1], updated]

# ─── 5) UI COMPONENT ──────────────────────────────────────────────────────────

@solara.component
def Page():
    with solara.Column(style={"width": "100%", "height": "50vh"}):
        # Chat history
        with ChatBox():
            for msg in messages.value:
                with ChatMessage(
                    user=(msg["role"] == "user"),
                    avatar=False,
                    name="ChatGPT" if msg["role"] == "assistant" else "User",
                    color="rgba(0,0,0,0.06)" if msg["role"] == "assistant" else "#ff991f",
                    avatar_background_color=("primary" if msg["role"] == "assistant" else None),
                    border_radius="20px",
                ):
                    solara.Markdown(msg["content"])

        # Loading indicator
        if promt_ai.pending:
            solara.Text("I'm thinking...", style={"font-size": "1rem", "padding-left": "20px"})
            solara.ProgressLinear()

        # Chat input (unique key to preserve typed text)
        ChatInput(
            send_callback=promt_ai,
            disabled_send=promt_ai.pending,
            autofocus=True,
        ).key("chat-input")
Py.Cafe

chatbot

GPT-4 Chatbot with OpenAI Streaming
