Spaces:
Sleeping
Sleeping
| import json | |
| import os | |
| import re | |
| import gradio as gr | |
| from groq import Groq | |
| # ============================================================ | |
| # GROQ CONFIG | |
| # HuggingFace Spaces: Settings → Variables and Secrets | |
| # → New Secret → Name: GROQ_API_KEY → Value: gsk_... | |
| # ============================================================ | |
| GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "") | |
| GROQ_MODEL = "llama-3.1-8b-instant" | |
| _client = None | |
| def get_client(): | |
| global _client | |
| if _client is None: | |
| if not GROQ_API_KEY: | |
| raise ValueError("GROQ_API_KEY not set. Add it in Space Settings → Secrets.") | |
| _client = Groq(api_key=GROQ_API_KEY) | |
| return _client | |
| # ============================================================ | |
| FLAG_EXPLANATIONS = { | |
| "NEG_DEV": "Current score is below the Gold baseline for this checkpoint.", | |
| "REL_COLLAPSE": "Recent performance is dropping relative to earlier checkpoints.", | |
| "STRONG_COLLAPSE": "Performance collapse is severe and sustained.", | |
| "LOW_AND_COLLAPSING": "Player is both below baseline and still getting worse.", | |
| "DEATH_SPIKE": "Deaths increased quickly over the last 6-minute window.", | |
| "UNSUPPORTED_DEATHS": "Many deaths happened without nearby team support or trade value.", | |
| "LOW_IMPACT": "Recent kills + assists are too low during a bad stretch.", | |
| "NO_OBJECTIVE": "No objective participation detected in midgame.", | |
| "LOW_FARM": "Farm gain is stagnating over the recent 6-minute window.", | |
| "NO_VISION": "Vision contribution is too low for jungle/support while already collapsing.", | |
| } | |
| SAMPLE_ROWS = [ | |
| {"minute":6,"role":"JUNGLE","kda":"2/0/3","gold":4800,"cs":42,"score_10":7.2, | |
| "gold_avg":5.0,"delta":2.2,"delta_recent":2.1,"delta_base":2.1,"rel_collapse":0.0, | |
| "risk_score":0,"risk_label":"LOW","flags":[],"deaths_gain_2":0,"kills_gain_2":2, | |
| "assists_gain_2":3,"cs_gain_2":20,"ward_gain_2":1,"unsupported_death_ratio":0.0, | |
| "objective_involvement":0.5,"game_num":1}, | |
| {"minute":12,"role":"JUNGLE","kda":"2/2/4","gold":7200,"cs":71,"score_10":4.1, | |
| "gold_avg":5.1,"delta":-1.0,"delta_recent":-1.4,"delta_base":2.1,"rel_collapse":-3.5, | |
| "risk_score":3,"risk_label":"MED","flags":["REL_COLLAPSE","DEATH_SPIKE(+2)"], | |
| "deaths_gain_2":2,"kills_gain_2":0,"assists_gain_2":1,"cs_gain_2":14,"ward_gain_2":0, | |
| "unsupported_death_ratio":0.5,"objective_involvement":0.2,"game_num":1}, | |
| {"minute":18,"role":"JUNGLE","kda":"2/5/5","gold":9100,"cs":91,"score_10":2.3, | |
| "gold_avg":5.2,"delta":-2.9,"delta_recent":-2.6,"delta_base":0.4,"rel_collapse":-3.0, | |
| "risk_score":5,"risk_label":"HIGH", | |
| "flags":["NEG_DEV","REL_COLLAPSE","LOW_AND_COLLAPSING","DEATH_SPIKE(+3)","NO_OBJECTIVE"], | |
| "deaths_gain_2":3,"kills_gain_2":0,"assists_gain_2":1,"cs_gain_2":10,"ward_gain_2":0, | |
| "unsupported_death_ratio":0.8,"objective_involvement":0.0,"game_num":1}, | |
| ] | |
| # ============================================================ | |
| def clean_flag(flag): | |
| return re.sub(r"\(.*?\)", "", flag).strip() | |
| def parse_flags(flag_input): | |
| if isinstance(flag_input, list): | |
| return [str(x).strip() for x in flag_input if str(x).strip()] | |
| if not flag_input: | |
| return [] | |
| if isinstance(flag_input, str): | |
| return [x.strip() for x in flag_input.split(",") if x.strip()] | |
| return [] | |
| def coerce_number(value, default=0.0): | |
| try: | |
| return float(value) if value not in (None, "") else default | |
| except: | |
| return default | |
| def normalize_row(row): | |
| out = dict(row) | |
| out["minute"] = int(coerce_number(out.get("minute"), 0)) | |
| out["role"] = str(out.get("role", "UNKNOWN")).upper() | |
| out["kda"] = str(out.get("kda", "0/0/0")) | |
| out["gold"] = int(coerce_number(out.get("gold"), 0)) | |
| out["cs"] = int(coerce_number(out.get("cs"), 0)) | |
| out["score_10"] = round(coerce_number(out.get("score_10"), 0.0), 2) | |
| out["gold_avg"] = round(coerce_number(out.get("gold_avg"), 5.0), 2) | |
| out["delta"] = round(coerce_number(out.get("delta"), 0.0), 2) | |
| out["delta_recent"] = round(coerce_number(out.get("delta_recent"), out["delta"]), 2) | |
| out["delta_base"] = round(coerce_number(out.get("delta_base"), out["delta_recent"]), 2) | |
| out["rel_collapse"] = round(coerce_number(out.get("rel_collapse"), 0.0), 2) | |
| out["risk_score"] = int(coerce_number(out.get("risk_score"), 0)) | |
| out["risk_label"] = str(out.get("risk_label", "LOW")).upper() | |
| out["flags"] = parse_flags(out.get("flags", [])) | |
| out["deaths_gain_2"] = int(coerce_number(out.get("deaths_gain_2"), 0)) | |
| out["kills_gain_2"] = int(coerce_number(out.get("kills_gain_2"), 0)) | |
| out["assists_gain_2"] = int(coerce_number(out.get("assists_gain_2"), 0)) | |
| out["cs_gain_2"] = int(coerce_number(out.get("cs_gain_2"), 0)) | |
| out["ward_gain_2"] = int(coerce_number(out.get("ward_gain_2"), 0)) | |
| out["unsupported_death_ratio"] = round(coerce_number(out.get("unsupported_death_ratio"), 0.0), 2) | |
| out["objective_involvement"] = round(coerce_number(out.get("objective_involvement"), 0.0), 2) | |
| return out | |
| def classify_state(row): | |
| flags = {clean_flag(f) for f in row["flags"]} | |
| risk = row["risk_score"] | |
| if risk >= 4 or "STRONG_COLLAPSE" in flags or "LOW_AND_COLLAPSING" in flags: | |
| return "critical_breakdown" | |
| if "NO_OBJECTIVE" in flags: | |
| return "objective_misalignment" | |
| if "UNSUPPORTED_DEATHS" in flags or "DEATH_SPIKE" in flags: | |
| return "unsafe_isolation_pattern" | |
| if "REL_COLLAPSE" in flags or "NEG_DEV" in flags: | |
| return "performance_drop" | |
| if "LOW_FARM" in flags or "NO_VISION" in flags: | |
| return "resource_stagnation" | |
| return "stable" | |
| def build_evidence(row): | |
| evidence = [] | |
| for f in row["flags"]: | |
| base = clean_flag(f) | |
| if base in FLAG_EXPLANATIONS: | |
| evidence.append(f"{f}: {FLAG_EXPLANATIONS[base]}") | |
| if row["delta"] <= -2: | |
| evidence.append(f"delta={row['delta']:.2f}: clearly below Gold checkpoint baseline") | |
| if row["rel_collapse"] <= -1.5: | |
| evidence.append(f"rel_collapse={row['rel_collapse']:.2f}: recent trend worse than earlier game") | |
| if row["unsupported_death_ratio"] >= 0.5: | |
| evidence.append(f"unsupported_death_ratio={row['unsupported_death_ratio']:.2f}: deaths without team support") | |
| if row["objective_involvement"] == 0 and row["minute"] >= 12: | |
| evidence.append("objective_involvement=0 after midgame threshold") | |
| if row["cs_gain_2"] <= 12 and row["role"] != "SUPPORT" and row["minute"] >= 9: | |
| evidence.append(f"cs_gain_2={row['cs_gain_2']}: low 6-minute farm growth") | |
| if not evidence: | |
| evidence.append("No major breakdown signals in current checkpoint.") | |
| return evidence | |
| def llm_suggestion(row, state, evidence): | |
| try: | |
| flags_str = ", ".join(row["flags"]) if row["flags"] else "none" | |
| evidence_str = "\n".join(f"- {e}" for e in evidence) | |
| system = ( | |
| "You are a real-time League of Legends gameplay coach. " | |
| "You monitor a player at 3-minute checkpoints and send short, direct, " | |
| "actionable interventions like a human coach sitting next to the player. " | |
| "Rules: one sentence only, no bullet points, no generic advice, " | |
| "be specific to the role and minute, do not mention trolling or cheating." | |
| ) | |
| user = ( | |
| f"Player role: {row['role']}\n" | |
| f"Game minute: {row['minute']}\n" | |
| f"KDA: {row['kda']} Gold: {row['gold']:,} CS: {row['cs']}\n" | |
| f"Score_10: {row['score_10']:.2f} vs Gold avg: {row['gold_avg']:.2f} " | |
| f"(delta={row['delta']:+.2f})\n" | |
| f"Trend: delta_recent={row['delta_recent']:+.2f} " | |
| f"rel_collapse={row['rel_collapse']:+.2f}\n" | |
| f"Risk: {row['risk_label']} (score={row['risk_score']})\n" | |
| f"Active flags: {flags_str}\n" | |
| f"Evidence:\n{evidence_str}\n" | |
| f"Current state: {state}\n\n" | |
| "Give one short, specific coaching intervention for this player right now." | |
| ) | |
| response = get_client().chat.completions.create( | |
| model=GROQ_MODEL, | |
| messages=[{"role": "system", "content": system}, | |
| {"role": "user", "content": user}], | |
| max_tokens=100, | |
| temperature=0.7, | |
| ) | |
| return state, response.choices[0].message.content.strip() | |
| except Exception as e: | |
| return state, f"[Groq error: {e}] Avoid isolated plays and regroup before the next objective." | |
| def draft_fallback(row, state): | |
| role, minute = row["role"], row["minute"] | |
| if state == "critical_breakdown": | |
| return "mitigation", f"Minute {minute}: {role} is in a high-risk state. Avoid isolated plays, regroup before the next objective." | |
| if state == "objective_misalignment": | |
| return "objective_focus", f"Minute {minute}: reset vision and regroup 20-40s before the next objective." | |
| if state == "unsafe_isolation_pattern": | |
| return "safer_play", f"Minute {minute}: avoid side-lane exposure, wait for a teammate before committing." | |
| if state == "performance_drop": | |
| return "re_engagement", f"Minute {minute}: farm one safe wave, then return to team setup." | |
| if state == "resource_stagnation": | |
| return "resource_recovery", f"Minute {minute}: prioritize safe CS and vision before the next fight." | |
| return "stable", f"Minute {minute}: state is stable. Keep syncing around the next objective." | |
| def analyze_row(row, use_llm=True): | |
| row = normalize_row(row) | |
| state = classify_state(row) | |
| evidence = build_evidence(row) | |
| confidence = min(0.98, max(0.35, 0.35 + 0.08 * row["risk_score"] + 0.06 * len(row["flags"]))) | |
| stype, refined = llm_suggestion(row, state, evidence) if use_llm else draft_fallback(row, state) | |
| return { | |
| "input": row, "state": state, "confidence": round(confidence, 2), | |
| "evidence": evidence, "suggestion_type": stype, | |
| "refined_suggestion": refined, | |
| "source": "groq-llm" if use_llm else "rule-based", | |
| } | |
| def _analyze_list(rows, use_llm): | |
| outputs, log_text = [], [] | |
| for idx, row in enumerate(rows, start=1): | |
| try: | |
| result = analyze_row(row, use_llm=use_llm) | |
| outputs.append({ | |
| "game": row.get("game_num", "?"), | |
| "minute": result["input"]["minute"], | |
| "role": result["input"]["role"], | |
| "kda": result["input"]["kda"], | |
| "score_10": result["input"]["score_10"], | |
| "risk_label": result["input"]["risk_label"], | |
| "state": result["state"], | |
| "source": result["source"], | |
| "suggestion": result["refined_suggestion"], | |
| }) | |
| log_text.append( | |
| f"[Game {row.get('game_num','?')} min {result['input']['minute']}] " | |
| f"{result['input']['risk_label']} → {result['refined_suggestion']}" | |
| ) | |
| except Exception as e: | |
| log_text.append(f"Row {idx} failed: {e}") | |
| return outputs, "\n\n".join(log_text) | |
| def analyze_uploaded_file(file, use_llm): | |
| if file is None: | |
| return [], "No file uploaded." | |
| try: | |
| path = file if isinstance(file, str) else file.name | |
| with open(path, "r") as f: | |
| rows = json.load(f) | |
| if isinstance(rows, dict): | |
| rows = [rows] | |
| except Exception as e: | |
| return [], f"Failed to read file: {e}" | |
| return _analyze_list(rows, use_llm) | |
| # ============================================================ | |
| # GRADIO UI | |
| # ============================================================ | |
| with gr.Blocks(title="PandaSkill Gameplay Agent") as demo: | |
| gr.Markdown(""" | |
| # 🐼 PandaSkill — Gameplay Process Manager Agent | |
| Adapted from **McGee et al. (2026)** — AI facilitator using collective intelligence principles, | |
| applied to League of Legends player performance monitoring. | |
| Upload a `_rows.json` file from `test_player.py` to get per-checkpoint AI coaching interventions. | |
| """) | |
| use_llm_toggle = gr.Checkbox( | |
| value=True, | |
| label="Use Groq LLM (uncheck = rule-based fallback, no API needed)" | |
| ) | |
| with gr.Tab("📂 Upload JSON ← start here"): | |
| gr.Markdown("Upload the `*_rows.json` file exported by `test_player.py` after running locally.") | |
| upload_file = gr.File(label="Upload *_rows.json", file_types=[".json"]) | |
| btn_upload = gr.Button("▶ Run Agent on File", variant="primary") | |
| out_table = gr.JSON(label="All Checkpoint Results") | |
| out_log = gr.Textbox(label="Coaching Suggestions (one per checkpoint)", | |
| lines=25) | |
| btn_upload.click(analyze_uploaded_file, | |
| inputs=[upload_file, use_llm_toggle], | |
| outputs=[out_table, out_log]) | |
| with gr.Tab("🎮 Live Demo"): | |
| gr.Markdown( | |
| "No file needed — runs on a built-in JUNGLE sample showing a " | |
| "performance collapse at minute 12–18." | |
| ) | |
| btn_demo = gr.Button("▶ Run Demo", variant="secondary") | |
| demo_table = gr.JSON(label="Demo Results") | |
| demo_log = gr.Textbox(label="Demo Suggestions", lines=12) | |
| btn_demo.click(_analyze_list, | |
| inputs=[gr.State(SAMPLE_ROWS), use_llm_toggle], | |
| outputs=[demo_table, demo_log]) | |
| with gr.Tab("🔢 Single Snapshot"): | |
| with gr.Row(): | |
| m_min = gr.Number(value=18, label="minute") | |
| m_role = gr.Dropdown(["TOP","JUNGLE","MID","BOT","SUPPORT"], value="JUNGLE", label="role") | |
| m_kda = gr.Textbox(value="2/5/5", label="kda") | |
| m_gold = gr.Number(value=9100, label="gold") | |
| m_cs = gr.Number(value=91, label="cs") | |
| with gr.Row(): | |
| m_s10 = gr.Number(value=2.3, label="score_10") | |
| m_avg = gr.Number(value=5.2, label="gold_avg") | |
| m_dlt = gr.Number(value=-2.9, label="delta") | |
| m_dr = gr.Number(value=-2.6, label="delta_recent") | |
| m_db = gr.Number(value=0.4, label="delta_base") | |
| m_rc = gr.Number(value=-3.0, label="rel_collapse") | |
| with gr.Row(): | |
| m_rs = gr.Number(value=5, label="risk_score") | |
| m_rl = gr.Dropdown(["LOW","MED","HIGH"], value="HIGH", label="risk_label") | |
| m_fl = gr.Textbox(value="NEG_DEV, REL_COLLAPSE, LOW_AND_COLLAPSING, DEATH_SPIKE(+3), NO_OBJECTIVE", label="flags") | |
| with gr.Row(): | |
| m_d2 = gr.Number(value=3, label="deaths_gain_2") | |
| m_k2 = gr.Number(value=0, label="kills_gain_2") | |
| m_a2 = gr.Number(value=1, label="assists_gain_2") | |
| m_cs2 = gr.Number(value=10, label="cs_gain_2") | |
| m_w2 = gr.Number(value=0, label="ward_gain_2") | |
| m_udr = gr.Number(value=0.8, label="unsupported_death_ratio") | |
| m_obj = gr.Number(value=0.0, label="objective_involvement") | |
| btn_m = gr.Button("▶ Analyze", variant="primary") | |
| out_mr = gr.JSON(label="Agent Output") | |
| out_mm = gr.Textbox(label="Coaching Intervention") | |
| def run_manual(minute,role,kda,gold,cs,score_10,gold_avg,delta, | |
| delta_recent,delta_base,rel_collapse,risk_score,risk_label, | |
| flags,deaths_gain_2,kills_gain_2,assists_gain_2,cs_gain_2, | |
| ward_gain_2,unsupported_death_ratio,objective_involvement,use_llm): | |
| row = dict(minute=minute,role=role,kda=kda,gold=gold,cs=cs, | |
| score_10=score_10,gold_avg=gold_avg,delta=delta, | |
| delta_recent=delta_recent,delta_base=delta_base, | |
| rel_collapse=rel_collapse,risk_score=risk_score, | |
| risk_label=risk_label,flags=flags, | |
| deaths_gain_2=deaths_gain_2,kills_gain_2=kills_gain_2, | |
| assists_gain_2=assists_gain_2,cs_gain_2=cs_gain_2, | |
| ward_gain_2=ward_gain_2, | |
| unsupported_death_ratio=unsupported_death_ratio, | |
| objective_involvement=objective_involvement) | |
| result = analyze_row(row, use_llm=use_llm) | |
| return result, result["refined_suggestion"] | |
| btn_m.click(run_manual, | |
| inputs=[m_min,m_role,m_kda,m_gold,m_cs,m_s10,m_avg,m_dlt, | |
| m_dr,m_db,m_rc,m_rs,m_rl,m_fl, | |
| m_d2,m_k2,m_a2,m_cs2,m_w2,m_udr,m_obj,use_llm_toggle], | |
| outputs=[out_mr, out_mm]) | |
| gr.Markdown(""" | |
| --- | |
| **Reference:** McGee, E. S., Cagan, J., & McComb, C. (2026). | |
| Guiding Generalized Team Problem-Solving Through a Collective Intelligence-Based AI Facilitator. | |
| *ASME Journal of Mechanical Design.* | |
| """) | |
| if __name__ == "__main__": | |
| demo.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| share=True, | |
| ssr_mode=False | |
| ) |