Intégration Multi-Outils IA : n8n + LangChain + Claude + Ollama

Vue d'ensemble de l'architecture

Le système combine quatre outils, chacun faisant ce qu'il fait de mieux : n8n gère l'orchestration événementielle (aucun code pour la logique de routage), Ollama exécute les embeddings localement (privé, rapide, gratuit), LangChain gère le pipeline de récupération, et Claude assure la classification par raisonnement via API. Aucun de ces outils ne connaît nativement les autres — FastAPI est la fine couche de glue qui les fait interopérer.

Email client / formulaire support
          │
          ▼
┌─────────────────────┐
│  Webhook n8n        │  POST /webhook/ticket
│  (node déclencheur) │  { text, customer_id, channel }
└─────────┬───────────┘
          │ HTTP POST
          ▼
┌─────────────────────┐
│  FastAPI :8001      │  /classify
│  (couche de glue)   │
└──────┬──────┬───────┘
       │      │
       ▼      ▼
┌──────────┐ ┌─────────────────────┐
│  Ollama  │ │  API Claude Haiku   │
│  :11434  │ │  (classification)   │
│  nomic-  │ │  catégorie, priorité│
│  embed   │ │  sentiment, routage │
└──────────┘ └─────────────────────┘
       │      │
       ▼      │
┌──────────┐  │
│ ChromaDB │  │
│ (tickets │  │
│ similair.)│  │
└──────┬───┘  │
       └──────┘
          │ JSON résultat
          ▼
┌─────────────────────┐
│  Node Switch n8n    │  routing_target
└──┬──────┬──────┬────┘
   ▼      ▼      ▼
Slack  Linear  HubSpot
(ops)  (tech)  (sales)

Prérequis

Ces éléments doivent être installés avant de commencer :

• n8n self-hosted (Docker : docker run -p 5678:5678 n8nio/n8n) ou n8n Cloud
• Python 3.11+ avec pip install langchain langchain-community langchain-anthropic chromadb fastapi uvicorn httpx
• Ollama avec nomic-embed-text téléchargé : ollama pull nomic-embed-text
• Clé API Anthropic (Claude Haiku — modèle le moins cher, idéal pour la classification)

# Installer toutes les dépendances Python
pip install   langchain==0.3.x   langchain-community==0.3.x   langchain-anthropic==0.3.x   chromadb==0.6.x   fastapi==0.115.x   uvicorn==0.34.x   httpx==0.28.x

# Télécharger le modèle d'embedding (147 Mo)
ollama pull nomic-embed-text

# Vérifier qu'Ollama tourne
curl http://localhost:11434/api/tags
# → {"models":[{"name":"nomic-embed-text:latest",...}]}

Étape 1 — Webhook n8n

Le node webhook n8n écoute les tickets support entrants. Quand un client soumet un formulaire ou envoie un email (parsé par le node Email de n8n), n8n le normalise et déclenche le webhook. Créez un nouveau workflow dans n8n avec un node déclencheur Webhook configuré comme suit :

// Paramètres du node Webhook n8n (vue JSON)
{
  "node": "n8n-nodes-base.webhook",
  "parameters": {
    "path": "ticket",
    "httpMethod": "POST",
    "responseMode": "lastNode",
    "options": {
      "rawBody": false
    }
  }
}

// Payload entrant attendu depuis votre formulaire/parseur email :
{
  "text": "Ma facture #4521 affiche un double prélèvement pour mars. Corrigez au plus vite.",
  "customer_id": "cust_8f2a91",
  "channel": "email",
  "customer_tier": "premium"
}

Après le node Webhook, ajoutez un node HTTP Request qui poste vers votre service FastAPI. Définissez l'URL sur http://localhost:8001/classify(ou l'IP de votre VPS si n8n et FastAPI tournent sur des hôtes différents). Définissez le corps avec l'expression {{ $json }} pour transférer l'intégralité du payload webhook.

Étape 2 — LangChain + Embeddings Ollama

Le pipeline d'embedding fait deux choses : (1) embarquer le texte du ticket entrant via le modèle nomic-embed-text d'Ollama, puis (2) rechercher dans ChromaDB les 3 tickets historiques les plus similaires pour contextualiser Claude. Cela améliore considérablement la précision de classification — Claude voit dans quelle catégorie des tickets similaires ont été classés par le passé.

# embedding_service.py
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.schema import Document
import chromadb

# Embeddings Ollama (local, privé, gratuit)
embeddings = OllamaEmbeddings(
    model="nomic-embed-text",
    base_url="http://localhost:11434",
)

# ChromaDB avec persistance
chroma_client = chromadb.PersistentClient(path="./chroma_tickets")

# Wrapper LangChain autour de ChromaDB
vectorstore = Chroma(
    client=chroma_client,
    collection_name="tickets_resolus",
    embedding_function=embeddings,
)


def get_tickets_similaires(texte_ticket: str, k: int = 3) -> list[dict]:
    """Retourne les k tickets résolus les plus similaires avec leurs catégories."""
    resultats = vectorstore.similarity_search_with_score(texte_ticket, k=k)
    return [
        {
            "texte": doc.page_content,
            "categorie": doc.metadata.get("categorie", "inconnu"),
            "priorite": doc.metadata.get("priorite", "moyenne"),
            "score": round(float(score), 3),
        }
        for doc, score in resultats
    ]


def ingerer_ticket_resolu(ticket_id: str, texte: str, categorie: str, priorite: str) -> None:
    """Ajoute un ticket résolu au store vectoriel pour les recherches futures."""
    doc = Document(
        page_content=texte,
        metadata={"ticket_id": ticket_id, "categorie": categorie, "priorite": priorite},
    )
    vectorstore.add_documents([doc])


# Données d'amorçage — à exécuter une seule fois au démarrage
TICKETS_EXEMPLES = [
    ("t001", "Mon abonnement a été prélevé deux fois ce mois-ci", "facturation", "haute"),
    ("t002", "Les limites de taux API nous impactent sur l'endpoint de production", "technique", "critique"),
    ("t003", "Je souhaite passer à l'offre entreprise et discuter des tarifs volume", "commercial", "moyenne"),
    ("t004", "L'email de réinitialisation de mot de passe n'arrive pas", "technique", "moyenne"),
    ("t005", "Puis-je obtenir un remboursement pour la partie non utilisée de mon forfait annuel ?", "facturation", "moyenne"),
    ("t006", "La fonctionnalité d'export CSV renvoie une erreur sur les fichiers de plus de 10 Mo", "technique", "haute"),
]

def initialiser_vectorstore():
    existant = vectorstore.get(limit=1)
    if not existant["ids"]:
        for tid, texte, cat, pri in TICKETS_EXEMPLES:
            ingerer_ticket_resolu(tid, texte, cat, pri)
        print(f"Vectorstore initialisé avec {len(TICKETS_EXEMPLES)} tickets exemples.")

Étape 3 — Classification Claude API

Claude Haiku reçoit le texte du ticket ainsi que les 3 tickets similaires en contexte, puis retourne une classification JSON structurée. L'utilisation de la sortie structurée (via with_structured_output de LangChain) garantit que la réponse est toujours parsable — sans regex, sans acrobaties de prompt.

# classification_service.py
from langchain_anthropic import ChatAnthropic
from langchain.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from typing import Literal
import os


class ClassificationTicket(BaseModel):
    categorie: Literal["facturation", "technique", "commercial", "general"] = Field(
        description="Catégorie principale du ticket"
    )
    priorite: Literal["faible", "moyenne", "haute", "critique"] = Field(
        description="Priorité basée sur l'urgence et le niveau client"
    )
    sentiment: Literal["positif", "neutre", "frustre", "en_colere"] = Field(
        description="Sentiment du client"
    )
    equipe_cible: Literal["equipe_facturation", "equipe_tech", "equipe_commercial", "support_general"] = Field(
        description="Quelle équipe doit traiter ce ticket"
    )
    resume: str = Field(description="Résumé en une phrase du problème, 20 mots max")
    confiance: float = Field(description="Niveau de confiance de la classification entre 0.0 et 1.0")


# Claude Haiku — modèle le plus rapide et le moins cher, idéal pour la classification
llm = ChatAnthropic(
    model="claude-haiku-4-5",
    api_key=os.environ["ANTHROPIC_API_KEY"],
    max_tokens=256,
    temperature=0,  # sortie déterministe pour la classification
)

structured_llm = llm.with_structured_output(ClassificationTicket)

PROMPT_CLASSIFICATION = ChatPromptTemplate.from_messages([
    ("system", """Tu es un classificateur de tickets support. Classe le ticket dans exactement une catégorie
et une équipe cible. Utilise les tickets similaires résolus comme référence pour les patterns de catégorie et priorité.

Tickets résolus similaires (pour contexte) :
{tickets_similaires}

Règles :
- facturation : problèmes de paiement, factures, remboursements, prélèvements abonnement
- technique : bugs, erreurs API, problèmes de performance, dysfonctionnements fonctionnalité
- commercial : montées en gamme, questions tarifaires, démos, demandes entreprise
- general : tout le reste

Déclencheurs d'escalade priorité : double prélèvement → haute, API indisponible → critique, client entreprise → +1 niveau"""),
    ("human", "Classifie ce ticket :

{texte_ticket}"),
])

chaine_classification = PROMPT_CLASSIFICATION | structured_llm


def classifier_ticket(texte_ticket: str, tickets_similaires: list[dict]) -> ClassificationTicket:
    similaires_str = "
".join(
        f"- [{t['categorie']}/{t['priorite']}] {t['texte']} (similarité : {t['score']})"
        for t in tickets_similaires
    )
    return chaine_classification.invoke({
        "texte_ticket": texte_ticket,
        "tickets_similaires": similaires_str or "Aucun ticket similaire trouvé.",
    })

Étape 4 — Couche d'intégration FastAPI

FastAPI relie les deux services en un seul endpoint HTTP que n8n appelle. Il expose également un endpoint /feedback pour ingérer les tickets résolus dans ChromaDB, bouclant la boucle d'auto-amélioration.

# main.py — couche d'intégration FastAPI
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from contextlib import asynccontextmanager
import logging

from embedding_service import get_tickets_similaires, ingerer_ticket_resolu, initialiser_vectorstore
from classification_service import classifier_ticket, ClassificationTicket

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("Initialisation du vectorstore...")
    initialiser_vectorstore()
    logger.info("Prêt.")
    yield

app = FastAPI(title="Classificateur Tickets Support", version="1.0.0", lifespan=lifespan)


class RequeteTicket(BaseModel):
    text: str
    customer_id: str
    channel: str = "email"
    customer_tier: str = "standard"


class ReponseTicket(BaseModel):
    customer_id: str
    classification: ClassificationTicket
    nb_tickets_similaires: int


class RequeteFeedback(BaseModel):
    ticket_id: str
    texte: str
    categorie: str
    priorite: str


@app.post("/classify", response_model=ReponseTicket)
async def classifier(req: RequeteTicket) -> ReponseTicket:
    if not req.text.strip():
        raise HTTPException(status_code=422, detail="Le texte du ticket ne peut pas être vide")

    # 1. Tickets similaires depuis ChromaDB via embeddings Ollama
    similaires = get_tickets_similaires(req.text, k=3)
    logger.info(f"Trouvé {len(similaires)} tickets similaires pour client {req.customer_id}")

    # 2. Classification avec Claude Haiku + contexte
    classification = classifier_ticket(req.text, similaires)

    # 3. Escalader la priorité pour les clients premium
    if req.customer_tier == "premium" and classification.priorite == "moyenne":
        classification.priorite = "haute"

    return ReponseTicket(
        customer_id=req.customer_id,
        classification=classification,
        nb_tickets_similaires=len(similaires),
    )


@app.post("/feedback", status_code=204)
async def feedback(req: RequeteFeedback) -> None:
    """Ingère un ticket résolu dans ChromaDB pour améliorer les recherches futures."""
    ingerer_ticket_resolu(req.ticket_id, req.texte, req.categorie, req.priorite)
    logger.info(f"Ticket résolu ingéré {req.ticket_id} → {req.categorie}/{req.priorite}")


@app.get("/health")
async def health() -> dict:
    return {"status": "ok"}


# Lancer avec : uvicorn main:app --host 0.0.0.0 --port 8001 --workers 2

# Tester l'endpoint en local
curl -X POST http://localhost:8001/classify   -H "Content-Type: application/json"   -d '{
    "text": "Ma facture #4521 affiche un double prélèvement pour mars. Corrigez au plus vite.",
    "customer_id": "cust_8f2a91",
    "channel": "email",
    "customer_tier": "premium"
  }'

# Réponse attendue :
{
  "customer_id": "cust_8f2a91",
  "classification": {
    "categorie": "facturation",
    "priorite": "haute",
    "sentiment": "frustre",
    "equipe_cible": "equipe_facturation",
    "resume": "Client signale double prélèvement sur facture mars #4521",
    "confiance": 0.97
  },
  "nb_tickets_similaires": 3
}

Étape 5 — Logique de routage n8n

De retour dans n8n, le node HTTP Request reçoit le JSON de classification. Un node Switch lit equipe_cible et branche vers l'intégration appropriée. Voici le JSON du workflow n8n pour la section de routage — importez-le directement via l'option "Importer depuis JSON" de n8n.

// Section routage du workflow n8n (JSON prêt à importer)
{
  "nodes": [
    {
      "name": "Classifier le Ticket",
      "type": "n8n-nodes-base.httpRequest",
      "parameters": {
        "method": "POST",
        "url": "http://host.docker.internal:8001/classify",
        "sendBody": true,
        "bodyParameters": {
          "parameters": [
            { "name": "text",          "value": "={{ $json.text }}" },
            { "name": "customer_id",   "value": "={{ $json.customer_id }}" },
            { "name": "channel",       "value": "={{ $json.channel }}" },
            { "name": "customer_tier", "value": "={{ $json.customer_tier ?? 'standard' }}" }
          ]
        },
        "options": { "timeout": 10000 }
      }
    },
    {
      "name": "Router par Équipe",
      "type": "n8n-nodes-base.switch",
      "parameters": {
        "dataType": "string",
        "value1": "={{ $json.classification.equipe_cible }}",
        "rules": {
          "rules": [
            { "value2": "equipe_facturation", "output": 0 },
            { "value2": "equipe_tech",        "output": 1 },
            { "value2": "equipe_commercial",  "output": 2 }
          ]
        },
        "fallbackOutput": 3
      }
    },
    {
      "name": "Notifier Facturation (Slack)",
      "type": "n8n-nodes-base.slack",
      "parameters": {
        "channel": "#support-facturation",
        "text": "=🔴 *{{ $json.classification.priorite.toUpperCase() }}* | {{ $json.classification.resume }}
Client : {{ $json.customer_id }} | Confiance : {{ $json.classification.confiance }}"
      }
    },
    {
      "name": "Créer Issue Linear (Tech)",
      "type": "n8n-nodes-base.linear",
      "parameters": {
        "team": "Engineering",
        "title": "=[Support] {{ $json.classification.resume }}",
        "priority": "={{ $json.classification.priorite === 'critique' ? 1 : $json.classification.priorite === 'haute' ? 2 : 3 }}",
        "description": "={{ $json.customer_id }} : {{ $json.text }}"
      }
    },
    {
      "name": "Deal HubSpot (Commercial)",
      "type": "n8n-nodes-base.hubspot",
      "parameters": {
        "resource": "deal",
        "operation": "create",
        "additionalFields": {
          "dealname": "=Demande upgrade : {{ $json.customer_id }}",
          "pipeline": "pipeline_commercial",
          "dealstage": "rendez_vous_planifie"
        }
      }
    }
  ]
}

Considérations production

Dégradation gracieuse

Encapsulez les appels Ollama et Claude avec des timeouts et des fallbacks. Si Ollama est indisponible, ignorez la recherche de similarité et classifiez sans contexte (la précision chute d'environ 6%). Si Claude renvoie une 5xx, réessayez une fois puis basculez sur un classifieur à mots-clés. Ne laissez jamais la panne d'un service unique perdre un ticket client.

# Fallback gracieux dans classification_service.py
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(2), wait=wait_exponential(multiplier=0.5, max=2))
def classifier_avec_retry(texte: str, similaires: list[dict]) -> ClassificationTicket:
    return classifier_ticket(texte, similaires)

def classifier_avec_fallback(texte: str, similaires: list[dict]) -> ClassificationTicket:
    try:
        return classifier_avec_retry(texte, similaires)
    except Exception:
        # Fallback à règles — coût API nul, ~80% de précision
        t = texte.lower()
        if any(w in t for w in ["facture", "prélèvement", "remboursement", "paiement"]):
            cat, equipe = "facturation", "equipe_facturation"
        elif any(w in t for w in ["erreur", "api", "bug", "plantage", "lent"]):
            cat, equipe = "technique", "equipe_tech"
        elif any(w in t for w in ["upgrade", "tarif", "entreprise", "démo"]):
            cat, equipe = "commercial", "equipe_commercial"
        else:
            cat, equipe = "general", "support_general"
        return ClassificationTicket(
            categorie=cat, priorite="moyenne", sentiment="neutre",
            equipe_cible=equipe, resume="[Classifieur de secours]", confiance=0.6
        )

Déploiement Docker Compose

# docker-compose.yml
services:
  fastapi:
    build: .
    ports: ["8001:8001"]
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
    volumes:
      - ./chroma_tickets:/app/chroma_tickets
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8001/health"]
      interval: 30s
      timeout: 5s
      retries: 3

  ollama:
    image: ollama/ollama:latest
    ports: ["11434:11434"]
    volumes:
      - ollama_models:/root/.ollama
    restart: unless-stopped

volumes:
  ollama_models:

Analyse des coûts

Cette architecture est conçue pour la frugalité. Voici le coût mensuel réel pour trois scénarios de volume :

Comparaison : l'extension IA Zendesk à 50 €/agent/mois × 5 agents = 250 €/mois, quel que soit le volume de tickets. À 100 000 tickets/mois, cette stack open source coûte 86% moins cher tout en vous donnant un contrôle total sur la logique de routage, la résidence des données et la sélection du modèle.

FAQ