Dify Chatflow показывает ответ только в конце: проверяем Agent, Fail Branch и Answer

Если Dify Chatflow показывает весь ответ одной пачкой в самом конце, это ещё не доказывает, что модель не умеет стриминг. Смена провайдера тоже не является автоматическим решением. Зафиксируйте один короткий промпт, одну модель и одну конфигурацию Agent, а затем сравните четыре времени: первый токен на стороне модели, SSE-события, agent_log и пользовательские чанки message.

Задача диагностики — понять, на каком уровне текст ждёт или буферизуется. Убирать рабочую обработку ошибок заранее не нужно: отключение Fail Branch используется только как контроль в копии Chatflow.

Определите слой по последовательности событий

Что видно	Что проверять первым	Следующий шаг
Даже контроль LLM → Answer не даёт раннего текста	Модель, provider, шлюз или параметры модели	Проверьте короткий текстовый запрос и поддержку стриминга моделью
workflow_started, node_started и другие события тоже приходят поздно	Прокси, API-шлюз или SSE-транспорт	Повторите тот же запрос напрямую к Dify и исключите буферизацию ответа
agent_log идёт во время выполнения, а message приходит пачкой в конце	Путь Agent → Answer; Fail Branch — главный контрольный параметр	Сохраните fallback в продакшене и сравните тайминг в тестовой копии
message начинается только после завершения другой переменной	Более ранняя незавершённая переменная в Answer	Поменяйте порядок переменных или разбейте выдачу на несколько Answer

В документации Answer указано, что узел отдаёт контент по мере готовности переменных: он доходит до первой ещё не разрешённой переменной, ждёт её и лишь затем продолжает. Поэтому поведение определяет порядок переменных в Answer, а не порядок завершения вышестоящих узлов.

Проведите четыре запуска, меняя только один параметр

Скопируйте существующий Chatflow для теста. Запишите его опубликованную версию, Model ID, temperature, стратегию Agent, список инструментов, максимум итераций, настройки памяти и режим обработки ошибок. Во всех запусках используйте один и тот же короткий промпт, например:

Напиши 12 пронумерованных проверок для диагностики потокового ответа, по одному предложению в каждом пункте.

Промпт должен создавать достаточно длинный ответ, чтобы увидеть временной интервал, но не должен содержать реальные пользовательские данные.

Перед каждым контролем A, B, C и D повторяйте один и тот же порядок: внесите только изменение этого запуска, опубликуйте текущее состояние тестовой копии, запишите метку запуска и опубликованную версию workflow, затем вызывайте Service API Base со страницы API Access именно этой копии (полный путь обычно заканчивается на /v1) с её App API Key. Не используйте ключ исходного приложения. Если workflow_id не передан, Dify запускает последнюю опубликованную версию, поэтому вызов до публикации может несколько раз проверить одну и ту же старую сборку.

Контроль A: только LLM → Answer

В тестовой копии направьте выход той же модели из LLM сразу в Answer, без Agent и инструментов. Dify указывает, что LLM-узлы по умолчанию поддерживают потоковый вывод. Если message и здесь появляется только в конце, пока не обвиняйте Agent: сначала проверьте provider, OpenAI-compatible шлюз, обратный прокси и порядок переменных Answer.

Контроль B: текущий Agent с сохранённым Fail Branch

Верните реальный путь Agent и действующую аварийную ветку. Это базовый запуск. Не удаляйте fallback ради более быстрой анимации: Fail Branch предназначен для отдельного сценария после ошибки — логирования, уведомления, альтернативного действия или резервного сервиса.

Контроль C: отключите Fail Branch только в тестовой копии

Проводите этот запуск только если текущий Agent показывает тот же параметр обработки ошибок. В копии верните режим по умолчанию, не меняя модель, промпт, инструменты и остальные настройки. Такой контроль показывает лишь корреляцию аварийной ветки с задержкой в вашей среде; он не доказывает, что продакшен безопасно оставлять без fallback.

В пользовательском отчёте для Dify 1.17.1, langgenius/agent 0.0.33 и graphon 0.7.0 при включённом Fail Branch 805 событий message пришли между 29,819 и 30,035 секунды — интервал составил 0,216 секунды. В контроле со стандартной обработкой 866 сообщений распределились от 3,397 до 40,991 секунды. При этом agent_log приходил во время выполнения. Это данные одной конфигурации, а не общий вывод для всех версий, и подтверждённой версии с исправлением в отчёте нет.

Контроль D: меняйте только порядок переменных Answer

Сначала оставьте в Answer только text узла Agent, затем возвращайте остальные переменные по одной. Если поздняя переменная стоит перед текстом Agent, Answer может ждать её, хотя Agent уже генерирует текст. Поставьте потоковую текстовую переменную раньше или используйте несколько Answer-узлов, которые официально поддерживаются для выдачи на разных этапах.

Запишите реальное время прибытия через API

Вызывайте опубликованный API Chatflow, а не оценивайте только анимацию в браузере. Согласно руководству Dify по streaming, response_mode: "streaming" возвращает SSE. В Chatflow пользовательский ответ получается последовательным объединением событий message. agent_log — информационный журнал шага Agent: его получение подтверждает работу канала событий, но не доставку текста пользователю.

Для каждого запуска берите Service API Base и App API Key со страницы API Access тестовой копии. Указывайте полный Base, обычно с /v1; скрипт добавит /chat-messages. Если доступна фиксация версии, поместите ID опубликованной версии workflow этого запуска в DIFY_WORKFLOW_ID. Оставляйте поле пустым только намеренно, чтобы использовать последнюю опубликованную версию. В любом случае сначала публикуйте изменения и включайте версию в DIFY_RUN_LABEL.

Значение {} для DIFY_TEST_INPUTS_JSON допустимо только тогда, когда у копии нет обязательных пользовательских входов. Если Start/User Input задаёт обязательные поля, передавайте во всех четырёх запусках одинаковые обезличенные JSON-значения с точными именами и типами параметров приложения.

В Bash считайте App API Key скрытым вводом и задайте параметры запуска:

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

Сохраните следующий файл как 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)

Скрипт не записывает текст ответа на диск. Он сохраняет время, ID/название узла, data.status, длину текста, а для события error — верхнеуровневые status и code, заменяя message маркером с количеством символов. Отдельно различаются succeeded, failed, partial-succeeded, stopped, paused, верхнеуровневый error и поток без конечного события.

Имена файлов формируются из DIFY_RUN_LABEL. При обрыве соединения или ошибке разбора JSON уже собранные времена сохраняются, но run_outcome становится incomplete, а complete_for_timing_acceptance — false. Такие интервалы пригодны только для диагностики. Нормальный контроль проходит проверку времени лишь после message_end и workflow_finished.data.status=succeeded. При усечении HTTP/1.1 chunked SSE исключение http.client.IncompleteRead также перехватывается: времена уже полученных событий остаются в summary, acceptance_checks_passed=false, а фрагмент тела ответа из исключения не выводится.

Сравнивайте четыре времени

1. Первый токен upstream. Если provider или собственный шлюз публикует такую метрику, сохраните её. Иначе используйте контроль LLM → Answer как практическое разделение уровней.
2. Первое SSE-событие с данными. Ранний workflow_started означает, что запуск принят, но ещё не доказывает потоковую доставку ответа.
3. Первый agent_log. Ранние журналы показывают, что шаги Agent проходят по SSE. Если журнал идёт, а текст ждёт, весь канал, вероятно, не буферизуется одним объектом.
4. Первый и последний message. Это пользовательские времена. Сравните first_s, last_s и span_s. Много сообщений в очень коротком интервале — измеримое подтверждение финального «залпа».

Успешный Chatflow должен выдать message_end, затем workflow_finished с data.status=succeeded. Официальный контракт SSE перечисляет для workflow_finished.data.status значения succeeded, failed, partial-succeeded и stopped; фиксируйте их отдельно. При ожидании ввода пользователя вместо этого приходят human_input_required, затем отдельное событие workflow_paused: именно по нему скрипт записывает run_outcome=paused. Перечень статусов записи о запуске нельзя переносить на контракт события workflow_finished. У события error статус, тип ошибки и сообщение находятся на верхнем уровне; скрипт показывает status, code и обезличенный message. Если соединение заканчивается без конечного события либо прерывается транспорт/разбор JSON, summary получает incomplete. Даже при наличии нескольких message их интервал нельзя считать доказательством исправления, пока complete_for_timing_acceptance не равен true.

Выберите действие по результату

Контроль LLM тоже ждёт: сначала исправьте upstream

Проверьте Model ID, endpoint, потоковое поведение provider и промежуточный шлюз. Если BetterToken используется как дополнительный OpenAI-compatible базовый provider, его инструкция для Dify рекомендует сначала запустить короткий обычный текст без инструментов, а Agent/tool calls включать только после подтверждения поддержки выбранной моделью и endpoint. Такой тест изолирует модельный слой, но не обещает отсутствие буферизации внутри Dify.

Все события приходят поздно: проверяйте транспорт

Повторите тот же запрос с минимальным числом CDN, reverse proxy и API-шлюзов. Если прямое соединение с Dify даёт постепенные события, а после возвращения посредника они группируются, проверяйте потоковую передачу и буферизацию на этом посреднике. Модель при этом не меняйте.

Только Fail Branch группирует message

Формулируйте вывод узко: задержка связана с текущей версией Dify, плагином Agent и конфигурацией данного workflow. По умолчанию сохраняйте продакшен-fallback. Если живой вывод важнее, сначала спроектируйте и проверьте другую схему восстановления; один контроль не оправдывает удаление уведомлений, резервных вызовов и журналов ошибок.

Перед обновлением Dify или плагина Agent сохраните один полный нормальный запуск и одну проверку контролируемого сбоя. После обновления опубликуйте тестовую копию и повторите проверку с тем же запросом, моделью, обезличенными inputs и способом фиксации версии. Одной сохранённой временной шкалы сбоя недостаточно: целевой узел должен действительно завершиться ошибкой, соответствующий Fail Branch и резервный Answer должны выполниться, а пользователь — получить обезличенное сообщение.

Перестановка Answer вернула стриминг

Возможно, upstream-модель всё время отдавала поток, а Answer ждал более раннюю переменную. Сохраните новый порядок или несколько Answer, затем выполните проверку сбоя ниже. error_type и error_message можно использовать для маршрутизации и внутренних журналов, но пользовательский Answer должен содержать безопасный текст, а не исходные детали ошибки.

Проверьте один контролируемый сбой только в тестовой копии

Предпочтителен уже существующий повторяемый сценарий без реальных пользовательских данных, при котором проверяемый целевой узел выдаёт node_finished с признаком ошибки; после обработки через Fail Branch допустим data.status=exception, поэтому требовать именно failed нельзя. Внутренняя ошибка инструмента, запись agent_log или журнал модели сами по себе не доказывают, что узел Agent завершился ошибкой.

В исходном коде соответствующих версий graphon 0.7.0 преобразует сбой, обработанный через FAIL_BRANCH, в NodeRunExceptionEvent со статусом exception. Dify 1.17.1 передаёт его как node_finished.data.status=exception, тогда как необработанный сбой узла имеет статус failed. Поле скрипта target_node_failed принимает exception или failed как подтверждение ошибки самого целевого узла, а не всего workflow и не успешности резервной ветки. Для приёмки по-прежнему нужны успешный соответствующий fallback-узел, видимый пользователю безопасный ответ без чувствительных данных и один из допустимых конечных исходов, перечисленных ниже.

Если подходящего сценария нет, временно добавьте в диагностическую ветку тестовой копии узел HTTP Request, укажите https://fail-branch-check.invalid/ и короткий timeout. Домен .invalid зарезервирован для заведомо недействительных имён. Установите для этого узла обработку Fail Branch и соедините оранжевую ветку с проверяемым резервным Answer. Так проверяется проводка Fail Branch/Answer временного HTTP-узла; это не заменяет сбой самого Agent, если принимается именно его ветка.

1. После временного изменения опубликуйте тестовую копию. Запишите опубликованную версию, ID целевого узла, ID резервного Answer и короткую безопасную фразу, встречающуюся только в fallback.
2. Используйте Base и Key копии, те же обезличенные inputs и эту опубликованную версию. Задайте все три значения и запустите тот же скрипт:

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. В summary должны одновременно быть target_node_failed: true, fallback_node_succeeded: true, expected_fallback_seen: true и failure_validation_passed: true. Конечный статус может быть succeeded или partial-succeeded, но не failed, stopped, paused или incomplete.
4. Удалите временный узел либо восстановите исходное контролируемое условие, нормальные связи и настройки, затем снова опубликуйте копию. Очистите три проверочные переменные и повторите нормальный запрос. Он должен дать complete_for_timing_acceptance: true; только после этого снова сравнивайте первое и последнее время message.

Принимайте исправление только после двух проверок

Не принимайте исправление только из-за более раннего первого фрагмента. Сохраните два summary: полный нормальный запрос, где message распределены по времени генерации, и контролируемый сбой, доказывающий ошибку целевого узла, выполнение соответствующего Fail Branch/резервного Answer и получение пользователем обезличенного fallback. Ошибка инструмента или Agent-журнал без ошибки целевого узла, отсутствие видимого fallback либо отсутствие нормального повторного запуска после восстановления конфигурации означают, что приёмка не пройдена.