Outputs Structurés RAG : JSON, tableaux et formats personnalisés

TL;DR

Les outputs structurés permettent de générer des réponses RAG dans des formats exploitables programmatiquement : JSON, tableaux, listes typées. Cette approche est essentielle pour les intégrations API, les workflows automatisés et les interfaces riches. Ce guide couvre les techniques de génération, validation et parsing des outputs structurés.

Pourquoi les outputs structurés ?

Le problème du texte libre

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

DEVELOPERpython
# ❌ Réponse texte libre (difficile à parser)
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.
"""

# Comment extraire le prix ? La disponibilité ? Les couleurs ?

La solution structurée

DEVELOPERpython
# ✅ Réponse JSON structurée
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')

    # Trouver les lignes du tableau
    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

    # Données spécifiques selon l'action
    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)

        # Tenter le parsing
        try:
            result = json.loads(response)
        except json.JSONDecodeError:
            # Tenter d'extraire le JSON du texte
            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

        # Chercher un bloc JSON
        json_match = re.search(r'```json\s*(.*?)\s*```', text, re.DOTALL)
        if json_match:
            return json.loads(json_match.group(1))

        # Chercher des accolades
        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

        # Fallback final: retourner le texte brut
        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")

# Comparaison de produits structurée
comparison = client.chat(
    channel_id="ecommerce-widget",
    message="Compare le MacBook Pro et le Dell XPS",
    output_format=ProductComparison,  # Schema Pydantic
)

print(comparison.products)  # Liste typée
print(comparison.recommendation)

# Ticket support structuré
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.