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 )