Strukturierte Ausgaben RAG: JSON, Tabellen und benutzerdefinierte Formate

TL;DR

Die strukturieren Outputs ermöglichen es, RAG-Antworten in programmatisch verwertbaren Formaten zu erzeugen: JSON, Tabellen, typisierte Listen. Dieser Ansatz ist essenziell für API-Integrationen, automatisierte Workflows und reichhaltige Oberflächen. Dieser Leitfaden behandelt Techniken zur Generierung, Validierung und zum Parsing strukturierter Outputs.

Warum strukturierte Outputs?

Das Problem mit Freitext

Freitext-Antworten sind schwer weiterzuverarbeiten:

DEVELOPERpython
# ❌ Freitext-Antwort (schwer zu parsen)
response = """
Produkt X kostet 49,99€ und ist auf Lager verfügbar.
Die Lieferung dauert 3-5 Werktage. Es ist in Blau,
Rot und Grün erhältlich. Die Garantie beträgt 2 Jahre.
"""

# Wie extrahiert man den Preis? Die Verfügbarkeit? Die Farben?

Die strukturierte Lösung

DEVELOPERpython
# ✅ Strukturierte JSON-Antwort
response = {
    "product": {
        "name": "Produkt X",
        "price": 49.99,
        "currency": "EUR",
        "in_stock": True,
        "colors": ["blau", "rot", "grün"],
        "shipping": {
            "min_days": 3,
            "max_days": 5,
            "type": "business_days"
        },
        "warranty_years": 2
    },
    "sources": ["fiche-produit-x.pdf", "conditions-garantie.pdf"]
}

Anwendungsfälle

Techniken zur strukturierten Generierung

1. Prompting mit Beispielen

DEVELOPERpython
STRUCTURED_PROMPT = """
Du bist ein Assistent, der AUSSCHLIESSLICH in gültigem JSON antwortet.

## Verpflichtendes Antwortformat
```json
{
  "answer": "Hauptantwort",
  "confidence": 0.0-1.0,
  "sources": ["source1", "source2"],
  "entities": {
    "prices": [{"value": 0, "currency": "EUR"}],
    "dates": ["YYYY-MM-DD"],
    "quantities": [{"value": 0, "unit": "string"}]
  },
  "follow_up_questions": ["Vorgeschlagene Frage 1"]
}

Dokumente

{context}

Frage

{query}

JSON-Antwort (nichts anderes)

"""

### 2. JSON-Modus der LLMs

Die meisten modernen LLMs unterstützen einen "JSON-Modus":

```python
from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4-turbo",
    response_format={"type": "json_object"},  # Force JSON
    messages=[
        {
            "role": "system",
            "content": "Du antwortest immer in gültigem JSON mit den Feldern: answer, confidence, sources."
        },
        {
            "role": "user",
            "content": f"Context: {context}\n\nQuestion: {query}"
        }
    ]
)

# Garantiert gültiges JSON
result = json.loads(response.choices[0].message.content)

3. JSON Schema mit Validierung

DEVELOPERpython
from pydantic import BaseModel, Field
from typing import List, Optional
import instructor

# Schema mit Pydantic definieren
class ProductInfo(BaseModel):
    name: str = Field(..., description="Produktname")
    price: float = Field(..., ge=0, description="Preis in Euro")
    in_stock: bool = Field(..., description="Verfügbarkeit")
    colors: List[str] = Field(default=[], description="Verfügbare Farben")

class RAGResponse(BaseModel):
    answer: str = Field(..., description="Hauptantwort")
    confidence: float = Field(..., ge=0, le=1, description="Konfidenzwert")
    products: List[ProductInfo] = Field(default=[], description="Erwähnte Produkte")
    sources: List[str] = Field(default=[], description="Verwendete Quellen")

# instructor verwenden, um das Schema zu garantieren
client = instructor.from_openai(OpenAI())

response = client.chat.completions.create(
    model="gpt-4-turbo",
    response_model=RAGResponse,  # Erzwingt das Schema
    messages=[
        {"role": "user", "content": f"Context: {context}\n\nQuestion: {query}"}
    ]
)

# response ist bereits typisiert und validiert
print(response.answer)
print(response.confidence)
for product in response.products:
    print(f"{product.name}: {product.price}€")

4. Function Calling

Funktionen nutzen, um den Output zu strukturieren:

DEVELOPERpython
from openai import OpenAI

client = OpenAI()

tools = [
    {
        "type": "function",
        "function": {
            "name": "provide_answer",
            "description": "Liefert eine strukturierte Antwort auf die Frage",
            "parameters": {
                "type": "object",
                "properties": {
                    "answer": {
                        "type": "string",
                        "description": "Die Antwort auf die Frage"
                    },
                    "confidence": {
                        "type": "number",
                        "minimum": 0,
                        "maximum": 1,
                        "description": "Konfidenzwert"
                    },
                    "sources": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "Quelldokumente"
                    },
                    "action_required": {
                        "type": "boolean",
                        "description": "Ob eine menschliche Aktion erforderlich ist"
                    }
                },
                "required": ["answer", "confidence", "sources"]
            }
        }
    }
]

response = client.chat.completions.create(
    model="gpt-4-turbo",
    messages=[
        {"role": "user", "content": f"Context: {context}\n\nQuestion: {query}"}
    ],
    tools=tools,
    tool_choice={"type": "function", "function": {"name": "provide_answer"}}
)

# Die Funktionsargumente extrahieren
import json
result = json.loads(response.choices[0].message.tool_calls[0].function.arguments)

Gängige Ausgabeformate

JSON-Format für APIs

DEVELOPERpython
API_RESPONSE_SCHEMA = {
    "type": "object",
    "properties": {
        "success": {"type": "boolean"},
        "data": {
            "type": "object",
            "properties": {
                "answer": {"type": "string"},
                "formatted_answer": {"type": "string"},  # HTML/Markdown
                "entities": {
                    "type": "object",
                    "properties": {
                        "products": {"type": "array"},
                        "prices": {"type": "array"},
                        "dates": {"type": "array"}
                    }
                }
            }
        },
        "metadata": {
            "type": "object",
            "properties": {
                "confidence": {"type": "number"},
                "sources": {"type": "array"},
                "processing_time_ms": {"type": "integer"}
            }
        }
    },
    "required": ["success", "data", "metadata"]
}

Vergleichstabellenformat

DEVELOPERpython
COMPARISON_PROMPT = """
Vergleiche die in den Dokumenten erwähnten Produkte.

## Dokumente
{context}

## Antwortformat (Markdown)
| Produkt | Preis | Lager | Garantie | Bewertung |
|---------|------|-------|----------|------|
| Name 1   | XX€  | Ja/Nein | X Jahre | X/5 |
| Name 2   | XX€  | Ja/Nein | X Jahre | X/5 |

## Zusammenfassung
[Empfehlungssatz basierend auf dem Vergleich]
"""

def parse_markdown_table(markdown: str) -> list[dict]:
    """Parst eine Markdown-Tabelle in eine Liste von Dictionaries."""
    lines = markdown.strip().split('\n')

    # Tabellenszeilen finden
    table_lines = [l for l in lines if l.startswith('|')]
    if len(table_lines) < 3:
        return []

    # Headers
    headers = [h.strip() for h in table_lines[0].split('|')[1:-1]]

    # Datenzeilen (Trennzeile überspringen)
    data = []
    for line in table_lines[2:]:
        values = [v.strip() for v in line.split('|')[1:-1]]
        if len(values) == len(headers):
            data.append(dict(zip(headers, values)))

    return data

Aktions-/Workflow-Format

DEVELOPERpython
from enum import Enum
from pydantic import BaseModel

class ActionType(str, Enum):
    ANSWER = "answer"
    ESCALATE = "escalate"
    CLARIFY = "clarify"
    REDIRECT = "redirect"

class WorkflowResponse(BaseModel):
    action: ActionType
    content: str
    next_steps: List[str] = []
    requires_human: bool = False
    confidence: float

    # Aktionsspezifische Daten
    escalation_reason: Optional[str] = None
    clarification_questions: Optional[List[str]] = None
    redirect_url: Optional[str] = None

WORKFLOW_PROMPT = """
Analysiere die Frage und bestimme die beste Aktion.

Mögliche Aktionen:
- ANSWER: Direkt antworten, wenn die Info in den Dokumenten steht
- ESCALATE: An einen Menschen weiterleiten, wenn komplex oder sensibel
- CLARIFY: Um Präzisierung bitten, wenn die Frage mehrdeutig ist
- REDIRECT: Auf eine Ressource verweisen, wenn außerhalb des Scopes

Dokumente: {context}
Frage: {query}

Antworte in JSON mit: action, content, next_steps, requires_human, confidence
"""

Robuste Validierung und Parsing

Validierung mit Retry

DEVELOPERpython
import json
from tenacity import retry, stop_after_attempt, retry_if_exception_type

class StructuredOutputGenerator:
    def __init__(self, llm_client, schema: dict):
        self.llm = llm_client
        self.schema = schema

    @retry(
        stop=stop_after_attempt(3),
        retry=retry_if_exception_type(json.JSONDecodeError)
    )
    async def generate(self, context: str, query: str) -> dict:
        """Generiert einen strukturierten Output mit automatischem Retry."""

        prompt = self._build_prompt(context, query)
        response = await self.llm.generate(prompt)

        # Parsing versuchen
        try:
            result = json.loads(response)
        except json.JSONDecodeError:
            # Versuchen, JSON aus dem Text zu extrahieren
            result = self._extract_json(response)

        # Gegen das Schema validieren
        self._validate(result)

        return result

    def _extract_json(self, text: str) -> dict:
        """Extrahiert das JSON aus einem Text, der auch anderes enthalten kann."""
        import re

        # Suche nach einem JSON-Block
        json_match = re.search(r'```json\s*(.*?)\s*```', text, re.DOTALL)
        if json_match:
            return json.loads(json_match.group(1))

        # Suche nach geschweiften Klammern
        brace_match = re.search(r'\{.*\}', text, re.DOTALL)
        if brace_match:
            return json.loads(brace_match.group(0))

        raise json.JSONDecodeError("No JSON found", text, 0)

    def _validate(self, data: dict) -> None:
        """Validiert die Daten gegen das Schema."""
        from jsonschema import validate, ValidationError

        try:
            validate(instance=data, schema=self.schema)
        except ValidationError as e:
            raise ValueError(f"Schema validation failed: {e.message}")

Parsing mit Fallback

DEVELOPERpython
class RobustParser:
    """Parser mit mehreren Fallback-Strategien."""

    def parse(self, response: str, expected_format: str) -> dict:
        strategies = [
            self._parse_json,
            self._parse_json_block,
            self._parse_key_value,
            self._parse_with_llm
        ]

        for strategy in strategies:
            try:
                result = strategy(response)
                if self._validate_structure(result, expected_format):
                    return result
            except Exception:
                continue

        # Letzter Fallback: rohen Text zurückgeben
        return {"raw_response": response, "parse_failed": True}

    def _parse_json(self, text: str) -> dict:
        return json.loads(text)

    def _parse_json_block(self, text: str) -> dict:
        import re
        match = re.search(r'```(?:json)?\s*(.*?)\s*```', text, re.DOTALL)
        if match:
            return json.loads(match.group(1))
        raise ValueError("No JSON block found")

    def _parse_key_value(self, text: str) -> dict:
        """Parst das Format Schlüssel: Wert."""
        result = {}
        for line in text.split('\n'):
            if ':' in line:
                key, value = line.split(':', 1)
                result[key.strip().lower().replace(' ', '_')] = value.strip()
        return result

    async def _parse_with_llm(self, text: str) -> dict:
        """Verwendet ein LLM, um die Struktur zu extrahieren."""
        prompt = f"""
        Extrahiere die strukturierten Informationen aus diesem Text als JSON:

        {text}

        JSON:
        """
        response = await self.llm.generate(prompt, temperature=0)
        return json.loads(response)

Spezialisierte Formate nach Anwendungsfall

E-Commerce: Produktdatenblatt

DEVELOPERpython
PRODUCT_SCHEMA = {
    "type": "object",
    "properties": {
        "product": {
            "type": "object",
            "properties": {
                "sku": {"type": "string"},
                "name": {"type": "string"},
                "description": {"type": "string"},
                "price": {
                    "type": "object",
                    "properties": {
                        "amount": {"type": "number"},
                        "currency": {"type": "string"},
                        "discount_percent": {"type": "number"}
                    }
                },
                "availability": {
                    "type": "object",
                    "properties": {
                        "in_stock": {"type": "boolean"},
                        "quantity": {"type": "integer"},
                        "delivery_days": {"type": "integer"}
                    }
                },
                "variants": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "color": {"type": "string"},
                            "size": {"type": "string"},
                            "sku_variant": {"type": "string"}
                        }
                    }
                }
            },
            "required": ["name", "price", "availability"]
        },
        "recommendations": {
            "type": "array",
            "items": {"type": "string"}
        }
    }
}

Support: Strukturiertes Ticket

DEVELOPERpython
from pydantic import BaseModel
from enum import Enum

class Priority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    URGENT = "urgent"

class Category(str, Enum):
    BILLING = "billing"
    TECHNICAL = "technical"
    SHIPPING = "shipping"
    PRODUCT = "product"
    OTHER = "other"

class TicketResponse(BaseModel):
    summary: str
    category: Category
    priority: Priority
    resolution: Optional[str]
    requires_action: bool
    action_items: List[str] = []
    related_articles: List[str] = []
    sentiment: str  # positive, neutral, negative

TICKET_PROMPT = """
Analysiere diese Kundenanfrage und strukturiere die Antwort.

Dokumente: {context}
Anfrage: {query}

Antworte in JSON mit:
- summary: Zusammenfassung der Anfrage
- category: billing/technical/shipping/product/other
- priority: low/medium/high/urgent
- resolution: Lösung, falls gefunden
- requires_action: true, wenn menschliche Aktion erforderlich
- action_items: Liste der durchzuführenden Aktionen
- related_articles: Relevante Artikel
- sentiment: positive/neutral/negative
"""

HR: Richtlinienextraktion

DEVELOPERpython
POLICY_EXTRACTION_SCHEMA = {
    "type": "object",
    "properties": {
        "policy_name": {"type": "string"},
        "effective_date": {"type": "string", "format": "date"},
        "key_points": {
            "type": "array",
            "items": {"type": "string"}
        },
        "eligibility": {
            "type": "object",
            "properties": {
                "who": {"type": "array", "items": {"type": "string"}},
                "conditions": {"type": "array", "items": {"type": "string"}}
            }
        },
        "process": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "step": {"type": "integer"},
                    "action": {"type": "string"},
                    "responsible": {"type": "string"}
                }
            }
        },
        "exceptions": {"type": "array", "items": {"type": "string"}},
        "contact": {
            "type": "object",
            "properties": {
                "email": {"type": "string"},
                "department": {"type": "string"}
            }
        }
    }
}

Integration mit Ailog

Ailog unterstützt strukturierte Outputs nativ:

DEVELOPERpython
from ailog import AilogClient
from ailog.schemas import ProductComparison, SupportTicket

client = AilogClient(api_key="your-key")

# Strukturierter Produktvergleich
comparison = client.chat(
    channel_id="ecommerce-widget",
    message="Vergleiche das MacBook Pro und das Dell XPS",
    output_format=ProductComparison,  # Pydantic-Schema
)

print(comparison.products)  # Typisierte Liste
print(comparison.recommendation)

# Strukturiertes Support-Ticket
ticket = client.chat(
    channel_id="support-widget",
    message="Meine Bestellung 12345 ist nicht angekommen",
    output_format=SupportTicket,
)

if ticket.requires_action:
    create_zendesk_ticket(ticket)

Fazit

Strukturierte Outputs verwandeln Ihr RAG in ein leistungsstarkes Integrationswerkzeug. Schlüsselpunkte:

1. JSON Schema zur Struktursicherung
2. Pydantic/instructor für die Python-Validierung
3. Retry mit Fallback für Robustheit
4. Spezialisierte Formate nach Anwendungsfall
5. Function Calling für komplexe Workflows

Weiterführende Ressourcen

• Einführung in RAG - Grundlagen
• LLM-Generierung für RAG - Übergeordneter Leitfaden
• Prompt Engineering RAG - Prompts optimieren
• Streaming RAG - Echtzeit-Antworten

Strukturierte Outputs ohne Komplexität? Testen Sie Ailog - integrierte Schemas, automatische Validierung, fertige E-Commerce- und Support-Formate.