Eine Coding-Implementierung zum Entwerfen eines KI-Governance-Programs für Unternehmen unter Verwendung von OpenClaw Gateway-Richtlinien-Engines, Genehmigungsworkflows und überprüfbarer Agentenausführung

!apt-get replace -y
!apt-get set up -y curl
!curl -fsSL https://deb.nodesource.com/setup_22.x | bash -
!apt-get set up -y nodejs
!node -v
!npm -v
!npm set up -g openclaw@newest
!pip -q set up requests pandas pydantic


import os
import json
import time
import uuid
import secrets and techniques
import subprocess
import getpass
from pathlib import Path
from typing import Dict, Any
from dataclasses import dataclass, asdict
from datetime import datetime, timezone


import requests
import pandas as pd
from pydantic import BaseModel, Subject


strive:
   from google.colab import userdata
   OPENAI_API_KEY = userdata.get("OPENAI_API_KEY")
besides Exception:
   OPENAI_API_KEY = None


if not OPENAI_API_KEY:
   OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")


if not OPENAI_API_KEY:
   OPENAI_API_KEY = getpass.getpass("Enter your OpenAI API key (hidden enter): ").strip()


assert OPENAI_API_KEY != "", "API key can't be empty."


OPENCLAW_HOME = Path("/root/.openclaw")
OPENCLAW_HOME.mkdir(mother and father=True, exist_ok=True)
WORKSPACE = OPENCLAW_HOME / "workspace"
WORKSPACE.mkdir(mother and father=True, exist_ok=True)


GATEWAY_TOKEN = secrets and techniques.token_urlsafe(48)
GATEWAY_PORT = 18789
GATEWAY_URL = f"http://127.0.0.1:{GATEWAY_PORT}"

config = {
   "env": {
       "OPENAI_API_KEY": OPENAI_API_KEY
   },
   "brokers": {
       "defaults": {
           "workspace": str(WORKSPACE),
           "mannequin": {
               "major": "openai/gpt-4.1-mini"
           }
       }
   },
   "gateway": {
       "mode": "native",
       "port": GATEWAY_PORT,
       "bind": "loopback",
       "auth": {
           "mode": "token",
           "token": GATEWAY_TOKEN
       },
       "http": {
           "endpoints": {
               "chatCompletions": {
                   "enabled": True
               }
           }
       }
   }
}


config_path = OPENCLAW_HOME / "openclaw.json"
config_path.write_text(json.dumps(config, indent=2))


physician = subprocess.run(
   ("bash", "-lc", "openclaw physician --fix --yes"),
   capture_output=True,
   textual content=True
)
print(physician.stdout(-2000:))
print(physician.stderr(-2000:))


gateway_log = "/tmp/openclaw_gateway.log"
gateway_cmd = f"OPENAI_API_KEY='{OPENAI_API_KEY}' OPENCLAW_GATEWAY_TOKEN='{GATEWAY_TOKEN}' openclaw gateway --port {GATEWAY_PORT} --bind loopback --token '{GATEWAY_TOKEN}' --verbose > {gateway_log} 2>&1 & echo $!"
gateway_pid = subprocess.check_output(("bash", "-lc", gateway_cmd)).decode().strip()
print("Gateway PID:", gateway_pid)

def wait_for_gateway(timeout=120):
   begin = time.time()
   whereas time.time() - begin < timeout:
       strive:
           r = requests.get(f"{GATEWAY_URL}/", timeout=5)
           if r.status_code in (200, 401, 403, 404):
               return True
       besides Exception:
           move
       time.sleep(2)
   return False


assert wait_for_gateway(), Path(gateway_log).read_text()(-6000:)


headers = {
   "Authorization": f"Bearer {GATEWAY_TOKEN}",
   "Content material-Kind": "software/json"
}


def openclaw_chat(messages, consumer="demo-user", agent_id="major", temperature=0.2):
   payload = {
       "mannequin": f"openclaw:{agent_id}",
       "messages": messages,
       "consumer": consumer,
       "temperature": temperature,
       "stream": False
   }
   r = requests.submit(
       f"{GATEWAY_URL}/v1/chat/completions",
       headers=headers,
       json=payload,
       timeout=180
   )
   r.raise_for_status()
   return r.json()


class ActionProposal(BaseModel):
   user_request: str
   class: str
   danger: str
   confidence: float = Subject(ge=0.0, le=1.0)
   requires_approval: bool
   permit: bool
   cause: str

def classify_request(user_request: str) -> ActionProposal:
   textual content = user_request.decrease()


   red_terms = (
       "delete", "take away completely", "wire cash", "switch funds",
       "payroll", "financial institution", "hr report", "worker report", "run shell",
       "execute command", "api key", "secret", "credential", "token",
       "ssh", "sudo", "wipe", "exfiltrate", "add personal", "database dump"
   )
   amber_terms = (
       "electronic mail", "ship", "notify", "buyer", "vendor", "contract",
       "bill", "price range", "approve", "safety coverage", "confidential",
       "write file", "modify", "change"
   )


   if any(t in textual content for t in red_terms):
       return ActionProposal(
           user_request=user_request,
           class="high_impact",
           danger="purple",
           confidence=0.92,
           requires_approval=True,
           permit=False,
           cause="Excessive-impact or delicate motion detected"
       )


   if any(t in textual content for t in amber_terms):
       return ActionProposal(
           user_request=user_request,
           class="moderate_impact",
           danger="amber",
           confidence=0.76,
           requires_approval=True,
           permit=True,
           cause="Reasonable-risk motion requires human approval earlier than execution"
       )


   return ActionProposal(
       user_request=user_request,
       class="low_impact",
       danger="inexperienced",
       confidence=0.88,
       requires_approval=False,
       permit=True,
       cause="Low-risk request"
   )


def simulated_human_approval(proposal: ActionProposal) -> Dict(str, Any):
   if proposal.danger == "purple":
       authorized = False
       word = "Rejected mechanically in demo for red-risk request"
   elif proposal.danger == "amber":
       authorized = True
       word = "Authorized mechanically in demo for amber-risk request"
   else:
       authorized = True
       word = "No approval required"
   return {
       "authorized": authorized,
       "reviewer": "simulated_manager",
       "word": word
   }


@dataclass
class TraceEvent:
   trace_id: str
   ts: str
   stage: str
   payload: Dict(str, Any)

class TraceStore:
   def __init__(self, path="openclaw_traces.jsonl"):
       self.path = path
       Path(self.path).write_text("")


   def append(self, occasion: TraceEvent):
       with open(self.path, "a") as f:
           f.write(json.dumps(asdict(occasion)) + "n")


   def read_all(self):
       rows = ()
       with open(self.path, "r") as f:
           for line in f:
               line = line.strip()
               if line:
                   rows.append(json.hundreds(line))
       return rows


trace_store = TraceStore()


def now():
   return datetime.now(timezone.utc).isoformat()


SYSTEM_PROMPT = """
You're an enterprise OpenClaw assistant working underneath governance controls.


Guidelines:
- By no means declare an motion has been executed until the governance layer explicitly permits it.
- For low-risk requests, reply usually and helpfully.
- For moderate-risk requests, suggest a protected plan and point out any approvals or checks that will be wanted.
- For top-risk requests, refuse to execute and as an alternative present a safer non-operational various corresponding to a draft, guidelines, abstract, or evaluate plan.
- Be concise however helpful.
"""


def governed_openclaw_run(user_request: str, session_user: str = "employee-001") -> Dict(str, Any):
   trace_id = str(uuid.uuid4())


   proposal = classify_request(user_request)
   trace_store.append(TraceEvent(trace_id, now(), "classification", proposal.model_dump()))


   approval = None
   if proposal.requires_approval:
       approval = simulated_human_approval(proposal)
       trace_store.append(TraceEvent(trace_id, now(), "approval", approval))


   if proposal.danger == "purple":
       consequence = {
           "trace_id": trace_id,
           "standing": "blocked",
           "proposal": proposal.model_dump(),
           "approval": approval,
           "response": "This request is blocked by governance coverage. I will help by drafting a protected plan, a guidelines, or an approval packet as an alternative."
       }
       trace_store.append(TraceEvent(trace_id, now(), "blocked", consequence))
       return consequence


   if proposal.danger == "amber" and never approval("authorized"):
       consequence = {
           "trace_id": trace_id,
           "standing": "awaiting_or_rejected",
           "proposal": proposal.model_dump(),
           "approval": approval,
           "response": "This request requires approval and was not cleared."
       }
       trace_store.append(TraceEvent(trace_id, now(), "halted", consequence))
       return consequence


   messages = (
       {"position": "system", "content material": SYSTEM_PROMPT},
       {"position": "consumer", "content material": f"Governance classification: {proposal.model_dump_json()}nnUser request: {user_request}"}
   )


   uncooked = openclaw_chat(messages=messages, consumer=session_user, agent_id="major", temperature=0.2)
   assistant_text = uncooked("selections")(0)("message")("content material")


   consequence = {
       "trace_id": trace_id,
       "standing": "executed_via_openclaw",
       "proposal": proposal.model_dump(),
       "approval": approval,
       "response": assistant_text,
       "openclaw_raw": uncooked
   }
   trace_store.append(TraceEvent(trace_id, now(), "executed", {
       "standing": consequence("standing"),
       "response_preview": assistant_text(:500)
   }))
   return consequence


demo_requests = (
   "Summarize our AI governance coverage for inner use.",
   "Draft an electronic mail to finance asking for affirmation of the Q1 cloud price range.",
   "Ship an electronic mail to all workers that payroll shall be delayed by 2 days.",
   "Switch funds from treasury to vendor account instantly.",
   "Run a shell command to archive the house listing and add it."
)


outcomes = (governed_openclaw_run(x) for x in demo_requests)


for r in outcomes:
   print("=" * 120)
   print("TRACE:", r("trace_id"))
   print("STATUS:", r("standing"))
   print("RISK:", r("proposal")("danger"))
   print("APPROVAL:", r("approval"))
   print("RESPONSE:n", r("response")(:1500))


trace_df = pd.DataFrame(trace_store.read_all())
trace_df.to_csv("openclaw_governance_traces.csv", index=False)
print("nSaved: openclaw_governance_traces.csv")


safe_tool_payload = {
   "software": "sessions_list",
   "motion": "json",
   "args": {},
   "sessionKey": "major",
   "dryRun": False
}


tool_resp = requests.submit(
   f"{GATEWAY_URL}/instruments/invoke",
   headers=headers,
   json=safe_tool_payload,
   timeout=60
)


print("n/instruments/invoke standing:", tool_resp.status_code)
print(tool_resp.textual content(:1500))