Spaces:
Running
Running
| 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() | |