| """FastAPI + Gradio entrypoint for Hugging Face Spaces.""" |
|
|
| from __future__ import annotations |
|
|
| import inspect |
|
|
| from fastapi import FastAPI |
|
|
| from gamemaster_copilot.catalog import build_catalog_index, catalog_summary_markdown, get_catalog |
| from gamemaster_copilot.config import MODES, get_settings |
| from gamemaster_copilot.schemas import ChatRequest, ChatResponse |
| from gamemaster_copilot.service import CopilotService |
|
|
|
|
| settings = get_settings() |
| service = CopilotService(settings) |
|
|
| app = FastAPI( |
| title="GameMaster Design Copilot", |
| version="0.1.0", |
| description="CPU-first game design assistant with local RAG and a plugin-friendly JSON API.", |
| ) |
|
|
|
|
| @app.get("/api/health") |
| def health() -> dict: |
| return service.health() |
|
|
|
|
| @app.post("/api/chat", response_model=ChatResponse) |
| def chat(request: ChatRequest) -> ChatResponse: |
| return service.chat(request) |
|
|
|
|
| def _format_response(response: ChatResponse) -> str: |
| parts = [response.answer] |
| if response.citations: |
| citation_lines = [] |
| for citation in response.citations: |
| target = citation.url or "local source" |
| attribution = f", {citation.attribution}" if citation.attribution else "" |
| citation_lines.append( |
| f"- {citation.source_id}: {citation.title} ({citation.license}{attribution}) - {target}" |
| ) |
| parts.append("Citations:\n" + "\n".join(citation_lines)) |
| if response.warnings: |
| parts.append("Warnings:\n" + "\n".join(f"- {warning}" for warning in response.warnings)) |
| return "\n\n".join(parts) |
|
|
|
|
| def _build_gradio_ui(): |
| try: |
| import gradio as gr |
| except Exception: |
| return None |
|
|
| catalog_entries = get_catalog() |
| catalog_choice_labels = [ |
| f"{entry.id} | {entry.label} | {entry.license}" for entry in catalog_entries |
| ] |
| catalog_label_to_id = { |
| f"{entry.id} | {entry.label} | {entry.license}": entry.id for entry in catalog_entries |
| } |
|
|
| def submit(message: str, history: list[dict], project_context: str, mode: str, retrieval_k: int): |
| history = history or [] |
| if not message or not message.strip(): |
| return "", history |
| response = service.chat( |
| ChatRequest( |
| message=message, |
| project_context=project_context or "", |
| mode=mode, |
| retrieval_k=int(retrieval_k), |
| ) |
| ) |
| history = history + [ |
| {"role": "user", "content": message}, |
| {"role": "assistant", "content": _format_response(response)}, |
| ] |
| return "", history |
|
|
| def refresh_index_status(): |
| return service.health() |
|
|
| def run_legal_scraper(selected_labels: list[str], max_docs_per_source: int): |
| selected_labels = selected_labels or [] |
| selected_ids = [catalog_label_to_id[label] for label in selected_labels if label in catalog_label_to_id] |
| if not selected_ids: |
| return "Select at least one catalog source before scraping.", service.health() |
|
|
| try: |
| manifest = build_catalog_index( |
| selected_ids=selected_ids, |
| index_dir=settings.index_dir, |
| embedding_backend=settings.embedding_backend, |
| embedding_model=settings.embedding_model, |
| embedding_dimensions=settings.embedding_dimensions, |
| max_docs_per_source=int(max_docs_per_source), |
| ) |
| health = service.reload_index() |
| except Exception as exc: |
| return f"Scrape failed: `{exc}`", service.health() |
|
|
| warning_text = "" |
| if manifest.get("warnings"): |
| warning_text = "\n\nWarnings:\n" + "\n".join(f"- {warning}" for warning in manifest["warnings"][:20]) |
| if len(manifest["warnings"]) > 20: |
| warning_text += f"\n- ... {len(manifest['warnings']) - 20} more warnings" |
|
|
| summary = ( |
| f"Built legal catalog index with **{manifest['scraped_document_count']} documents** " |
| f"and **{manifest['chunk_count']} chunks** from **{len(selected_ids)} catalog sources**.\n\n" |
| f"Embedding backend: `{manifest['embedding_backend']}`\n\n" |
| f"The chat retrieval layer has been reloaded. Current index chunks: `{health['chunk_count']}`." |
| f"{warning_text}" |
| ) |
| return summary, manifest |
|
|
| with gr.Blocks(title="GameMaster Design Copilot") as demo: |
| gr.HTML( |
| """ |
| <style> |
| .gm-hero { |
| border-radius: 24px; |
| padding: 24px; |
| background: |
| radial-gradient(circle at 20% 20%, rgba(124, 187, 122, 0.35), transparent 32%), |
| linear-gradient(135deg, #17211b 0%, #293f34 48%, #0f1714 100%); |
| color: #f4f0e6; |
| } |
| .gm-hero h1 { margin-bottom: 6px; } |
| </style> |
| <div class="gm-hero"> |
| <h1>GameMaster Design Copilot</h1> |
| <p>CPU-first design assistant for mechanics, level flow, balance critique, and live GM prep.</p> |
| </div> |
| """ |
| ) |
| with gr.Tabs(): |
| with gr.Tab("Copilot"): |
| with gr.Row(): |
| with gr.Column(scale=3): |
| chatbot_kwargs = {"label": "Design conversation", "height": 520} |
| if "type" in inspect.signature(gr.Chatbot).parameters: |
| chatbot_kwargs["type"] = "messages" |
| chatbot = gr.Chatbot(**chatbot_kwargs) |
| message = gr.Textbox( |
| label="Design request", |
| placeholder="Example: Critique this stamina system for dominant strategies...", |
| lines=3, |
| ) |
| with gr.Row(): |
| send = gr.Button("Send", variant="primary") |
| clear = gr.Button("Clear") |
| with gr.Column(scale=1): |
| mode = gr.Dropdown( |
| choices=list(MODES.keys()), |
| value="brainstorm", |
| label="Mode", |
| ) |
| retrieval_k = gr.Slider( |
| minimum=0, |
| maximum=10, |
| value=4, |
| step=1, |
| label="Retrieved chunks", |
| ) |
| project_context = gr.Textbox( |
| label="Project context", |
| placeholder="Genre, target audience, platform, rules constraints, design goals...", |
| lines=14, |
| ) |
| gr.Markdown( |
| "Grounded answers require approved sources and a built RAG index. " |
| "Use the Legal Scraper tab to build one from the curated catalog." |
| ) |
|
|
| send.click( |
| submit, |
| inputs=[message, chatbot, project_context, mode, retrieval_k], |
| outputs=[message, chatbot], |
| ) |
| message.submit( |
| submit, |
| inputs=[message, chatbot, project_context, mode, retrieval_k], |
| outputs=[message, chatbot], |
| ) |
| clear.click(lambda: [], outputs=[chatbot]) |
|
|
| with gr.Tab("Legal Scraper"): |
| gr.Markdown( |
| "This scraper is intentionally allowlist-only. It does not crawl arbitrary sites, " |
| "does not ingest unclear copyrighted material, skips binary assets, checks robots.txt, " |
| "and preserves license/attribution metadata in every chunk." |
| ) |
| gr.Markdown(catalog_summary_markdown()) |
| source_picker = gr.CheckboxGroup( |
| choices=catalog_choice_labels, |
| value=catalog_choice_labels, |
| label="Catalog sources to scrape", |
| ) |
| max_docs = gr.Slider( |
| minimum=1, |
| maximum=80, |
| value=20, |
| step=1, |
| label="Max documents per source", |
| info="Use a lower number for faster CPU Space runs; raise it for deeper crawling.", |
| ) |
| with gr.Row(): |
| scrape_button = gr.Button("Scrape Selected Sources And Rebuild Index", variant="primary") |
| status_button = gr.Button("Refresh Index Status") |
| scrape_status = gr.Markdown() |
| index_manifest = gr.JSON(label="Index manifest / status") |
|
|
| scrape_button.click( |
| run_legal_scraper, |
| inputs=[source_picker, max_docs], |
| outputs=[scrape_status, index_manifest], |
| ) |
| status_button.click(refresh_index_status, outputs=[index_manifest]) |
| return demo |
|
|
|
|
| demo = _build_gradio_ui() |
| if demo is not None: |
| import gradio as gr |
|
|
| app = gr.mount_gradio_app(app, demo, path="/") |
|
|
