Request-Aufbau
Modellauswahl
Prompt-Übergabe
safety_identifier
Fehlerbehandlung
Text-Extraktion
Logging der Nutzung

src/openai_client.py

import hashlib
from dataclasses import dataclass
from typing import Any

from openai import OpenAI
from openai import APIConnectionError
from openai import APIStatusError
from openai import APITimeoutError
from openai import RateLimitError


@dataclass(frozen=True)
class OpenAiTextRequest:
    model: str
    instructions: str
    input_text: str
    safety_identifier: str
    max_output_tokens: int


@dataclass(frozen=True)
class OpenAiTextResult:
    text: str
    model: str
    input_tokens: int
    output_tokens: int
    total_tokens: int
    response_id: str


class OpenAiClient:
    def __init__(self) -> None:
        self.client = OpenAI()

    def create_text(self, request: OpenAiTextRequest) -> OpenAiTextResult:
        response = self.client.responses.create(
            model=request.model,
            instructions=request.instructions,
            input=request.input_text,
            safety_identifier=request.safety_identifier,
            max_output_tokens=request.max_output_tokens,
            store=True,
        )

        text = response.output_text

        if isinstance(text, str) is False:
            raise RuntimeError("OpenAI response output_text is not a string.")

        if text.strip() == "":
            raise RuntimeError("OpenAI response output_text is empty.")

        input_tokens = 0
        output_tokens = 0
        total_tokens = 0

        if response.usage is not None:
            input_tokens = response.usage.input_tokens
            output_tokens = response.usage.output_tokens
            total_tokens = response.usage.total_tokens

        return OpenAiTextResult(
            text=text,
            model=response.model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            total_tokens=total_tokens,
            response_id=response.id,
        )


class SafetyIdentifierFactory:
    def create(self, raw_identifier: str) -> str:
        if raw_identifier.strip() == "":
            raise RuntimeError("Safety identifier source must not be empty.")

        digest = hashlib.sha256(raw_identifier.encode("utf-8")).hexdigest()

        return digest[:64]

instructions
input

from src.openai_client import OpenAiClient
from src.openai_client import OpenAiTextRequest
from src.openai_client import SafetyIdentifierFactory


def main() -> None:
    safety_identifier_factory = SafetyIdentifierFactory()
    safety_identifier = safety_identifier_factory.create("publisher-system:xyz-123")

    request = OpenAiTextRequest(
        model="gpt-4.1-mini",
        instructions=(
            "Du schreibst sachliche Social-Media-Texte für einen kleinen "
            "deutschen E-Book-Verlag. Keine Emojis. Keine Floskeln. "
            "Keine übertriebene Werbung."
        ),
        input_text=(
            "Erzeuge einen LinkedIn-Beitrag über die verlegerische Arbeit "
            "an gemeinfreien Klassikern. Maximal 900 Zeichen."
        ),
        safety_identifier=safety_identifier,
        max_output_tokens=500,
    )

    client = OpenAiClient()
    result = client.create_text(request)

    print(result.text)


if __name__ == "__main__":
    main()

openai:
  default_model: "gpt-4.1-mini"
  quality_model: "gpt-4.1-mini"
  comment_model: "gpt-4.1-mini"
  max_output_tokens_post: 700
  max_output_tokens_quality: 500
  max_output_tokens_comment: 300

from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True)
class OpenAiConfig:
    default_model: str
    quality_model: str
    comment_model: str
    max_output_tokens_post: int
    max_output_tokens_quality: int
    max_output_tokens_comment: int


class OpenAiConfigReader:
    def __init__(self, raw_config: dict[str, Any]) -> None:
        self.raw_config = raw_config

    def read(self) -> OpenAiConfig:
        openai_config = self.raw_config.get("openai")

        if isinstance(openai_config, dict) is False:
            raise RuntimeError("Missing openai configuration.")

        default_model = self._read_string(openai_config, "default_model")
        quality_model = self._read_string(openai_config, "quality_model")
        comment_model = self._read_string(openai_config, "comment_model")

        max_output_tokens_post = self._read_int(openai_config, "max_output_tokens_post")
        max_output_tokens_quality = self._read_int(openai_config, "max_output_tokens_quality")
        max_output_tokens_comment = self._read_int(openai_config, "max_output_tokens_comment")

        return OpenAiConfig(
            default_model=default_model,
            quality_model=quality_model,
            comment_model=comment_model,
            max_output_tokens_post=max_output_tokens_post,
            max_output_tokens_quality=max_output_tokens_quality,
            max_output_tokens_comment=max_output_tokens_comment,
        )

    def _read_string(self, payload: dict[str, Any], key: str) -> str:
        value = payload.get(key)

        if isinstance(value, str) is False:
            raise RuntimeError(f"openai.{key} must be a string.")

        if value.strip() == "":
            raise RuntimeError(f"openai.{key} must not be empty.")

        return value

    def _read_int(self, payload: dict[str, Any], key: str) -> int:
        value = payload.get(key)

        if isinstance(value, int) is False:
            raise RuntimeError(f"openai.{key} must be an integer.")

        if value <= 0:
            raise RuntimeError(f"openai.{key} must be greater than zero.")

        return value

publisher-system:xyz-123

user:{interne_user_id}

E-Mail-Adresse im Klartext
Name im Klartext
Kundendaten im Klartext

import hashlib


class SafetyIdentifierFactory:
    def create(self, raw_identifier: str) -> str:
        cleaned_identifier = raw_identifier.strip()

        if cleaned_identifier == "":
            raise RuntimeError("Safety identifier source must not be empty.")

        digest = hashlib.sha256(cleaned_identifier.encode("utf-8")).hexdigest()

        return digest[:64]

safety_identifier_factory = SafetyIdentifierFactory()
safety_identifier = safety_identifier_factory.create("publisher-system:xyz-123")

response = client.responses.create(
    model="gpt-4.1-mini",
    instructions=instructions,
    input=input_text,
    safety_identifier=safety_identifier,
    max_output_tokens=500,
)

temporärer Fehler
Rate-Limit
Timeout
API-Fehler
leere Antwort
unvollständige Antwort
Validierungsfehler

class OpenAiClientError(Exception):
    pass


class OpenAiTemporaryError(OpenAiClientError):
    pass


class OpenAiRateLimitClientError(OpenAiTemporaryError):
    pass


class OpenAiPermanentError(OpenAiClientError):
    pass


class OpenAiEmptyResponseError(OpenAiPermanentError):
    pass


class OpenAiInvalidResponseError(OpenAiPermanentError):
    pass

class RobustOpenAiClient:
    def __init__(self) -> None:
        self.client = OpenAI()

    def create_text(self, request: OpenAiTextRequest) -> OpenAiTextResult:
        try:
            response = self.client.responses.create(
                model=request.model,
                instructions=request.instructions,
                input=request.input_text,
                safety_identifier=request.safety_identifier,
                max_output_tokens=request.max_output_tokens,
                store=True,
            )
        except RateLimitError as exception:
            raise OpenAiRateLimitClientError(str(exception)) from exception
        except APITimeoutError as exception:
            raise OpenAiTemporaryError(str(exception)) from exception
        except APIConnectionError as exception:
            raise OpenAiTemporaryError(str(exception)) from exception
        except APIStatusError as exception:
            if exception.status_code >= 500:
                raise OpenAiTemporaryError(str(exception)) from exception

            raise OpenAiPermanentError(str(exception)) from exception

        if response.status != "completed":
            raise OpenAiInvalidResponseError(f"OpenAI response status is {response.status}.")

        if response.incomplete_details is not None:
            raise OpenAiInvalidResponseError("OpenAI response is incomplete.")

        text = response.output_text

        if isinstance(text, str) is False:
            raise OpenAiInvalidResponseError("OpenAI response output_text is not a string.")

        cleaned_text = text.strip()

        if cleaned_text == "":
            raise OpenAiEmptyResponseError("OpenAI response output_text is empty.")

        input_tokens = 0
        output_tokens = 0
        total_tokens = 0

        if response.usage is not None:
            input_tokens = response.usage.input_tokens
            output_tokens = response.usage.output_tokens
            total_tokens = response.usage.total_tokens

        return OpenAiTextResult(
            text=cleaned_text,
            model=response.model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            total_tokens=total_tokens,
            response_id=response.id,
        )

LinkedIn-Beitrag: 500–900 Zeichen
Instagram-Caption: 400–800 Zeichen
Kommentarvorschlag: 100–400 Zeichen
Quality-Gate JSON: 200–600 Tokens

request = OpenAiTextRequest(
    model="gpt-4.1-mini",
    instructions=instructions,
    input_text=input_text,
    safety_identifier=safety_identifier,
    max_output_tokens=500,
)

{
  "event": "openai_request_completed",
  "model": "gpt-4.1-mini",
  "input_tokens": 421,
  "output_tokens": 186,
  "total_tokens": 607,
  "response_id": "resp_..."
}

from dataclasses import asdict
import json
from pathlib import Path


class OpenAiUsageLogger:
    def __init__(self, log_file: Path) -> None:
        self.log_file = log_file

    def write(self, result: OpenAiTextResult) -> None:
        self.log_file.parent.mkdir(parents=True, exist_ok=True)

        payload = {
            "event": "openai_request_completed",
            "model": result.model,
            "input_tokens": result.input_tokens,
            "output_tokens": result.output_tokens,
            "total_tokens": result.total_tokens,
            "response_id": result.response_id,
        }

        with self.log_file.open("a", encoding="utf-8") as file:
            file.write(json.dumps(payload, ensure_ascii=False))
            file.write("\n")

prompts/
├── linkedin_post.txt
├── instagram_caption.txt
├── facebook_post.txt
├── quality_check.txt
├── comment_suggestion.txt
└── reply_to_comment.txt

prompts/linkedin_post.txt

Du schreibst einen sachlichen LinkedIn-Beitrag für einen kleinen deutschen E-Book-Verlag.

Thema:
{{ title }}

Ausgangsmaterial:
{{ body }}

Autor:
{{ author }}

Buchtitel:
{{ book_title }}

Ziel:
Fachliche Sichtbarkeit für den Verlag erzeugen, ohne werblich oder generisch zu klingen.

Regeln:
- maximal 900 Zeichen
- keine Emojis
- keine Hashtag-Liste
- keine Floskeln
- keine Übertreibung
- sachlich, konkret, leicht pointiert
- kein direkter Kaufdruck

Erzeuge nur den Beitragstext.

from pathlib import Path


class PromptTemplateRenderer:
    def __init__(self, prompts_dir: Path) -> None:
        self.prompts_dir = prompts_dir

    def render(self, template_name: str, variables: dict[str, str]) -> str:
        template_path = self.prompts_dir / template_name

        if template_path.exists() is False:
            raise RuntimeError(f"Prompt template does not exist: {template_name}")

        content = template_path.read_text(encoding="utf-8")

        rendered = content

        for key, value in variables.items():
            placeholder = "{{ " + key + " }}"
            rendered = rendered.replace(placeholder, value)

        if "{{ " in rendered:
            raise RuntimeError(f"Prompt template contains unresolved placeholders: {template_name}")

        return rendered

renderer = PromptTemplateRenderer(root_dir / "prompts")

input_text = renderer.render(
    "linkedin_post.txt",
    {
        "title": "Warum gemeinfreie Literatur trotzdem Verlagsarbeit macht",
        "body": "OCR-Korrektur, Typografie, Metadaten und EPUB3-Erstellung bleiben echte Arbeit.",
        "author": "",
        "book_title": "",
    },
)

from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True)
class OpenAiJsonRequest:
    model: str
    instructions: str
    input_text: str
    safety_identifier: str
    max_output_tokens: int
    schema_name: str
    schema: dict[str, Any]

import json


class OpenAiStructuredClient:
    def __init__(self) -> None:
        self.client = OpenAI()

    def create_json(self, request: OpenAiJsonRequest) -> dict[str, Any]:
        response = self.client.responses.create(
            model=request.model,
            instructions=request.instructions,
            input=request.input_text,
            safety_identifier=request.safety_identifier,
            max_output_tokens=request.max_output_tokens,
            text={
                "format": {
                    "type": "json_schema",
                    "name": request.schema_name,
                    "strict": True,
                    "schema": request.schema,
                }
            },
        )

        text = response.output_text

        if isinstance(text, str) is False:
            raise RuntimeError("OpenAI JSON response output_text is not a string.")

        if text.strip() == "":
            raise RuntimeError("OpenAI JSON response output_text is empty.")

        decoded = json.loads(text)

        if isinstance(decoded, dict) is False:
            raise RuntimeError("OpenAI JSON response must be an object.")

        return decoded

QUALITY_GATE_SCHEMA: dict[str, object] = {
    "type": "object",
    "properties": {
        "passed": {
            "type": "boolean"
        },
        "reason": {
            "type": "string"
        },
        "checks": {
            "type": "object",
            "properties": {
                "not_generic": {
                    "type": "boolean"
                },
                "not_too_salesy": {
                    "type": "boolean"
                },
                "has_clear_topic": {
                    "type": "boolean"
                },
                "within_length": {
                    "type": "boolean"
                },
                "safe_for_publishing": {
                    "type": "boolean"
                }
            },
            "required": [
                "not_generic",
                "not_too_salesy",
                "has_clear_topic",
                "within_length",
                "safe_for_publishing"
            ],
            "additionalProperties": False
        }
    },
    "required": [
        "passed",
        "reason",
        "checks"
    ],
    "additionalProperties": False
}

request = OpenAiJsonRequest(
    model="gpt-4.1-mini",
    instructions="Prüfe Social-Media-Beiträge für einen deutschen E-Book-Verlag.",
    input_text=(
        "Prüfe diesen Beitrag:\n\n"
        "Jane Austens Roman ist bis heute lesbar, weil er gesellschaftliche Erwartungen nicht behauptet, sondern sichtbar macht."
    ),
    safety_identifier=safety_identifier,
    max_output_tokens=500,
    schema_name="quality_gate_result",
    schema=QUALITY_GATE_SCHEMA,
)

client = OpenAiStructuredClient()
result = client.create_json(request)

print(result)

from dataclasses import dataclass


@dataclass(frozen=True)
class QualityGateChecks:
    not_generic: bool
    not_too_salesy: bool
    has_clear_topic: bool
    within_length: bool
    safe_for_publishing: bool


@dataclass(frozen=True)
class QualityGateResult:
    passed: bool
    reason: str
    checks: QualityGateChecks

from typing import Any


class QualityGateResultValidator:
    def validate(self, payload: dict[str, Any]) -> QualityGateResult:
        passed = payload.get("passed")
        reason = payload.get("reason")
        checks = payload.get("checks")

        if isinstance(passed, bool) is False:
            raise RuntimeError("quality_gate.passed must be a boolean.")

        if isinstance(reason, str) is False:
            raise RuntimeError("quality_gate.reason must be a string.")

        if isinstance(checks, dict) is False:
            raise RuntimeError("quality_gate.checks must be an object.")

        not_generic = self._read_bool(checks, "not_generic")
        not_too_salesy = self._read_bool(checks, "not_too_salesy")
        has_clear_topic = self._read_bool(checks, "has_clear_topic")
        within_length = self._read_bool(checks, "within_length")
        safe_for_publishing = self._read_bool(checks, "safe_for_publishing")

        if reason.strip() == "":
            raise RuntimeError("quality_gate.reason must not be empty.")

        return QualityGateResult(
            passed=passed,
            reason=reason,
            checks=QualityGateChecks(
                not_generic=not_generic,
                not_too_salesy=not_too_salesy,
                has_clear_topic=has_clear_topic,
                within_length=within_length,
                safe_for_publishing=safe_for_publishing,
            ),
        )

    def _read_bool(self, payload: dict[str, Any], key: str) -> bool:
        value = payload.get(key)

        if isinstance(value, bool) is False:
            raise RuntimeError(f"quality_gate.checks.{key} must be a boolean.")

        return value

class PostTextValidator:
    def validate_linkedin_text(self, text: str) -> None:
        cleaned_text = text.strip()

        if cleaned_text == "":
            raise RuntimeError("Post text must not be empty.")

        if len(cleaned_text) > 900:
            raise RuntimeError("LinkedIn post text exceeds 900 characters.")

        forbidden_fragments = [
            "Jetzt kaufen",
            "unbedingt lesen",
            "Dieses Meisterwerk",
            "Tauchen Sie ein",
        ]

        for forbidden_fragment in forbidden_fragments:
            if forbidden_fragment in cleaned_text:
                raise RuntimeError(f"Post text contains forbidden fragment: {forbidden_fragment}")

Prompt aus Template rendern
Request bauen
safety_identifier setzen
OpenAI aufrufen
Antwort extrahieren
Antwort validieren
Tokenverbrauch loggen
Entwurf speichern

eine zentrale Client-Schicht
Responses API statt verstreuter Einzelaufrufe
safety_identifier korrekt im API-Request
Modellwahl über config.yaml
harte Fehlerklassifikation
max_output_tokens für Kostenkontrolle
Prompt-Templates als Dateien
strukturierte JSON-Ausgabe für Prüfprozesse
Validierung jeder KI-Antwort vor Weiterverarbeitung