Upload app.py with huggingface_hub

app.py

@@ -0,0 +1,329 @@
+
+import gradio as gr
+from app import demo as app
+import os
+
+_docs = {'WordleBoard': {'description': 'Interactive Wordle board component.', 'members': {'__init__': {'word_length': {'type': 'int', 'default': '5', 'description': None}, 'max_attempts': {'type': 'int', 'default': '6', 'description': None}, 'return': {'type': 'None', 'description': None}}, 'postprocess': {'value': {'type': 'typing.Union[\n    gradio_wordleboard.wordleboard.PublicWordleState,\n    typing.Dict,\n    str,\n    NoneType,\n][PublicWordleState, Dict, str, None]', 'description': None}}, 'preprocess': {'return': {'type': 'typing.Optional[typing.Dict][Dict, None]', 'description': "The preprocessed input data sent to the user's function in the backend."}, 'value': None}}, 'events': {}}, '__meta__': {'additional_interfaces': {'PublicWordleState': {'source': '@dataclass\nclass PublicWordleState:\n    board: List[WordleRow]\n    current_row: int\n    status: str\n    message: str\n    max_rows: int', 'refs': ['WordleRow']}, 'WordleRow': {'source': '@dataclass\nclass WordleRow:\n    letters: List[str] = field(\n        default_factory=lambda: [""] * 5\n    )\n    statuses: List[TileStatus] = field(\n        default_factory=lambda: ["empty"] * 5\n    )'}}, 'user_fn_refs': {'WordleBoard': ['PublicWordleState']}}}
+
+abs_path = os.path.join(os.path.dirname(__file__), "css.css")
+
+with gr.Blocks(
+    css=abs_path,
+    theme=gr.themes.Default(
+        font_mono=[
+            gr.themes.GoogleFont("Inconsolata"),
+            "monospace",
+        ],
+    ),
+) as demo:
+    gr.Markdown(
+"""
+# `gradio_wordleboard`
+
+<div style="display: flex; gap: 7px;">
+<img alt="Static Badge" src="https://img.shields.io/badge/version%20-%200.0.1%20-%20orange">  
+</div>
+
+A custom Gradio component that renders and plays the Wordle word game
+""", elem_classes=["md-custom"], header_links=True)
+    app.render()
+    gr.Markdown(
+"""
+## Installation
+
+```bash
+pip install gradio_wordleboard
+```
+
+## Usage
+
+```python
+
+from __future__ import annotations
+
+import asyncio
+import os
+import re
+from typing import AsyncIterator, Dict, List
+
+import gradio as gr
+from gradio_wordleboard import WordleBoard
+from openai import AsyncOpenAI
+
+from envs.textarena_env import TextArenaAction, TextArenaEnv
+from envs.textarena_env.models import TextArenaMessage
+
+
+API_BASE_URL = os.getenv("API_BASE_URL", "https://router.huggingface.co/v1")
+API_KEY = os.getenv("API_KEY") or os.getenv("HF_TOKEN")
+MODEL = os.getenv("MODEL", "openai/gpt-oss-120b:novita")
+MAX_TURNS = int(os.getenv("MAX_TURNS", "6"))
+DOCKER_IMAGE = os.getenv("TEXTARENA_IMAGE", "textarena-env:latest")
+
+
+def _format_history(messages: List[TextArenaMessage]) -> str:
+    lines: List[str] = []
+    for message in messages:
+        tag = message.category or "MESSAGE"
+        lines.append(f"[{tag}] {message.content}")
+    return "\n".join(lines)
+
+
+def _make_user_prompt(prompt_text: str, messages: List[TextArenaMessage]) -> str:
+    history = _format_history(messages)
+    return (
+        f"Current prompt:\n{prompt_text}\n\n"
+        f"Conversation so far:\n{history}\n\n"
+        "Reply with your next guess enclosed in square brackets."
+    )
+
+
+async def _generate_guesses(client: AsyncOpenAI, prompt: str, history: List[TextArenaMessage]) -> str:
+    response = await client.chat.completions.create(
+        model=MODEL,
+        messages=[
+            {
+                "role": "system",
+                "content": (
+                    "You are an expert Wordle solver."
+                    " Always respond with a single guess inside square brackets, e.g. [crane]."
+                    " Use lowercase letters, exactly one five-letter word per reply."
+                    " Reason about prior feedback before choosing the next guess."
+                    " Words must be 5 letters long and real English words."
+                    " Do not include any other text in your response."
+                    " Do not repeat the same guess twice."
+                ),
+            },
+            {"role": "user", "content": _make_user_prompt(prompt, history)},
+        ],
+        max_tokens=64,
+        temperature=0.7,
+    )
+
+    content = response.choices[0].message.content
+    response_text = content.strip() if content else ""
+    print(f"Response text: {response_text}")
+    return response_text
+
+
+async def _play_wordle(env: TextArenaEnv, client: AsyncOpenAI) -> AsyncIterator[Dict[str, str]]:
+    state = await asyncio.to_thread(env.reset)
+    observation = state.observation
+
+    for turn in range(1, MAX_TURNS + 1):
+        if state.done:
+            break
+
+        model_output = await _generate_guesses(client, observation.prompt, observation.messages)
+        guess = _extract_guess(model_output)
+
+        state = await asyncio.to_thread(env.step, TextArenaAction(message=guess))
+        observation = state.observation
+
+        feedback = _collect_feedback(observation.messages)
+        yield {"guess": guess, "feedback": feedback}
+
+    yield {
+        "guess": "",
+        "feedback": _collect_feedback(observation.messages),
+    }
+
+
+def _extract_guess(text: str) -> str:
+    if not text:
+        return "[crane]"
+
+    match = re.search(r"\[([A-Za-z]{5})\]", text)
+    if match:
+        guess = match.group(1).lower()
+        return f"[{guess}]"
+
+    cleaned = re.sub(r"[^a-zA-Z]", "", text).lower()
+    if len(cleaned) >= 5:
+        return f"[{cleaned[:5]}]"
+
+    return "[crane]"
+
+
+def _collect_feedback(messages: List[TextArenaMessage]) -> str:
+    parts: List[str] = []
+    for message in messages:
+        tag = message.category or "MESSAGE"
+        if tag.upper() in {"FEEDBACK", "SYSTEM", "MESSAGE"}:
+            parts.append(message.content.strip())
+    return "\n".join(parts).strip()
+
+
+async def inference_handler(api_key: str) -> AsyncIterator[str]:
+    if not api_key:
+        raise RuntimeError("HF_TOKEN or API_KEY environment variable must be set.")
+
+    client = AsyncOpenAI(base_url=API_BASE_URL, api_key=api_key)
+    env = TextArenaEnv.from_docker_image(
+        DOCKER_IMAGE,
+        env_vars={
+            "TEXTARENA_ENV_ID": "Wordle-v0",
+            "TEXTARENA_NUM_PLAYERS": "1",
+        },
+        ports={8000: 8000},
+    )
+
+    try:
+        async for result in _play_wordle(env, client):
+            yield result["feedback"]
+    finally:
+        env.close()
+
+
+wordle_component = WordleBoard()
+
+
+async def run_inference() -> AsyncIterator[Dict]:
+    feedback_history: List[str] = []
+
+    async for feedback in inference_handler(API_KEY):
+        stripped = feedback.strip()
+        if not stripped:
+            continue
+
+        feedback_history.append(stripped)
+        combined_feedback = "\n\n".join(feedback_history)
+        state = wordle_component.parse_feedback(combined_feedback)
+        yield wordle_component.to_public_dict(state)
+
+    if not feedback_history:
+        yield wordle_component.to_public_dict(wordle_component.create_game_state())
+
+
+with gr.Blocks() as demo:
+    gr.Markdown("# Wordle TextArena Inference Demo")
+
+    board = WordleBoard(value=wordle_component.to_public_dict(wordle_component.create_game_state()))
+    run_button = gr.Button("Run Inference", variant="primary")
+
+    run_button.click(
+        fn=run_inference,
+        inputs=None,
+        outputs=board,
+        show_progress=True,
+        api_name="run",
+    )
+
+demo.queue()
+
+
+if __name__ == "__main__":
+    if not API_KEY:
+        raise SystemExit("HF_TOKEN (or API_KEY) must be set to query the model.")
+
+    demo.launch()
+
+```
+""", elem_classes=["md-custom"], header_links=True)
+
+
+    gr.Markdown("""
+## `WordleBoard`
+
+### Initialization
+""", elem_classes=["md-custom"], header_links=True)
+
+    gr.ParamViewer(value=_docs["WordleBoard"]["members"]["__init__"], linkify=['PublicWordleState', 'WordleRow'])
+
+
+
+
+    gr.Markdown("""
+
+### User function
+
+The impact on the users predict function varies depending on whether the component is used as an input or output for an event (or both).
+
+- When used as an Input, the component only impacts the input signature of the user function.
+- When used as an output, the component only impacts the return signature of the user function.
+
+The code snippet below is accurate in cases where the component is used as both an input and an output.
+
+- **As input:** Is passed, the preprocessed input data sent to the user's function in the backend.
+
+
+ ```python
+def predict(
+    value: typing.Optional[typing.Dict][Dict, None]
+) -> typing.Union[
+    gradio_wordleboard.wordleboard.PublicWordleState,
+    typing.Dict,
+    str,
+    NoneType,
+][PublicWordleState, Dict, str, None]:
+    return value
+```
+""", elem_classes=["md-custom", "WordleBoard-user-fn"], header_links=True)
+
+
+
+
+    code_PublicWordleState = gr.Markdown("""
+## `PublicWordleState`
+```python
+@dataclass
+class PublicWordleState:
+    board: List[WordleRow]
+    current_row: int
+    status: str
+    message: str
+    max_rows: int
+```""", elem_classes=["md-custom", "PublicWordleState"], header_links=True)
+
+    code_WordleRow = gr.Markdown("""
+## `WordleRow`
+```python
+@dataclass
+class WordleRow:
+    letters: List[str] = field(
+        default_factory=lambda: [""] * 5
+    )
+    statuses: List[TileStatus] = field(
+        default_factory=lambda: ["empty"] * 5
+    )
+```""", elem_classes=["md-custom", "WordleRow"], header_links=True)
+
+    demo.load(None, js=r"""function() {
+    const refs = {
+            PublicWordleState: ['WordleRow'], 
+            WordleRow: [], };
+    const user_fn_refs = {
+          WordleBoard: ['PublicWordleState'], };
+    requestAnimationFrame(() => {
+
+        Object.entries(user_fn_refs).forEach(([key, refs]) => {
+            if (refs.length > 0) {
+                const el = document.querySelector(`.${key}-user-fn`);
+                if (!el) return;
+                refs.forEach(ref => {
+                    el.innerHTML = el.innerHTML.replace(
+                        new RegExp("\\b"+ref+"\\b", "g"),
+                        `<a href="#h-${ref.toLowerCase()}">${ref}</a>`
+                    );
+                })
+            }
+        })
+
+        Object.entries(refs).forEach(([key, refs]) => {
+            if (refs.length > 0) {
+                const el = document.querySelector(`.${key}`);
+                if (!el) return;
+                refs.forEach(ref => {
+                    el.innerHTML = el.innerHTML.replace(
+                        new RegExp("\\b"+ref+"\\b", "g"),
+                        `<a href="#h-${ref.toLowerCase()}">${ref}</a>`
+                    );
+                })
+            }
+        })
+    })
+}
+
+""")
+
+demo.launch()