Dify Chatflow responde solo al final: diagnostica Agent, Fail Branch y Answer

Que una respuesta de Dify Chatflow aparezca completa al final no demuestra que el modelo haya dejado de transmitir. Tampoco significa que cambiar de proveedor vaya a solucionarlo. Mantén fijos un prompt corto, un modelo y la configuración de Agent, y compara cuatro relojes: primer token aguas arriba, eventos SSE, agent_log y fragmentos message visibles para el usuario.

La pregunta útil es dónde se espera o se almacena el texto. No hace falta retirar el manejo de errores de producción: desactivar Fail Branch sirve únicamente como control en una copia de prueba.

Reconoce la capa por el patrón de eventos

Lo que observas	Primera sospecha	Siguiente comprobación
El control LLM → Answer tampoco entrega texto temprano	Modelo, provider, gateway o parámetros	Verifica una petición breve de texto y el streaming del modelo
workflow_started, node_started y los demás eventos llegan tarde	Proxy, gateway o transporte SSE	Repite la misma petición directamente contra Dify y busca buffering
agent_log llega durante la ejecución, pero message aparece en una ráfaga final	Ruta de texto Agent → Answer; Fail Branch como variable de control	Conserva el fallback de producción y compara tiempos en un flujo clonado
message empieza solo cuando termina otra variable	Una variable anterior no resuelta en Answer	Reordena variables o entrega etapas con varios nodos Answer

La documentación de Answer explica que el nodo emite contenido según la disponibilidad de las variables. Avanza hasta la primera variable sin resolver, espera y continúa después. Por eso importa el orden escrito en Answer, no el orden en que terminen los nodos anteriores.

Ejecuta cuatro controles cambiando una sola cosa

Duplica el Chatflow existente para las pruebas. Anota la versión publicada, Model ID, temperature, estrategia de Agent, herramientas, máximo de iteraciones, memoria y modo de manejo de errores. Usa exactamente el mismo prompt corto en cada ejecución, por ejemplo:

Escribe 12 comprobaciones numeradas para depurar una respuesta en streaming, una sola frase por punto.

Debe producir una respuesta suficientemente larga para medir un intervalo, sin incluir datos reales de usuarios.

Antes de los controles A, B, C y D, repite siempre la misma secuencia: aplica únicamente el cambio de esa ronda, publica el estado actual de la copia de prueba, registra una etiqueta de ejecución y la versión publicada del workflow, y llama al Service API Base mostrado en API Access de esa copia (la ruta completa suele terminar en /v1) con su propio App API Key. No reutilices la clave de la aplicación original. Si omites workflow_id, Dify ejecuta la última versión publicada; llamar antes de publicar puede hacer que varias rondas prueben la misma versión anterior.

Control A: solo LLM → Answer

En la copia, conecta la salida del mismo modelo desde LLM directamente con Answer, sin Agent ni herramientas. Dify documenta que los nodos LLM transmiten por defecto. Si message sigue llegando únicamente al final, no culpes todavía a Agent: revisa primero el provider, el gateway OpenAI-compatible, el proxy inverso y el orden de variables en Answer.

Control B: Agent actual con Fail Branch

Restaura la ruta real de Agent y el fallback utilizado en producción. Esta es la línea base. No elimines la recuperación solo para mejorar la animación: Fail Branch existe para ejecutar una ruta separada tras un fallo, como registrar, avisar, probar otro enfoque o llamar a un servicio de respaldo.

Control C: desactiva Fail Branch solo en la copia

Hazlo únicamente si tu nodo Agent actual ofrece esa misma opción. Devuelve el manejo de errores al valor predeterminado en la copia y conserva modelo, prompt, herramientas y parámetros. El resultado solo indica si la rama se correlaciona con la espera en tu entorno; no demuestra que producción pueda quedarse sin fallback.

Un informe de usuario sobre Dify 1.17.1, langgenius/agent 0.0.33 y graphon 0.7.0 registró 805 eventos message entre 29,819 y 30,035 segundos con Fail Branch, un intervalo de 0,216 segundos. En el control predeterminado, 866 mensajes se distribuyeron de 3,397 a 40,991 segundos. Durante la misma ejecución sí llegaron eventos agent_log. Es evidencia de un entorno concreto, no una conclusión universal ni una versión corregida confirmada.

Control D: cambia solo el orden de Answer

Primero deja que Answer use únicamente el text de Agent. Después añade las demás variables una a una. Si una variable lenta aparece antes que el texto, Answer puede esperar aunque Agent ya esté generando. Coloca antes la variable que transmite texto o utiliza varios nodos Answer, admitidos oficialmente para entregar contenido en distintas etapas.

Registra la llegada real con una llamada mínima

Llama a la API publicada del Chatflow en lugar de juzgar solo la animación del navegador. La guía de streaming de Dify define response_mode: "streaming" como una respuesta SSE. En Chatflow, la respuesta visible se obtiene concatenando los eventos message en orden. agent_log es un registro informativo de un paso del Agent: demuestra que el canal de eventos funciona, no que el texto ya llegó al usuario.

En cada ejecución toma el Service API Base y el App API Key de API Access de la copia de prueba. Introduce el Base completo, normalmente con /v1; el script añade /chat-messages. Si puedes fijar versión, pon el ID de la versión publicada de esa ronda en DIFY_WORKFLOW_ID. Déjalo vacío solo cuando quieras usar deliberadamente la última versión publicada. En ambos casos publica primero y anota la versión en DIFY_RUN_LABEL.

Usa {} en DIFY_TEST_INPUTS_JSON solo si la copia no tiene entradas personalizadas obligatorias. Si Start/User Input define campos requeridos, envía en las cuatro rondas los mismos valores JSON anonimizados con los nombres y tipos exactos de los parámetros.

En Bash, lee el App API Key sin mostrarlo e introduce los parámetros de la ronda:

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

Guarda lo siguiente como 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)

El script no escribe el texto de la respuesta en disco. Registra tiempos, ID/título de nodo, data.status, longitud del texto y, para un evento error, los campos superiores status y code, sustituyendo message por un marcador con su longitud. Distingue succeeded, failed, partial-succeeded, stopped, paused, un error superior y una transmisión sin evento terminal.

Los nombres de archivo usan DIFY_RUN_LABEL. Si se interrumpe la conexión o falla el análisis JSON de SSE, conserva los tiempos capturados, pero marca run_outcome como incomplete y complete_for_timing_acceptance como false. Esos intervalos solo sirven para diagnóstico. Un control normal pasa la aceptación temporal únicamente tras observar message_end y workflow_finished.data.status=succeeded. También se captura http.client.IncompleteRead cuando se trunca un SSE con transferencia chunked en HTTP/1.1: los tiempos de los eventos recibidos permanecen en el resumen, acceptance_checks_passed=false y no se muestra el cuerpo parcial de la respuesta contenido en la excepción.

Compara cuatro tiempos

1. Primer token aguas arriba. Registra la métrica del provider o gateway si existe. Si no, el control LLM → Answer sirve para aislar de forma práctica esa ruta.
2. Primer evento SSE con datos. Un workflow_started temprano confirma que la ejecución fue aceptada, no que la respuesta se entregue por fragmentos.
3. Primer agent_log. Si los registros llegan pronto, los pasos de Agent cruzan la conexión. Si el texto espera, probablemente no se está reteniendo todo el canal como un único objeto.
4. Primer y último message. Son los tiempos relevantes para el usuario. Compara first_s, last_s y span_s. Muchos mensajes comprimidos en un intervalo mínimo prueban la ráfaga final.

Un Chatflow correcto debe emitir message_end y después workflow_finished con data.status=succeeded. El contrato oficial de SSE enumera succeeded, failed, partial-succeeded y stopped para workflow_finished.data.status; regístralos por separado. Una pausa para solicitar datos al usuario emite, en cambio, human_input_required y después el evento independiente workflow_paused, a partir del cual el script registra run_outcome=paused. No traslades la enumeración de estados de un registro de ejecución al contrato del evento workflow_finished. En un evento error, el estado, el tipo y el mensaje están en el nivel superior; el script muestra status, code y un message anonimizado. Si la conexión termina sin evento final o se interrumpe el transporte/análisis JSON, el resumen queda incomplete. Aunque haya algunos message, su intervalo no demuestra una mejora salvo que complete_for_timing_acceptance sea true.

Aplica el cambio que corresponda al resultado

El control LLM también espera: corrige primero la capa upstream

Comprueba Model ID, endpoint, streaming del provider y gateways intermedios antes de volver a Agent. Si usas BetterToken como línea base opcional OpenAI-compatible, su guía para Dify recomienda validar primero una petición breve de texto sin herramientas y activar tool calls solo cuando el modelo y el endpoint lo soporten. Esto aísla la capa del modelo; no garantiza que Dify no almacene fragmentos.

Todos los eventos llegan tarde: inspecciona el transporte

Repite la misma petición reduciendo CDN, proxy inverso y gateways adicionales. Si la conexión directa a Dify transmite de forma progresiva y un intermediario vuelve a agrupar eventos, revisa el reenvío SSE y el buffering en esa capa. No cambies también el modelo.

Solo Fail Branch agrupa los message

Limita la conclusión a tu versión de Dify, plugin Agent y configuración. Conserva por defecto el fallback de producción. Si la entrega en vivo tiene más prioridad, diseña y prueba otra ruta de recuperación antes de migrar; una prueba de control no justifica borrar alertas, servicios de respaldo o registros de error.

Antes de actualizar Dify o el plugin Agent, conserva una ejecución normal completa y una verificación de fallo controlado. Después de actualizar, publica la copia de prueba y repite con el mismo prompt, modelo, inputs anonimizados y método de registro de versión. Guardar una cronología de fallo no basta: el nodo objetivo debe fallar realmente, deben ejecutarse su Fail Branch y el Answer de respaldo, y el usuario debe recibir un mensaje de respaldo anonimizado.

Reordenar Answer recupera el streaming

El modelo upstream pudo estar enviando streaming todo el tiempo mientras Answer esperaba una variable anterior. Conserva el nuevo orden o los Answer por etapas y ejecuta la verificación de fallo siguiente. error_type y error_message pueden usarse para enrutar y registrar internamente, pero el Answer visible debe usar texto anonimizado, no detalles crudos.

Verifica un fallo controlado solo en la copia de prueba

Prefiere una condición existente y repetible, sin datos reales, que haga que el nodo objetivo exacto emita un resultado anómalo en node_finished; un error gestionado por Fail Branch puede tener data.status=exception, por lo que no se debe exigir failed. Un error interno de una herramienta, un agent_log o un registro del modelo no prueban por sí solos que el nodo Agent haya fallado.

En el código de estas versiones, graphon 0.7.0 convierte un fallo gestionado mediante FAIL_BRANCH en NodeRunExceptionEvent con estado exception. Dify 1.17.1 lo emite como node_finished.data.status=exception, mientras que un fallo de nodo no gestionado tiene estado failed. El campo target_node_failed del script acepta exception o failed como evidencia de un error en el propio nodo objetivo, no de un fallo de todo el workflow ni de un fallback correcto. La validación sigue exigiendo que el nodo de fallback correspondiente termine correctamente, que el usuario vea una respuesta de fallback sanitizada y que se alcance uno de los resultados finales permitidos que se indican abajo.

Si no existe una condición adecuada, añade temporalmente un nodo HTTP Request en una ruta de diagnóstico de la copia, apunta a https://fail-branch-check.invalid/ y usa un timeout corto. .invalid está reservado para nombres que deben ser inválidos. Configura Fail Branch en ese nodo y conecta la rama naranja con el Answer de respaldo. Esto valida el cableado Fail Branch/Answer del nodo HTTP temporal; no sustituye una prueba de fallo del propio Agent cuando ese es el objetivo de aceptación.

1. Publica la copia tras el cambio temporal. Registra la versión publicada, el ID del nodo objetivo, el ID del Answer de respaldo y una frase corta no sensible que solo aparezca en el fallback.
2. Usa el Base y la Key de la copia, los mismos inputs anonimizados y esa versión. Define los tres valores y ejecuta el mismo 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. El resumen debe mostrar a la vez target_node_failed: true, fallback_node_succeeded: true, expected_fallback_seen: true y failure_validation_passed: true. El workflow puede terminar como succeeded o partial-succeeded, pero no failed, stopped, paused ni incomplete.
4. Elimina el nodo temporal o restaura la condición original, las conexiones y la configuración normal; vuelve a publicar. Limpia las tres variables y repite la solicitud normal. Esa ejecución restaurada debe mostrar complete_for_timing_acceptance: true antes de volver a comparar el primer y último message.

Acepta la solución solo si pasan streaming y fallback

No aceptes una primera parte más rápida como único criterio. Conserva dos resúmenes: una solicitud normal completa con message distribuidos durante la generación y un fallo controlado que pruebe el fallo del nodo objetivo, la ejecución de su Fail Branch/Answer de respaldo y la entrega de un fallback anonimizado. Un error de herramienta o log de Agent por sí solo, la falta del fallo del nodo, la ausencia de fallback visible o no repetir la ejecución normal tras restaurar el cambio implican rechazo.