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.

Pourquoi les outputs structurés ?

Le problème du texte libre

Les réponses en texte libre sont difficiles à exploiter :

DEVELOPERpython
# ❌ Freitext-Antwort (schwer zu parsen)
response = """
Le produit X coûte 49,99€ et est disponible en stock.
La livraison prend 3-5 jours ouvrés. Il existe en bleu,
rouge et vert. La garantie est de 2 ans.
"""

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

La solution structurée

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

Cas d'usage

Techniques de génération structurée

1. Prompting avec exemples

DEVELOPERpython
STRUCTURED_PROMPT = """
Tu es un assistant qui répond UNIQUEMENT en JSON valide.

## Format de réponse obligatoire
```json
{
  "answer": "Réponse principale",
  "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": ["Question suggérée 1"]
}

Documents

{context}

Question

{query}

Réponse JSON (rien d'autre)

"""

### 2. JSON Mode des LLMs

La plupart des LLMs modernes supportent un "JSON mode" :

```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": "Tu réponds toujours en JSON valide avec les champs: answer, confidence, sources."
        },
        {
            "role": "user",
            "content": f"Context: {context}\n\nQuestion: {query}"
        }
    ]
)

# Garanti d'être du JSON valide
result = json.loads(response.choices[0].message.content)

3. JSON Schema avec validation

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

# Définir le schéma avec Pydantic
class ProductInfo(BaseModel):
    name: str = Field(..., description="Nom du produit")
    price: float = Field(..., ge=0, description="Prix en euros")
    in_stock: bool = Field(..., description="Disponibilité")
    colors: List[str] = Field(default=[], description="Couleurs disponibles")

class RAGResponse(BaseModel):
    answer: str = Field(..., description="Réponse principale")
    confidence: float = Field(..., ge=0, le=1, description="Score de confiance")
    products: List[ProductInfo] = Field(default=[], description="Produits mentionnés")
    sources: List[str] = Field(default=[], description="Sources utilisées")

# Utiliser instructor pour garantir le schema
client = instructor.from_openai(OpenAI())

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

# response est déjà typé et validé
print(response.answer)
print(response.confidence)
for product in response.products:
    print(f"{product.name}: {product.price}€")

4. Function Calling

Utiliser les fonctions pour structurer l'output :

DEVELOPERpython
from openai import OpenAI

client = OpenAI()

tools = [
    {
        "type": "function",
        "function": {
            "name": "provide_answer",
            "description": "Fournit une réponse structurée à la question",
            "parameters": {
                "type": "object",
                "properties": {
                    "answer": {
                        "type": "string",
                        "description": "La réponse à la question"
                    },
                    "confidence": {
                        "type": "number",
                        "minimum": 0,
                        "maximum": 1,
                        "description": "Score de confiance"
                    },
                    "sources": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "Documents sources"
                    },
                    "action_required": {
                        "type": "boolean",
                        "description": "Si une action humaine est requise"
                    }
                },
                "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"}}
)

# Extraire les arguments de la fonction
import json
result = json.loads(response.choices[0].message.tool_calls[0].function.arguments)

Formats de sortie courants

Format JSON pour API

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"]
}

Format tableau comparatif

DEVELOPERpython
COMPARISON_PROMPT = """
Compare les produits mentionnés dans les documents.

## Documents
{context}

## Format de réponse (Markdown)
| Produit | Prix | Stock | Garantie | Note |
|---------|------|-------|----------|------|
| Nom 1   | XX€  | Oui/Non | X ans | X/5 |
| Nom 2   | XX€  | Oui/Non | X ans | X/5 |

## Résumé
[Phrase de recommandation basée sur la comparaison]
"""

def parse_markdown_table(markdown: str) -> list[dict]:
    """Parse un tableau Markdown en liste de dictionnaires."""
    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]]

    # Data rows (skip separator line)
    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

Format action/workflow

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 = """
Analyse la question et détermine la meilleure action.

Actions possibles:
- ANSWER: Répondre directement si l'info est dans les documents
- ESCALATE: Transférer à un humain si complexe ou sensible
- CLARIFY: Demander des précisions si la question est ambiguë
- REDIRECT: Rediriger vers une ressource si hors scope

Documents: {context}
Question: {query}

Réponds en JSON avec: action, content, next_steps, requires_human, confidence
"""

Validation et parsing robuste

Validation avec 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:
        """Génère un output structuré avec retry automatique."""

        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)

        # Valider contre le schema
        self._validate(result)

        return result

    def _extract_json(self, text: str) -> dict:
        """Extrait le JSON d'un texte qui peut contenir autre chose."""
        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:
        """Valide les données contre le 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 avec fallback

DEVELOPERpython
class RobustParser:
    """Parser avec multiples stratégies de fallback."""

    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:
        """Parse format clé: valeur."""
        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:
        """Utilise un LLM pour extraire la structure."""
        prompt = f"""
        Extrait les informations structurées de ce texte en JSON:

        {text}

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

Formats spécialisés par cas d'usage

E-commerce : Fiche produit

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 : Ticket structuré

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 = """
Analyse cette demande client et structure la réponse.

Documents: {context}
Demande: {query}

Réponds en JSON avec:
- summary: Résumé de la demande
- category: billing/technical/shipping/product/other
- priority: low/medium/high/urgent
- resolution: Solution si trouvée
- requires_action: true si action humaine requise
- action_items: Liste des actions à faire
- related_articles: Articles pertinents
- sentiment: positive/neutral/negative
"""

RH : Extraction de politique

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"}
            }
        }
    }
}

Intégration avec Ailog

Ailog supporte nativement les outputs structurés :

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="Compare le MacBook Pro et le Dell XPS",
    output_format=ProductComparison,  # Schema Pydantic
)

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

# Strukturiertes Support-Ticket
ticket = client.chat(
    channel_id="support-widget",
    message="Ma commande 12345 n'est pas arrivée",
    output_format=SupportTicket,
)

if ticket.requires_action:
    create_zendesk_ticket(ticket)

Conclusion

Les outputs structurés transforment votre RAG en outil d'intégration puissant. Points clés :

1. JSON Schema pour garantir la structure
2. Pydantic/instructor pour la validation Python
3. Retry avec fallback pour la robustesse
4. Formats spécialisés par cas d'usage
5. Function calling pour les workflows complexes

Ressources complémentaires

• Introduction au RAG - Fondamentaux
• Génération LLM pour RAG - Guide parent
• Prompt Engineering RAG - Optimiser les prompts
• Streaming RAG - Réponses temps réel

Besoin d'outputs structurés sans complexité ? Essayez Ailog - schemas intégrés, validation automatique, formats e-commerce et support prêts à l'emploi.