AI-Agent-Plattform: Ein robuster Entwurf

Einen AI-Agenten zu entwickeln, der auf einen Prompt antwortet, ist vergleichsweise einfach.

Eine AI-Agent-Plattform zu entwickeln, die reale Workloads zuverlässig ausführt, sich nach Fehlern wiederherstellt, Tools sicher aufruft, Zustände beibehält und über mehrere Worker skaliert, ist eine andere Engineering-Herausforderung.

Diese Referenzarchitektur zeigt, wie TypeScript, Python, Redis, PostgreSQL, Docker und Kubernetes zuverlässige Agent-Workflows unterstützen können.

Das Ziel dieses Entwurfs ist nicht ein weiterer Chatbot, sondern eine wiederverwendbare Plattform für intelligente, toolgestützte Workflows innerhalb eines zuverlässigen Backend-Systems.

Das Problem#

Eine einfache AI-Integration beginnt häufig mit einer einzelnen API-Anfrage:

const response = await openai.responses.create({
  model: "gpt-5",
  input: "Analyze this customer request.",
})

Dieser Ansatz eignet sich für Demonstrationen, doch produktive Systeme stellen schnell deutlich höhere Anforderungen.

Eine robuste Agent-Plattform könnte Folgendes unterstützen müssen:

• Mehrere spezialisierte Agenten.
• Lang laufende Workflows.
• Strukturierte Tool-Aufrufe.
• Menschliche Freigabeschritte.
• Wiederholbare Hintergrundaufgaben.
• Persistente Konversations- und Workflow-Zustände.
• Lokale und in der Cloud gehostete Sprachmodelle.
• Horizontale Skalierung der Worker.
• Benutzerbezogene Berechtigungen und Nutzungslimits.
• Detaillierte Ausführungsprotokolle.
• Zuverlässige Wiederherstellung nach Prozess- oder Serverausfällen.

Die zentrale Herausforderung besteht darin, probabilistisches AI-Reasoning mit deterministischer Backend-Infrastruktur zu verbinden.

Sprachmodelle sind nicht deterministisch. Produktive Systeme dürfen bei Sicherheit, Zustandsübergängen, Abrechnung, Wiederholungsversuchen oder Datenintegrität jedoch nicht unvorhersehbar sein.

Die Referenzarchitektur trennt das AI-Reasoning daher von den Systemteilen, für die strikte Garantien erforderlich sind.

Designziele#

Der Referenzentwurf beginnt mit den folgenden Designzielen.

Zuverlässigkeit#

Eine fehlgeschlagene Modellanfrage, ein Neustart eines Workers, eine Netzwerkunterbrechung oder ein Tool-Timeout darf den Workflow nicht beschädigen.

Observability#

Jeder Workflow-Schritt muss genügend Informationen liefern, um folgende Fragen zu beantworten:

• Was ist passiert?
• Welches Modell wurde verwendet?
• Welches Tool wurde aufgerufen?
• Wie lange dauerte die Ausführung?
• Warum ist sie fehlgeschlagen?
• Kann sie sicher wiederholt werden?

Skalierbarkeit#

API-Server, Agent-Worker und Tool-Worker müssen unabhängig voneinander skalierbar sein.

Sicherheit#

Das Modell darf niemals uneingeschränkten Zugriff auf interne Services, Datenbanken, Dateien oder Benutzerzugangsdaten erhalten.

Erweiterbarkeit#

Neue Modelle, Tools, Agenten und Workflow-Typen müssen ergänzt werden können, ohne die gesamte Plattform neu zu implementieren.

Modellunabhängigkeit#

Die Anwendung muss Cloud-Anbieter und lokale Modelle über eine gemeinsame interne Abstraktion unterstützen.

Architektur auf hoher Ebene#

Die Referenzarchitektur ist in mehrere klar definierte Ebenen unterteilt.

Diese Trennung verhindert, dass das Sprachmodell zum Zentrum der gesamten Anwendung wird.

Das Modell nimmt an einem Workflow teil, besitzt den Workflow jedoch nicht.

Lebenszyklus einer Anfrage#

Eine typische Anfrage durchläuft das System in folgender Reihenfolge:

1. Ein Client erstellt über die API einen Agent-Run.
2. Die API validiert die Anfrage und prüft die Berechtigungen.
3. In PostgreSQL wird ein dauerhafter Workflow-Datensatz erstellt.
4. In Redis wird ein Job eingereiht.
5. Ein verfügbarer Agent-Worker übernimmt den Job.
6. Der Worker lädt den Workflow-Kontext.
7. Das Modell entscheidet, ob es antworten, ein Tool aufrufen oder eine Freigabe anfordern muss.
8. Tool-Aufrufe werden validiert und an isolierte Worker weitergeleitet.
9. Die Ergebnisse werden in PostgreSQL gespeichert.
10. Der nächste Workflow-Schritt wird geplant.
11. Der Client erhält Aktualisierungen per Polling, Server-Sent Events oder WebSockets.

Dadurch bleibt die API-Anfrage kurzlebig, während der eigentliche Agent-Workflow über Sekunden oder Minuten weiterlaufen kann.

Warum API und Agent-Ausführung getrennt werden sollten#

Die direkte Ausführung von Agent-Workflows innerhalb einer HTTP-Anfrage verursacht mehrere Probleme:

• Anfragen können die Timeouts der Plattform überschreiten.
• Ein Prozessneustart zerstört den im Arbeitsspeicher gehaltenen Zustand.
• Rechenintensive AI-Aufgaben konkurrieren mit normalem API-Traffic.
• Eine Wiederholung der HTTP-Anfrage kann Seiteneffekte erneut ausführen.
• Die Skalierung der API skaliert unnötig auch die ressourcenintensiven Agent-Komponenten.

In diesem Entwurf erstellt die API einen dauerhaften Run und plant die asynchrone Verarbeitung.

type CreateAgentRunInput = {
  userId: string
  agentType: string
  message: string
}

export async function createAgentRun(
  input: CreateAgentRunInput
) {
  const run = await database.transaction(async (transaction) => {
    const createdRun = await transaction.agentRun.create({
      data: {
        userId: input.userId,
        agentType: input.agentType,
        status: "queued",
        input: {
          message: input.message,
        },
      },
    })

    await transaction.outboxEvent.create({
      data: {
        type: "agent.run.created",
        aggregateId: createdRun.id,
        payload: {
          runId: createdRun.id,
        },
      },
    })

    return createdRun
  })

  return run
}

In diesem Beispiel veröffentlicht die API innerhalb der Transaktion nicht unmittelbar eine Nachricht in Redis. Stattdessen speichert sie ein Outbox-Ereignis.

Dieser Unterschied ist entscheidend.

Die transaktionale Outbox#

Ein typisches Fehlerszenario sieht folgendermaßen aus:

1. PostgreSQL erstellt den Workflow erfolgreich.
2. Die Anwendung versucht, eine Queue-Nachricht zu veröffentlichen.
3. Redis ist vorübergehend nicht verfügbar.
4. Die Datenbank enthält einen eingereihten Workflow, aber kein Worker weiß davon.

In diesem Entwurf adressiert eine transaktionale Outbox das Problem, indem Workflow und Ereignis innerhalb derselben PostgreSQL-Transaktion gespeichert werden.

Ein separater Publisher könnte unverarbeitete Outbox-Events lesen und in Redis veröffentlichen.

export async function publishOutboxBatch() {
  const events = await database.outboxEvent.findMany({
    where: {
      processedAt: null,
    },
    orderBy: {
      createdAt: "asc",
    },
    take: 100,
  })

  for (const event of events) {
    await queue.publish(event.type, event.payload)

    await database.outboxEvent.update({
      where: {
        id: event.id,
      },
      data: {
        processedAt: new Date(),
      },
    })
  }
}

Dauerhafter Workflow-Zustand#

Redis eignet sich für Queues, Locks, kurzlebige Caches und Koordination.

Es ist jedoch nicht die primäre Source of Truth für die Agent-Ausführung.

In diesem Referenzentwurf speichert PostgreSQL den dauerhaften Zustand jedes Runs.

CREATE TABLE agent_runs (
  id UUID PRIMARY KEY,
  user_id UUID NOT NULL,
  agent_type TEXT NOT NULL,
  status TEXT NOT NULL,
  input JSONB NOT NULL,
  output JSONB,
  current_step INTEGER NOT NULL DEFAULT 0,
  version INTEGER NOT NULL DEFAULT 1,
  started_at TIMESTAMPTZ,
  completed_at TIMESTAMPTZ,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Jeder einzelne Schritt wird separat gespeichert:

CREATE TABLE agent_run_steps (
  id UUID PRIMARY KEY,
  run_id UUID NOT NULL REFERENCES agent_runs(id),
  step_number INTEGER NOT NULL,
  step_type TEXT NOT NULL,
  status TEXT NOT NULL,
  model TEXT,
  tool_name TEXT,
  input JSONB,
  output JSONB,
  error JSONB,
  started_at TIMESTAMPTZ,
  completed_at TIMESTAMPTZ,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),

  UNIQUE (run_id, step_number)
);

Dieses Schema bildet die Grundlage für einen auditierbaren Ausführungsverlauf.

Mit entsprechender Recovery-Logik und Tests könnte ein Worker nach einem Neustart ab dem letzten abgeschlossenen Schritt fortfahren, statt den gesamten Workflow erneut zu starten.

Design des Agent-Workers#

Der illustrative Python-Worker ist für die Modellinteraktion, strukturierte Ausgaben und Reasoning-bezogene Operationen verantwortlich.

Er ist nicht für einen uneingeschränkten Zugriff auf die Infrastruktur zuständig.

from dataclasses import dataclass
from typing import Any

@dataclass
class AgentJob:
    run_id: str
    agent_type: str
    input: dict[str, Any]

async def process_agent_job(job: AgentJob) -> None:
    run = await workflow_repository.get_run(job.run_id)

    if run.status in {"completed", "cancelled"}:
        return

    await workflow_repository.mark_running(job.run_id)

    try:
        context = await context_builder.build(run)
        decision = await agent_runtime.execute(context)

        await workflow_engine.apply_decision(
            run_id=job.run_id,
            decision=decision,
        )

    except Exception as error:
        await workflow_repository.record_failure(
            run_id=job.run_id,
            error=str(error),
        )

        raise

In diesem Entwurf erhält der Worker eine Job-ID und lädt den aktuellen Zustand erneut aus PostgreSQL.

Dadurch muss kein großer oder veralteter Workflow-Zustand über die Queue transportiert werden.

Strukturierte Modellausgabe#

Frei formulierte Modellantworten lassen sich nur schwer sicher verarbeiten.

Der Agent erzeugt daher eine von wenigen strukturierten Entscheidungen.

import { z } from "zod"

export const agentDecisionSchema = z.discriminatedUnion("type", [
  z.object({
    type: z.literal("final_answer"),
    content: z.string().min(1),
  }),

  z.object({
    type: z.literal("tool_call"),
    tool: z.string().min(1),
    arguments: z.record(z.unknown()),
  }),

  z.object({
    type: z.literal("request_approval"),
    reason: z.string().min(1),
    proposedAction: z.record(z.unknown()),
  }),

  z.object({
    type: z.literal("continue"),
    reason: z.string().min(1),
  }),
])

export type AgentDecision = z.infer<
  typeof agentDecisionSchema
>

Dadurch wird das Modell nicht deterministisch, das Backend erhält jedoch einen kontrollierten Vertrag.

Die vorgesehene Grenze lehnt fehlerhaft formatierte Ausgaben ab, anstatt die Absicht des Modells zu erraten.

Grenzen der Tool-Ausführung#

Tools gehören zu den leistungsfähigsten und zugleich riskantesten Komponenten einer Agent-Plattform.

Ein Tool kann:

• Interne Datenbanken abfragen.
• Eine E-Mail senden.
• Einen Kalender ändern.
• Ein Support-Ticket erstellen.
• Code ausführen.
• Ein Dokument lesen.
• Eine externe API aufrufen.

Das Modell sollte niemals direkte Zugangsdaten erhalten.

Stattdessen fordert es ein benanntes Tool mit strukturierten Argumenten an.

{
  "type": "tool_call",
  "tool": "create_support_ticket",
  "arguments": {
    "customerId": "cus_123",
    "subject": "Payment issue",
    "priority": "high"
  }
}

Ein produktives Backend sollte anschließend mehrere Prüfungen durchführen:

1. Ist das Tool registriert?
2. Darf dieser Agent das Tool verwenden?
3. Darf dieser Benutzer die Aktion auslösen?
4. Entsprechen die Argumente dem Schema?
5. Erfordert die Aktion eine Freigabe?
6. Wurde derselbe Seiteneffekt bereits ausgeführt?

import { z } from "zod"

const createTicketInputSchema = z.object({
  customerId: z.string().min(1),
  subject: z.string().min(3),
  priority: z.enum(["low", "normal", "high"]),
})

export const toolRegistry = {
  create_support_ticket: {
    inputSchema: createTicketInputSchema,
    requiresApproval: true,

    execute: async (input: unknown) => {
      const validatedInput =
        createTicketInputSchema.parse(input)

      return supportClient.createTicket(validatedInput)
    },
  },
}

Idempotenz und Seiteneffekte#

Wiederholungsversuche sind in verteilten Systemen unvermeidbar.

Ein Worker kann einen Tool-Aufruf abschließen und abstürzen, bevor er die Queue-Nachricht bestätigt. Die Queue kann denselben Job anschließend erneut zustellen.

Ohne Idempotenz könnte die Plattform dieselbe E-Mail zweimal senden oder doppelte Tickets erstellen.

Jede Operation mit Seiteneffekten sollte einen Idempotenzschlüssel erhalten:

type ExecuteToolInput = {
  runId: string
  stepId: string
  toolName: string
  arguments: unknown
}

export async function executeTool(
  input: ExecuteToolInput
) {
  const idempotencyKey =
    `${input.runId}:${input.stepId}:${input.toolName}`

  const existingExecution =
    await database.toolExecution.findUnique({
      where: {
        idempotencyKey,
      },
    })

  if (existingExecution?.status === "completed") {
    return existingExecution.result
  }

  return toolExecutionService.execute({
    ...input,
    idempotencyKey,
  })
}

Mit einer Eindeutigkeitsbedingung und einem transaktionalen Ausführungsdatensatz kann dieses Muster sichere Wiederholungen unterstützen. Die illustrative Abfrage allein belegt diese Garantie nicht.

Aufgaben von Redis#

In diesem Referenzentwurf wird Redis für kurzlebige operative Aufgaben eingesetzt:

Redis sollte nicht als einziger Speicherort für geschäftskritische Workflow-Zustände verwendet werden.