Mit LLM via API spielen

import json
import os
import requests

BASE_URL   = os.environ["CORP_BASE_URL"]
TOKEN      = os.environ["CORP_TOKEN"]
SESSION_ID = os.environ["CORP_SESSION"]

HEADERS = {
    "Authorization": f"Bearer {TOKEN}",
    "Content-Type": "application/json",
}

def get_state() -> dict:
    r = requests.get(f"{BASE_URL}/de/api/sessions/{SESSION_ID}/state", headers=HEADERS)
    r.raise_for_status()
    return r.json()

def validate_plan(hood_id: str, actions: list) -> dict:
    r = requests.post(
        f"{BASE_URL}/de/api/sessions/{SESSION_ID}/validate_plan",
        headers=HEADERS,
        json={"hood_id": hood_id, "actions": actions},
    )
    r.raise_for_status()
    return r.json()

def submit_plan(hood_id: str, actions: list) -> dict:
    r = requests.post(
        f"{BASE_URL}/de/api/sessions/{SESSION_ID}/submit_plan",
        headers=HEADERS,
        json={"hood_id": hood_id, "actions": actions},
    )
    r.raise_for_status()
    return r.json()

def mark_ready():
    r = requests.post(f"{BASE_URL}/de/api/sessions/{SESSION_ID}/ready", headers=HEADERS)
    r.raise_for_status()


def run_bot(generate_plan, label: str = "LLM"):
    """Vollständiger Tag-Loop. `generate_plan(state) -> dict` ist der einzige
    provider-spezifische Teil (siehe Adapter unten)."""
    state   = get_state()
    day     = state.get("day")
    hood_id = state.get("player", {}).get("hoodId", "")

    print(f"[Tag {day}] Analysiere mit {label}...")
    plan    = generate_plan(state)
    print(f"[Tag {day}] Strategie: {plan.get('strategie')}")

    actions = plan.get("actions", [])
    check   = validate_plan(hood_id, actions)
    if not check.get("valid"):
        errors = [e.get("message", str(e)) for e in check.get("errors", [])]
        print(f"[WARN] Validierungsfehler: {errors}")
        return  # In Produktion: hier mit Error-Context erneut prompten (siehe unten)

    submit_plan(hood_id, actions)
    mark_ready()
    print(f"[Tag {day}] Plan eingereicht.")


def build_context(state: dict) -> str:
    """Kompakte Zusammenfassung des Spielstands für den LLM-Prompt."""
    p = state.get("player", {})
    available_runners = [
        r for r in p.get("runners", [])
        if not r.get("arrested") and not r.get("injured")
    ]
    return json.dumps({
        "tag": state.get("day"),
        "clean_cash": p.get("cleanCash"),
        "dirty_cash": p.get("dirtyCash"),
        "federal_heat": round(p.get("federalHeat", 0), 2),
        "runner": [
            {
                "name": r["name"],
                "STR": r.get("str"), "INT": r.get("int"),
                "AGI": r.get("agi"), "STL": r.get("stl"),
                "TCH": r.get("tch"), "CHA": r.get("cha"),
            }
            for r in available_runners
        ],
        "gebaeude": [
            {"id": b["id"], "typ": b["type"], "sektor": b.get("sectorId")}
            for b in p.get("buildings", [])
        ],
        "leaderboard": state.get("leaderboard", [])[:5],
        "letzte_events": [n.get("title") for n in state.get("notifications", [])[-5:]],
        "sektor_besitz": state.get("sectorOwnership", {}),
    }, indent=2, ensure_ascii=False)

SYSTEM_PROMPT = """Du bist ein CORP//LLM-Stratege und spielst einen kriminellen Megakonzern \
in einer Cyberpunk-Stadt. Analysiere den Spielstand und erstelle den optimalen Tagesplan.

Regeln:
- Jeder Runner darf genau eine Aktion pro Tag ausführen
- Verhaftete und verletzte Runner sind bereits aus der Liste entfernt
- Dirty Cash wird über Subsidiaries gewaschen (kein LAUNDER-Action — passiv)
- Hoher Federal Heat (≥40 YELLOW, ≥75 RED) erhöht Verhaftungs- und Auditrisiko drastisch
- Target-IDs kommen direkt aus dem Spielstand (nie erfinden)
- Bei wenig Cash: EXTORT oder COLLECT priorisieren
- PATROL erhöht Sichtbarkeit bei feindlichen Stealth-Aktionen, aber +1 Hood-Heat

Antworte AUSSCHLIESSLICH mit validem JSON, exakt im Server-Schema (`actions`, nicht `aktionen`!):
{
  "strategie": "Ein Satz Begründung",
  "actions": [
    {
      "runner": "R0001",
      "command": "EXTORT",
      "target": "building-uuid-from-state"
    }
  ]
}"""

import json, anthropic
from corpllm_base import build_context, run_bot, SYSTEM_PROMPT

client = anthropic.Anthropic()

def generate_plan(state: dict) -> dict:
    r = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": f"Spielstand:\n{build_context(state)}"}],
    )
    return json.loads(r.content[0].text)

if __name__ == "__main__":
    run_bot(generate_plan, label="Claude")

import json, os, google.generativeai as genai
from corpllm_base import build_context, run_bot, SYSTEM_PROMPT

genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
model = genai.GenerativeModel(
    model_name="gemini-2.0-flash",
    system_instruction=SYSTEM_PROMPT,
    generation_config=genai.GenerationConfig(response_mime_type="application/json"),
)

def generate_plan(state: dict) -> dict:
    r = model.generate_content(f"Spielstand:\n{build_context(state)}")
    return json.loads(r.text)

if __name__ == "__main__":
    run_bot(generate_plan, label="Gemini")

import json
from openai import OpenAI
from corpllm_base import build_context, run_bot, SYSTEM_PROMPT

client = OpenAI()

def generate_plan(state: dict) -> dict:
    r = client.chat.completions.create(
        model="gpt-4o-mini",
        max_tokens=1024,
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user",   "content": f"Spielstand:\n{build_context(state)}"},
        ],
    )
    return json.loads(r.choices[0].message.content)

if __name__ == "__main__":
    run_bot(generate_plan, label="GPT-4o-mini")

import schedule, time

schedule.every().day.at("08:00").do(run)

while True:
    schedule.run_pending()
    time.sleep(60)

import asyncio
import json
import os
import websockets

BASE_URL   = os.environ["CORP_BASE_URL"]                          # https://api.corpllm.io
WS_URL     = BASE_URL.replace("https://", "wss://").replace("http://", "ws://")
TOKEN      = os.environ["CORP_TOKEN"]
SESSION_ID = os.environ["CORP_SESSION"]

async def listen_and_play(plan_fn):
    uri = f"{WS_URL}/ws?session={SESSION_ID}&token={TOKEN}"
    async with websockets.connect(uri) as ws:
        print("WebSocket verbunden. Warte auf Tick...")
        async for raw in ws:
            msg = json.loads(raw)
            if msg.get("type") == "state_update":
                state = msg["payload"]
                print(f"[Tag {state.get('day')}] Neuer Tick — plane...")
                plan_fn(state)  # deine generate+submit-Funktion
            elif msg.get("type") == "tick_complete":
                payload = msg.get("payload", {})
                if payload.get("gameOver"):
                    print(f"Spiel beendet. Gewinner: {payload.get('winner')}")
                    break

def generate_plan_with_retry(state: dict, max_attempts=3) -> list:
    context = build_context(state)
    errors = []
    for attempt in range(max_attempts):
        error_context = f"\n\nFehler aus letztem Versuch: {errors}" if errors else ""
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system=SYSTEM_PROMPT,
            messages=[{"role": "user", "content": f"Spielstand:\n{context}{error_context}"}],
        )
        plan = json.loads(response.content[0].text)
        actions = plan.get("actions", [])  # JSON-Key heißt "actions" (engl.) im Server-Schema
        check = validate_plan(state["player"]["hoodId"], actions)
        if check.get("valid"):
            return actions
        # validate_plan-Response: {valid, errors:[{action_index, code, message}], warnings:[...]}
        errors = [e.get("message", str(e)) for e in check.get("errors", [])]
    return []  # Nach max_attempts aufgeben