Dify Chatflow ne répond qu’à la fin : diagnostiquer Agent, Fail Branch et Answer

Une réponse Dify Chatflow qui apparaît d’un seul bloc à la fin ne prouve pas que le modèle amont ne diffuse plus en continu. Elle ne prouve pas non plus qu’un changement de fournisseur résoudra le problème. Gardez un prompt court, un modèle et une configuration Agent identiques, puis comparez quatre horloges : premier token amont, événements SSE, agent_log et fragments message réellement visibles par l’utilisateur.

Le but est de trouver la couche qui attend ou met le texte en tampon. Il n’est pas nécessaire de supprimer la gestion des erreurs en production : la désactivation de Fail Branch sert seulement de contrôle dans une copie de test.

Repérez la couche grâce au motif des événements

Observation	Première piste	Vérification suivante
Le contrôle LLM → Answer ne fournit pas non plus de texte rapidement	Modèle, provider, passerelle ou paramètres	Vérifiez une courte requête texte et le streaming du modèle
workflow_started, node_started et les autres événements arrivent tous tard	Proxy, passerelle ou transport SSE	Répétez exactement la requête directement vers Dify et recherchez une mise en tampon
agent_log arrive pendant l’exécution, mais message n’arrive qu’en rafale finale	Chemin Agent → Answer ; Fail Branch comme variable de contrôle	Conservez le fallback de production et comparez les temps dans un flux cloné
message ne commence qu’après la fin d’une autre variable	Variable antérieure non résolue dans Answer	Réordonnez les variables ou livrez les étapes avec plusieurs nœuds Answer

La documentation d’Answer indique que le nœud diffuse le contenu selon la disponibilité des variables. Il émet jusqu’à la première variable non résolue, attend celle-ci, puis reprend. L’ordre des variables dans Answer, et non l’ordre d’exécution des nœuds amont, détermine donc ce que l’utilisateur voit en premier.

Exécutez quatre contrôles en ne changeant qu’un élément

Dupliquez le Chatflow existant pour les essais. Notez sa version publiée, son Model ID, la temperature, la stratégie Agent, les outils, le nombre maximal d’itérations, la mémoire et le mode de gestion des erreurs. Utilisez exactement le même prompt court à chaque exécution, par exemple :

Écris 12 contrôles numérotés pour diagnostiquer une réponse en streaming, avec une seule phrase par point.

La réponse doit être assez longue pour produire un intervalle mesurable, sans contenir de données utilisateur réelles.

Avant chacun des contrôles A, B, C et D, répétez la même séquence : appliquez uniquement la modification de ce tour, publiez l’état actuel de la copie de test, notez un libellé d’exécution et la version publiée du workflow, puis utilisez le Service API Base affiché dans API Access pour cette copie (le chemin complet se termine généralement par /v1) avec son propre App API Key. Ne réutilisez pas la clé de l’application d’origine. Sans workflow_id, Dify exécute la dernière version publiée ; appeler avant de publier peut donc faire tester la même ancienne version plusieurs fois.

Contrôle A : LLM → Answer uniquement

Dans la copie, envoyez la sortie du même modèle depuis LLM directement vers Answer, sans Agent ni outils. Dify précise que les nœuds LLM diffusent par défaut. Si message arrive encore seulement à la fin, n’accusez pas encore Agent : vérifiez d’abord le provider, la passerelle OpenAI-compatible, le proxy inverse et l’ordre des variables Answer.

Contrôle B : Agent actuel avec Fail Branch conservé

Rétablissez le chemin Agent réel et le fallback de production. C’est la référence. Ne retirez pas la reprise uniquement pour accélérer l’affichage : Fail Branch lance un flux distinct après une erreur, par exemple pour journaliser, notifier, essayer une autre approche ou appeler un service de secours.

Contrôle C : désactivez Fail Branch seulement dans la copie

N’effectuez ce test que si votre nœud Agent actuel expose la même option. Remettez la gestion des erreurs à sa valeur par défaut dans la copie, sans modifier modèle, prompt, outils ni autres paramètres. Ce résultat indique seulement si la branche est corrélée au délai dans votre environnement ; il ne prouve pas que la production peut fonctionner sans fallback.

Un rapport d’utilisateur sur Dify 1.17.1, langgenius/agent 0.0.33 et graphon 0.7.0 a enregistré 805 événements message entre 29,819 et 30,035 secondes avec Fail Branch, soit 0,216 seconde. Avec la gestion par défaut, 866 messages se sont étalés de 3,397 à 40,991 secondes. Des événements agent_log arrivaient pendant la même exécution. Il s’agit d’un environnement rapporté, pas d’une conclusion universelle ni d’une version corrigée confirmée.

Contrôle D : modifiez uniquement l’ordre d’Answer

Commencez avec un Answer qui référence seulement le text d’Agent, puis réintroduisez les autres variables une par une. Si une variable lente précède le texte Agent, Answer peut attendre alors qu’Agent génère déjà. Placez la variable textuelle en premier ou utilisez plusieurs nœuds Answer, officiellement pris en charge pour livrer du contenu à différentes étapes.

Enregistrez l’heure d’arrivée avec un appel API minimal

Appelez l’API publiée du Chatflow au lieu de juger uniquement l’animation du navigateur. Le guide de streaming Dify définit response_mode: "streaming" comme une réponse SSE. Pour Chatflow, la réponse visible est la concaténation ordonnée des événements message. agent_log est un journal informatif d’étape Agent : sa réception prouve que le canal d’événements fonctionne, pas que le texte est arrivé à l’utilisateur.

À chaque exécution, prenez le Service API Base et l’App API Key dans API Access de la copie de test. Saisissez le Base complet, normalement avec /v1 ; le script ajoute /chat-messages. Si le verrouillage de version est disponible, placez l’ID de la version publiée de ce tour dans DIFY_WORKFLOW_ID. Laissez vide uniquement pour utiliser volontairement la dernière version publiée. Dans tous les cas, publiez d’abord et inscrivez la version dans DIFY_RUN_LABEL.

Utilisez {} pour DIFY_TEST_INPUTS_JSON seulement si la copie n’a aucune entrée personnalisée obligatoire. Si Start/User Input définit des champs requis, envoyez aux quatre tours les mêmes valeurs JSON expurgées avec les noms et types exacts des paramètres.

Dans Bash, lisez l’App API Key sans l’afficher et saisissez les paramètres du tour :

unset DIFY_TARGET_NODE_ID DIFY_FALLBACK_NODE_ID DIFY_EXPECTED_FALLBACK
read -r DIFY_API_BASE && export DIFY_API_BASE
read -rs DIFY_API_KEY && export DIFY_API_KEY
read -r DIFY_TEST_QUERY && export DIFY_TEST_QUERY
read -r DIFY_RUN_LABEL && export DIFY_RUN_LABEL
read -r DIFY_WORKFLOW_ID && export DIFY_WORKFLOW_ID
read -r DIFY_TEST_INPUTS_JSON && export DIFY_TEST_INPUTS_JSON
python3 capture_dify_sse.py

Enregistrez ce qui suit sous capture_dify_sse.py :

import json
import os
import sys
import time
from collections import defaultdict
from http.client import IncompleteRead
from pathlib import Path
from urllib import error, request


def redacted(value):
    return None if value is None else f"<redacted:{len(str(value))} chars>"


def load_inputs():
    raw = os.environ.get("DIFY_TEST_INPUTS_JSON", "{}").strip() or "{}"
    try:
        value = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise SystemExit(f"DIFY_TEST_INPUTS_JSON must be valid JSON: {exc.msg}") from exc
    if not isinstance(value, dict):
        raise SystemExit("DIFY_TEST_INPUTS_JSON must decode to a JSON object")
    return value


api_base = os.environ["DIFY_API_BASE"].rstrip("/")
api_key = os.environ["DIFY_API_KEY"]
test_query = os.environ["DIFY_TEST_QUERY"]
raw_label = os.environ.get("DIFY_RUN_LABEL", "unnamed-run").strip() or "unnamed-run"
run_label = "".join(
    char if char.isalnum() or char in "-_" else "_" for char in raw_label
)[:80] or "unnamed-run"
workflow_id = os.environ.get("DIFY_WORKFLOW_ID", "").strip()
target_node_id = os.environ.get("DIFY_TARGET_NODE_ID", "").strip()
fallback_node_id = os.environ.get("DIFY_FALLBACK_NODE_ID", "").strip()
expected_fallback = os.environ.get("DIFY_EXPECTED_FALLBACK", "")

body = {
    "inputs": load_inputs(),
    "query": test_query,
    "response_mode": "streaming",
    "user": "streaming-probe",
}
if workflow_id:
    body["workflow_id"] = workflow_id

req = request.Request(
    f"{api_base}/chat-messages",
    data=json.dumps(body).encode(),
    headers={
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    },
    method="POST",
)

started = time.perf_counter()
stats = defaultdict(list)
answer_parts = []
message_end_seen = False
workflow_finished_seen = False
workflow_status = None
terminal_event = None
fatal_reason = None
target_node_failed = None if not target_node_id else False
fallback_node_succeeded = None if not fallback_node_id else False
top_error = {"status": None, "code": None, "message_redacted": None}

log_path = Path(f"{run_label}-events.jsonl")
summary_path = Path(f"{run_label}-summary.json")

with log_path.open("w", encoding="utf-8") as log_file:
    try:
        with request.urlopen(req, timeout=180) as response:
            event_name = None
            for line_no, raw_line in enumerate(response, 1):
                line = raw_line.decode("utf-8", errors="replace").strip()
                if line.startswith("event:"):
                    event_name = line.split(":", 1)[1].strip()
                    continue
                if not line.startswith("data:"):
                    continue

                elapsed = round(time.perf_counter() - started, 3)
                try:
                    event = json.loads(line.split(":", 1)[1].strip())
                except json.JSONDecodeError:
                    fatal_reason = f"json_parse_error_at_stream_line_{line_no}"
                    record = {
                        "elapsed_s": elapsed,
                        "event": "parse_error",
                        "detail": "<redacted:invalid JSON payload>",
                    }
                    log_file.write(json.dumps(record, ensure_ascii=False) + "\n")
                    log_file.flush()
                    stats["parse_error"].append(elapsed)
                    print(f"{elapsed:8.3f}s parse_error")
                    break

                kind = event.get("event") or event_name or "unknown"
                event_name = None
                data = event.get("data") if isinstance(event.get("data"), dict) else {}
                answer = event.get("answer") if isinstance(event.get("answer"), str) else ""
                answer_parts.append(answer)

                node_id = data.get("node_id")
                data_status = data.get("status")
                error_status = event.get("status") if kind == "error" else None
                error_code = event.get("code") if kind == "error" else None
                error_message = redacted(event.get("message")) if kind == "error" else None
                record = {
                    "elapsed_s": elapsed,
                    "event": kind,
                    "answer_chars": len(answer),
                    "task_id": event.get("task_id"),
                    "workflow_run_id": event.get("workflow_run_id"),
                    "node_id": node_id,
                    "node_title": data.get("title"),
                    "data_status": data_status,
                    "error_status": error_status,
                    "error_code": error_code,
                    "error_message_redacted": error_message,
                }
                log_file.write(json.dumps(record, ensure_ascii=False) + "\n")
                log_file.flush()
                stats[kind].append(elapsed)

                if kind == "message_end":
                    message_end_seen = True
                elif kind == "workflow_finished":
                    workflow_finished_seen = True
                    workflow_status = data_status
                    terminal_event = kind
                elif kind == "workflow_paused":
                    workflow_status = "paused"
                    terminal_event = kind
                elif kind == "error":
                    top_error = {
                        "status": error_status,
                        "code": error_code,
                        "message_redacted": error_message,
                    }
                    terminal_event = kind

                if kind == "node_finished" and node_id == target_node_id:
                    target_node_failed = data_status in {"failed", "exception"}
                if kind == "node_finished" and node_id == fallback_node_id:
                    fallback_node_succeeded = data_status == "succeeded"

                print(
                    f"{elapsed:8.3f}s {kind:<22} answer_chars={len(answer)} "
                    f"data_status={data_status} error_status={error_status} "
                    f"error_code={error_code} error_message={error_message}"
                )
    except KeyboardInterrupt:
        fatal_reason = "interrupted_by_user"
    except error.HTTPError as exc:
        fatal_reason = f"http_error_{exc.code}"
        top_error = {
            "status": exc.code,
            "code": "http_error",
            "message_redacted": redacted(exc.reason),
        }
    except error.URLError as exc:
        fatal_reason = "connection_error"
        top_error = {
            "status": None,
            "code": "connection_error",
            "message_redacted": redacted(exc.reason),
        }
    except (IncompleteRead, TimeoutError, OSError) as exc:
        fatal_reason = type(exc).__name__
        top_error = {
            "status": None,
            "code": type(exc).__name__,
            "message_redacted": redacted(exc),
        }

if fatal_reason:
    run_outcome, incomplete_reason = "incomplete", fatal_reason
elif workflow_status == "succeeded" and message_end_seen and workflow_finished_seen:
    run_outcome, incomplete_reason = "succeeded", None
elif terminal_event == "workflow_paused":
    run_outcome, incomplete_reason = "paused", None
elif workflow_status in {"failed", "partial-succeeded", "stopped"}:
    run_outcome, incomplete_reason = workflow_status, None
elif terminal_event == "error":
    run_outcome, incomplete_reason = "failed", None
elif workflow_finished_seen:
    run_outcome = "unknown_terminal_status"
    incomplete_reason = "workflow_finished_without_recognized_status"
else:
    run_outcome = "incomplete"
    incomplete_reason = "stream_ended_without_terminal_event"

expected_fallback_seen = (
    None if not expected_fallback else expected_fallback in "".join(answer_parts)
)
complete_for_timing_acceptance = run_outcome == "succeeded"
failure_validation_requested = any(
    (target_node_id, fallback_node_id, expected_fallback)
)
failure_validation_configured = all(
    (target_node_id, fallback_node_id, expected_fallback)
)
failure_validation_passed = (
    None
    if not failure_validation_requested
    else failure_validation_configured
    and run_outcome in {"succeeded", "partial-succeeded"}
    and all(
        value is not False
        for value in (
            target_node_failed,
            fallback_node_succeeded,
            expected_fallback_seen,
        )
    )
)
acceptance_checks_passed = (
    failure_validation_passed
    if failure_validation_requested
    else complete_for_timing_acceptance
)

summary = {
    "run_label": run_label,
    "requested_workflow_id": workflow_id or None,
    "version_selection": "pinned-published-version" if workflow_id else "latest-published-version",
    "run_outcome": run_outcome,
    "workflow_status": workflow_status,
    "terminal_event": terminal_event,
    "message_end_seen": message_end_seen,
    "workflow_finished_seen": workflow_finished_seen,
    "incomplete_reason": incomplete_reason,
    "top_level_error": top_error,
    "target_node_failed": target_node_failed,
    "fallback_node_succeeded": fallback_node_succeeded,
    "expected_fallback_seen": expected_fallback_seen,
    "complete_for_timing_acceptance": complete_for_timing_acceptance,
    "failure_validation_requested": failure_validation_requested,
    "failure_validation_configured": failure_validation_configured,
    "failure_validation_passed": failure_validation_passed,
    "acceptance_checks_passed": acceptance_checks_passed,
    "event_timings": {
        kind: {
            "count": len(times),
            "first_s": times[0],
            "last_s": times[-1],
            "span_s": round(times[-1] - times[0], 3),
        }
        for kind, times in sorted(stats.items())
    },
}
summary_path.write_text(
    json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
print(json.dumps(summary, ensure_ascii=False, indent=2))
if not acceptance_checks_passed:
    sys.exit(2)

Le script n’écrit jamais le texte de réponse sur disque. Il enregistre les temps, l’ID/titre du nœud, data.status, la longueur du texte et, pour un événement error, les champs supérieurs status et code, en remplaçant message par un marqueur de longueur. Il distingue succeeded, failed, partial-succeeded, stopped, paused, un error supérieur et un flux sans événement terminal.

Les noms de fichiers utilisent DIFY_RUN_LABEL. Si la connexion est interrompue ou si le JSON SSE ne peut pas être analysé, les temps déjà capturés sont conservés, mais run_outcome vaut incomplete et complete_for_timing_acceptance vaut false. Ces intervalles ne servent qu’au diagnostic. Un contrôle normal ne passe l’acceptation temporelle qu’après message_end et workflow_finished.data.status=succeeded. L’exception http.client.IncompleteRead provoquée par la troncature d’un flux SSE en transfert chunked HTTP/1.1 est également interceptée : les temps des événements reçus restent dans le résumé, acceptance_checks_passed=false, et le corps partiel de la réponse contenu dans l’exception n’est pas affiché.

Comparez quatre temps

1. Premier token amont. Enregistrez la métrique du provider ou de la passerelle si elle existe. Sinon, le contrôle LLM → Answer isole pratiquement ce trajet.
2. Premier événement SSE de données. Un workflow_started précoce confirme que l’exécution a été acceptée, pas que le texte est diffusé progressivement.
3. Premier agent_log. Des journaux précoces montrent que les étapes Agent traversent la connexion. Si les journaux arrivent alors que le texte attend, tout le canal n’est probablement pas retenu en un seul objet.
4. Premier et dernier message. Ce sont les temps qui comptent pour l’utilisateur. Comparez first_s, last_s et span_s. De nombreux événements concentrés dans un intervalle minuscule mesurent la rafale finale.

Un Chatflow réussi doit émettre message_end, puis workflow_finished avec data.status=succeeded. Le contrat SSE officiel énumère succeeded, failed, partial-succeeded et stopped pour workflow_finished.data.status ; consignez-les séparément. Une pause pour saisie humaine émet en revanche human_input_required, puis l’événement distinct workflow_paused, à partir duquel le script enregistre run_outcome=paused. Ne transposez pas l’énumération des statuts d’un enregistrement d’exécution au contrat de l’événement workflow_finished. Dans un événement error, le statut, le type et le message sont au niveau supérieur ; le script affiche status, code et un message expurgé. Si la connexion se termine sans événement final ou si le transport/l’analyse JSON est interrompu, le résumé devient incomplete. Même avec des message capturés, leur intervalle ne prouve aucune amélioration tant que complete_for_timing_acceptance n’est pas true.

Choisissez la modification correspondant au résultat

Le contrôle LLM attend aussi : corrigez d’abord l’amont

Vérifiez Model ID, endpoint, comportement streaming du provider et passerelles intermédiaires avant de revenir à Agent. Si BetterToken sert de référence OpenAI-compatible facultative, son guide Dify recommande de valider d’abord un court texte sans outils, puis d’activer les tool calls uniquement si le modèle et l’endpoint choisis les prennent en charge. Cela isole la couche modèle, sans garantir l’absence de tampon dans Dify.

Tous les événements arrivent tard : inspectez le transport

Répétez la même requête avec moins de couches CDN, proxy inverse ou API gateway. Si la connexion directe à Dify diffuse progressivement et qu’un intermédiaire regroupe à nouveau les événements, examinez la transmission SSE et le buffering sur cette couche. Ne changez pas aussi le modèle.

Seul Fail Branch regroupe les message

Limitez la conclusion à votre version Dify, votre plugin Agent et votre configuration. Conservez par défaut le fallback de production. Si la diffusion en direct est prioritaire, concevez et validez une autre reprise avant de migrer ; un seul contrôle ne justifie pas la suppression des alertes, services de secours ou journaux d’erreur.

Avant une mise à niveau de Dify ou du plugin Agent, conservez une exécution normale complète et une vérification de panne contrôlée. Après la mise à niveau, publiez la copie et répétez avec le même prompt, modèle, inputs expurgés et procédé de consignation de version. Une simple chronologie d’échec ne suffit pas : le nœud cible doit réellement échouer, le Fail Branch et l’Answer de secours correspondants doivent s’exécuter, et l’utilisateur doit recevoir un fallback expurgé.

Le réordonnancement d’Answer rétablit le streaming

Le modèle amont a peut-être streamé depuis le début tandis qu’Answer attendait une variable antérieure. Conservez le nouvel ordre ou les Answer par étapes, puis exécutez la vérification ci-dessous. error_type et error_message peuvent servir au routage et aux journaux internes, mais l’Answer visible doit employer un texte expurgé plutôt que les détails bruts.

Vérifiez une panne contrôlée uniquement dans la copie de test

Privilégiez une condition existante, répétable et sans données réelles qui fasse émettre au nœud cible exact un résultat anormal dans node_finished ; une erreur traitée par Fail Branch peut avoir data.status=exception, il ne faut donc pas exiger failed. Une erreur interne d’outil, une entrée agent_log ou un journal de modèle ne prouvent pas à eux seuls l’échec du nœud Agent.

Dans le code de ces versions, graphon 0.7.0 transforme un échec traité par FAIL_BRANCH en NodeRunExceptionEvent avec le statut exception. Dify 1.17.1 l’émet sous la forme node_finished.data.status=exception, tandis qu’un échec de nœud non traité a le statut failed. Le champ target_node_failed du script accepte exception ou failed comme preuve d’une anomalie du nœud cible lui-même, et non d’un échec de tout le workflow ou d’un repli réussi. La validation exige toujours la réussite du nœud de repli correspondant, une réponse de repli expurgée visible par l’utilisateur et l’un des résultats finaux autorisés ci-dessous.

S’il n’existe pas de condition adaptée, ajoutez temporairement un nœud HTTP Request sur un chemin de diagnostic de la copie, pointez-le vers https://fail-branch-check.invalid/ et utilisez un timeout court. .invalid est réservé aux noms nécessairement invalides. Réglez la gestion d’erreur de ce nœud sur Fail Branch et reliez la branche orange à l’Answer de secours. Cette méthode vérifie le câblage Fail Branch/Answer du nœud HTTP temporaire ; elle ne remplace pas un échec du nœud Agent lorsque celui-ci est la cible de l’acceptation.

1. Publiez la copie après la modification temporaire. Notez la version publiée, l’ID du nœud cible, l’ID de l’Answer de secours et une courte phrase non sensible qui n’apparaît que dans le fallback.
2. Utilisez le Base et la Key de la copie, les mêmes inputs expurgés et cette version. Définissez les trois valeurs puis lancez le même script :

read -r DIFY_TARGET_NODE_ID && export DIFY_TARGET_NODE_ID
read -r DIFY_FALLBACK_NODE_ID && export DIFY_FALLBACK_NODE_ID
read -r DIFY_EXPECTED_FALLBACK && export DIFY_EXPECTED_FALLBACK
python3 capture_dify_sse.py

3. Le résumé doit montrer simultanément target_node_failed: true, fallback_node_succeeded: true, expected_fallback_seen: true et failure_validation_passed: true. Le workflow peut finir en succeeded ou partial-succeeded, mais pas en failed, stopped, paused ou incomplete.
4. Supprimez le nœud temporaire ou restaurez la condition d’origine, les connexions et la configuration normales, puis republiez. Effacez les trois variables et relancez la requête normale. Cette exécution restaurée doit afficher complete_for_timing_acceptance: true avant de comparer à nouveau les premier et dernier message.

N’acceptez la correction que si le streaming et le fallback passent

N’acceptez pas la correction uniquement parce que le premier fragment arrive plus vite. Conservez deux résumés : une requête normale complète où les message sont répartis pendant la génération, et une panne contrôlée prouvant l’échec du nœud cible, l’exécution du Fail Branch/Answer de secours et l’affichage d’un fallback expurgé. Une erreur d’outil ou un log Agent seul, l’absence d’échec du nœud, l’absence de fallback visible ou l’absence de nouvelle exécution normale après restauration entraînent un refus.