Dify Chatflowの回答が最後に一括表示される原因を切り分ける

Dify Chatflowの回答が最後にまとめて表示されても、上流モデルがストリーミングしていないとは限りません。providerを変更すれば直るとも限りません。同じ短いプロンプト、同じモデル、同じAgent設定を固定し、上流の最初のtoken、SSEイベント、agent_log、ユーザーに届くmessageの4つの時刻を比較してください。

調べるべきなのは「Difyがstreaming対応か」ではなく、「どの層でテキストが待たされているか」です。本番のエラー処理を先に外す必要はありません。Fail Branchを無効にするのは、テスト用コピーで行う対照実験だけです。

イベントの出方から疑う層を決める

観測した状態	最初に疑う層	次に行う確認
LLM → Answerだけの対照でも早いテキストが来ない	モデル、provider、gateway、モデル設定	短いプレーンテキスト要求とモデル側streamingを確認する
workflow_startedやnode_startedもまとめて遅い	Dify手前のproxy、gateway、SSE経路	同じ要求をDifyへ直接送り、応答bufferingを切り分ける
実行中にagent_logは届くが、messageは最後に集中する	AgentからAnswerまでのテキスト経路。Fail Branchを対照にする	本番fallbackを残し、コピーでmessage時刻を比較する
別の変数が完了するまでmessageが始まらない	Answer内で先に置かれた未解決変数	変数順を変えるか、複数のAnswerで段階的に返す

DifyのAnswerドキュメントでは、変数が利用可能になる順に内容をstreamし、最初の未解決変数で待機すると説明されています。上流ノードの実行順ではなく、Answer本文に書いた変数順が、ユーザーに何が先に見えるかを決めます。

1回につき1項目だけ変える4つの対照を作る

既存Chatflowを複製してテスト用にします。公開版、Model ID、temperature、Agent strategy、tools、最大反復回数、memory設定、error handlingを控えてください。全実行で同じ短いプロンプトを使います。例：

ストリーミング応答を調査するための確認項目を、1項目1文で12個、番号付きで書いてください。

時間差を観測できる長さは必要ですが、実ユーザーデータは含めません。

対照A、B、C、Dの各回で同じ手順を繰り返します。その回の変更だけを反映し、テスト用コピーの現在状態を公開し、実行ラベルと公開済みworkflowバージョンを記録してから、そのコピーのAPI Accessに表示されるService API Base（完全なパスは通常/v1で終わる）とApp API Keyで呼び出します。元アプリのキーは流用しません。workflow_idを省略すると最新の公開済みバージョンが実行されるため、公開前に呼ぶと複数回とも古い同一版を測る可能性があります。

対照A：LLM → Answerだけにする

テスト用コピーで、同じモデルのLLM出力をAgentやtoolsを通さずAnswerへ直接渡します。DifyはLLMノードが標準でstreaming出力に対応すると説明しています。この対照でもmessageが最後だけなら、まだAgentを原因と決めないでください。provider、OpenAI-compatible gateway、reverse proxy、Answerの変数順を先に確認します。

対照B：現在のAgentとFail Branchを維持する

実際のAgent経路と本番fallbackを戻し、これを基準にします。表示を速くするためだけに復旧経路を削除しないでください。Fail Branchは、失敗時にログ記録、通知、別の処理、backup serviceなどの独立した経路を実行するための機能です。

対照C：テスト用コピーだけでFail Branchを無効にする

現在のAgentノードに同じerror handling設定がある場合だけ実施します。コピー側を既定設定へ戻し、モデル、プロンプト、tools、その他の値は変えません。この結果で分かるのは、現在の環境でFail Branchと遅延が関連するかどうかです。本番からfallbackを外してよいという結論にはなりません。

Dify 1.17.1、langgenius/agent 0.0.33、graphon 0.7.0を対象としたユーザー報告では、Fail Branch有効時に805件のmessageが29.819秒から30.035秒までの0.216秒に集中しました。既定のerror handlingでは866件が3.397秒から40.991秒まで分散し、同じ実行中にagent_logは継続して到着したとされています。これは特定環境の報告であり、全Difyバージョンの一般則でも、修正版の確定情報でもありません。

対照D：Answerの変数順だけを変える

最初はAnswerでAgentのtextだけを参照し、他の変数を1つずつ戻します。完了の遅い変数がAgentテキストより前にあると、Agentが生成中でもAnswerは待つ可能性があります。streamingするテキスト変数を先に置くか、公式にサポートされている複数のAnswerノードで段階的に返します。

最小API呼び出しで実際の到着時刻を保存する

ブラウザーの表示だけで判断せず、公開済みChatflow APIを呼び出します。Difyのstreamingガイドでは、response_mode: "streaming"はSSEを返します。Chatflowのユーザー向け回答はmessageイベントを順番に連結したものです。agent_logはAgentのステップログであり、受信できても回答本文がユーザーへ逐次届いた証明にはなりません。

各実行では、テスト用コピーのAPI AccessからService API BaseとApp API Keyを取得します。Baseは通常/v1を含む完全な値を入力し、スクリプトが/chat-messagesを追加します。バージョン固定が使える場合は、その回の公開済みworkflow version IDをDIFY_WORKFLOW_IDに入れます。空欄は意図的に最新公開版を使う場合だけです。どちらの場合も先に公開し、バージョンをDIFY_RUN_LABELへ記録します。

DIFY_TEST_INPUTS_JSONに{}を使えるのは、必須のカスタム入力がない場合だけです。Start/User Inputに必須項目があるなら、アプリの正確なパラメータ名と型に合わせた同じ匿名化JSON値を4回すべてで送ります。

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)

スクリプトは回答本文をディスクへ保存しません。時刻、node ID/title、data.status、文字数に加え、errorイベントではトップレベルのstatusとcodeを記録し、messageは文字数だけのマーカーへ置き換えます。succeeded、failed、partial-succeeded、stopped、paused、トップレベルerror、終端イベントなしを区別します。

ファイル名にはDIFY_RUN_LABELを使います。接続中断やSSE 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となり、例外に含まれる応答本文の断片は表示しません。

4つの時刻だけを比較する

1. 上流の最初のtoken。 providerや自前gatewayにfirst-token指標があれば保存します。なければLLM → Answer対照で上流経路を実用的に切り分けます。
2. 最初のSSE dataイベント。 早いworkflow_startedは実行受付を示しますが、回答が逐次届いた証拠ではありません。
3. 最初のagent_log。 早く届けばAgentのステップイベントは接続を通過しています。ログが来てもテキストが待つなら、接続全体が1つの応答として保持されている可能性は下がります。
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、error type、messageはトップレベルにあり、スクリプトはstatus、code、匿名化したmessageを表示します。終端イベントなしで接続が閉じた場合や、通信/JSON解析が中断した場合はsummaryがincompleteになります。messageを一部取得していても、complete_for_timing_acceptanceがtrueでなければ、そのspanで改善を主張できません。

結果に合う変更だけを選ぶ

LLM対照も待つ：先に上流層を直す

Agentへ戻る前に、Model ID、endpoint、providerのstreaming、途中のgatewayを確認します。BetterTokenを任意のOpenAI-compatibleモデルAPI基準として使う場合、Dify設定ガイドは、まずtoolsなしの短いプレーンテキスト要求を通し、選択したモデルとendpointが対応すると確認してからtool callsを有効にする手順を示しています。上流モデル層の切り分けには使えますが、Difyノードがbufferingしない保証ではありません。

すべてのイベントが遅い：通信経路を確認する

CDN、reverse proxy、追加API gatewayを減らし、同じ要求を再実行します。Difyへ直接つなぐと逐次届き、ある中間層を戻すとまとまる場合は、その層のSSE転送とresponse bufferingを確認します。同時にモデルを変えると原因を特定できません。

Fail Branch有効時だけmessageが集中する

結論は「現在のDify版、Agent plugin、workflow設定で関連が見られた」と限定します。本番fallbackは原則として維持してください。リアルタイム表示を優先する場合も、別の復旧経路を設計して動作確認してから移行します。1回の対照だけで通知、backup呼び出し、error logを削除しないでください。

DifyまたはAgentプラグインを更新する前に、完全な正常実行と制御した失敗検証を1件ずつ保存します。更新後はテスト用コピーを公開し、同じprompt、model、匿名化inputs、バージョン記録方法で再実行します。失敗タイムラインを保存しただけでは完了ではありません。対象node自体が実際に失敗し、対応するFail Branchとfallback Answerが実行され、ユーザーに匿名化fallbackが届く必要があります。

Answerの並べ替えでstreamingが戻る

上流modelは最初からstreamingしており、Answerが前の変数を待っていただけかもしれません。新しい順序または段階的なAnswerを維持し、次の失敗検証を行います。error_typeとerror_messageは分岐や内部ログに使えますが、ユーザー向けAnswerには生の詳細ではなく匿名化した文面を使います。

制御した失敗はテスト用コピーだけで検証する

実ユーザーデータを使わず再現可能で、検証対象のnodeそのものがnode_finishedで自身の異常を示す既存条件を優先します。Fail Branchで処理済みのエラーはdata.status=exceptionになり得るため、failedを必須にしてはいけません。tool内部のエラー、agent_log、modelログだけではAgent nodeの失敗を証明できません。

該当バージョンのソースでは、graphon 0.7.0がFAIL_BRANCHで処理した失敗を、状態exceptionのNodeRunExceptionEventに変換します。Dify 1.17.1はこれをnode_finished.data.status=exceptionとして出力し、未処理のノード失敗はfailedになります。スクリプトのtarget_node_failedはexceptionまたはfailedを対象ノード自身の異常として扱いますが、workflow全体の失敗やfallback成功を意味しません。検証には引き続き、対応するfallbackノードの成功、ユーザーに表示される機密情報を含まないfallback回答、下記の許可された終了結果が必要です。

適切な条件がない場合は、テスト用コピーの診断経路にHTTP Request nodeを一時追加し、URLをhttps://fail-branch-check.invalid/、timeoutを短めに設定します。.invalidは必ず無効になる名前用に予約されています。このnodeのerror handlingをFail Branchにし、オレンジのbranchを検証対象のfallback Answerへ接続します。これは一時HTTP nodeのFail Branch/Answer配線を検証する方法であり、Agent自体を対象にする場合のAgent-level failureの代替にはなりません。

1. 一時変更後にテスト用コピーを公開し、公開版、対象node ID、fallback Answer node ID、fallbackだけに現れる機密性のない短い文を記録します。
2. コピーのBase/Key、同じ匿名化inputs、その公開版を使い、3つの検証値を設定して同じスクリプトを実行します。

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を同時に確認します。workflow終端はsucceededまたはpartial-succeededでよいですが、failed、stopped、paused、incompleteは不可です。
4. 一時nodeを削除するか元の制御条件を戻し、通常の接続と設定を復元して再公開します。3つの検証変数をクリアし、通常リクエストを再実行します。復元後の実行でcomplete_for_timing_acceptance: trueを確認してから、messageのfirst/last timeを再比較します。

streamingとfallbackの両方が通ってから完了にする

最初のchunkが速くなったことだけで完了にしません。2つのsummaryを保存します。1つは生成中にmessageが分散して届く完全な正常実行、もう1つは対象nodeの失敗、対応するFail Branch/fallback Answerの実行、匿名化fallbackのユーザー表示を証明する制御失敗です。tool errorやAgent logだけ、対象node失敗なし、表示fallbackなし、または変更復元後の正常再実行なしなら受け入れ不可です。