Dify Chatflow का उत्तर अंत में एक साथ आए तो Agent, Fail Branch और Answer की जाँच करें

Dify Chatflow का पूरा उत्तर अंत में एक साथ दिखना यह साबित नहीं करता कि upstream model streaming नहीं कर रहा। यह भी साबित नहीं करता कि provider बदलने से समस्या ठीक हो जाएगी। एक छोटा prompt, एक model और वही Agent configuration स्थिर रखें, फिर चार समयों की तुलना करें: upstream first token, SSE events, agent_log और user को मिलने वाले message chunks।

उद्देश्य यह पता लगाना है कि text किस परत पर रुक रहा है या buffer हो रहा है। Production error handling पहले हटाने की जरूरत नहीं है। Fail Branch को बंद करना केवल test copy में diagnostic control है, सामान्य production fix नहीं।

Event pattern से समस्या की परत पहचानें

जो दिखाई दे रहा है	पहले किस पर संदेह करें	अगली जाँच
LLM → Answer control में भी text जल्दी नहीं आता	Model, provider, gateway या model settings	छोटा plain-text request और model streaming support जाँचें
workflow_started, node_started और बाकी events भी देर से आते हैं	Proxy, gateway या SSE transport	वही request सीधे Dify पर चलाएँ और response buffering अलग करें
agent_log पूरे run में आता है, पर message अंत में एक साथ आता है	Agent → Answer text path; Fail Branch को control बनाएँ	Production fallback रखें और cloned flow में timing तुलना करें
किसी दूसरी variable के पूरा होने के बाद ही message शुरू होता है	Answer में पहले रखी unresolved variable	Variables का क्रम बदलें या कई Answer nodes से चरणों में output दें

Dify Answer documentation के अनुसार Answer node variables उपलब्ध होने के आधार पर content stream करता है। वह पहली unresolved variable तक content भेजता है, फिर उसके resolve होने की प्रतीक्षा करता है। इसलिए user को पहले क्या दिखेगा, यह upstream nodes के execution order से नहीं बल्कि Answer में variables के क्रम से तय होता है।

चार controls बनाएँ, हर बार सिर्फ एक चीज बदलें

मौजूदा Chatflow की test copy बनाएँ। Published version, Model ID, temperature, Agent strategy, tools, maximum iterations, memory settings और error-handling mode लिख लें। हर run में बिल्कुल वही छोटा prompt रखें, जैसे:

Streaming response को debug करने के लिए 12 numbered checks लिखें, हर check केवल एक वाक्य हो।

उत्तर इतना लंबा होना चाहिए कि timing span दिखाई दे, लेकिन उसमें वास्तविक user data नहीं होना चाहिए।

Control A, B, C और D में हर बार यही क्रम दोहराएँ: केवल उस round का एक बदलाव करें, test copy की मौजूदा स्थिति publish करें, run label और published workflow version लिखें, फिर उसी copy के API Access page पर दिखे Service API Base (पूरा path आम तौर पर /v1 पर समाप्त होता है) और उसी का App API Key इस्तेमाल करें। मूल app का key दोबारा न लें। workflow_id न देने पर Dify latest published version चलाता है; इसलिए publish किए बिना call करने पर कई rounds में वही पुराना build चल सकता है।

Control A: केवल LLM → Answer

Test copy में उसी model का LLM output सीधे Answer में भेजें, Agent और tools के बिना। Dify बताता है कि LLM nodes default रूप से streaming output देते हैं। यदि यहाँ भी message केवल अंत में आए, तो अभी Agent को कारण न मानें। पहले provider, OpenAI-compatible gateway, reverse proxy और Answer variable order जाँचें।

Control B: मौजूदा Agent और Fail Branch बनाए रखें

वास्तविक Agent path और production fallback वापस रखें। यही baseline है। केवल display तेज करने के लिए recovery path न हटाएँ: Fail Branch node failure के बाद अलग flow चला सकता है, जैसे logging, notification, दूसरा तरीका या backup service।

Control C: Fail Branch केवल test copy में बंद करें

यह test तभी करें जब वर्तमान Agent node वही error-handling option दिखाता हो। Copy में error handling को default करें, लेकिन model, prompt, tools और बाकी settings न बदलें। इससे केवल यह पता चलता है कि आपकी मौजूदा environment में branch और delay साथ दिखाई देते हैं या नहीं। इससे production को बिना fallback चलाना सुरक्षित साबित नहीं होता।

Dify 1.17.1, langgenius/agent 0.0.33 और graphon 0.7.0 के एक user report में Fail Branch चालू होने पर 805 message events 29.819 से 30.035 seconds के बीच आए, यानी span केवल 0.216 second था। Default error-handling control में 866 messages 3.397 से 40.991 seconds तक फैले थे। उसी run में agent_log events लगातार आते रहे। यह एक विशेष reported environment का evidence है, सभी Dify versions का सामान्य निष्कर्ष या confirmed fixed version नहीं।

Control D: केवल Answer variable order बदलें

पहले Answer में केवल Agent का text रखें, फिर बाकी variables एक-एक करके जोड़ें। यदि देर से पूरी होने वाली variable Agent text से पहले है, तो Agent generate करते हुए भी Answer प्रतीक्षा कर सकता है। Streaming text variable को पहले रखें या अलग stages पर content देने के लिए officially supported multiple Answer nodes का उपयोग करें।

न्यूनतम API call से वास्तविक arrival time दर्ज करें

Browser animation पर निर्भर रहने के बजाय published Chatflow API को call करें। Dify streaming guide के अनुसार response_mode: "streaming" SSE लौटाता है। Chatflow में user-facing answer क्रम से मिले message events को जोड़कर बनता है। agent_log केवल Agent step का informational log है; उसका आना event channel के सक्रिय होने का प्रमाण है, answer text के user तक पहुँचने का नहीं।

हर run के लिए test copy के API Access page से Service API Base और App API Key लें। पूरा Base दें, सामान्यतः /v1 सहित; script /chat-messages जोड़ता है। Version pinning उपलब्ध हो तो उस round की published workflow version ID DIFY_WORKFLOW_ID में दें। इसे खाली तभी छोड़ें जब जानबूझकर latest published version चलाना हो। दोनों स्थितियों में पहले publish करें और version को DIFY_RUN_LABEL में दर्ज करें।

DIFY_TEST_INPUTS_JSON में {} केवल तब दें जब copy में कोई required custom input न हो। यदि Start/User Input में required fields हैं, तो app के exact parameter names और types के अनुसार वही sanitized JSON values चारों runs में भेजें।

Bash में App API Key को बिना दिखाए पढ़ें और run parameters भरें:

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 के रूप में save करें:

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)

Script answer text को disk पर नहीं लिखता। यह timing, node ID/title, data.status, answer length और error event के top-level status तथा code दर्ज करता है; message को character-count marker से redact करता है। यह succeeded, failed, partial-succeeded, stopped, paused, top-level error और terminal event के बिना समाप्त stream को अलग रखता है।

Output filenames DIFY_RUN_LABEL से बनते हैं। Connection टूटने या SSE JSON parse न होने पर पहले से मिली timings save रहती हैं, लेकिन run_outcome incomplete और complete_for_timing_acceptance false होता है। ऐसी spans केवल diagnosis के लिए हैं। Normal control तभी timing acceptance pass करता है जब message_end और workflow_finished.data.status=succeeded दोनों मिले हों। HTTP/1.1 chunked SSE कटने पर उठने वाला http.client.IncompleteRead भी पकड़ा जाता है: मिले हुए events की timings summary में रहती हैं, acceptance_checks_passed=false होता है, और exception में मौजूद अधूरा response body नहीं दिखाया जाता।

केवल चार समयों की तुलना करें

1. Upstream first token। Provider या self-hosted gateway में metric उपलब्ध हो तो उसे दर्ज करें। Metric न हो तो LLM → Answer control upstream path को व्यावहारिक रूप से अलग करता है।
2. पहला SSE data event। जल्दी आया workflow_started बताता है कि run स्वीकार हुआ, लेकिन progressive answer delivery साबित नहीं करता।
3. पहला agent_log। Early logs बताते हैं कि Agent step events connection से गुजर रहे हैं। यदि logs आते रहें पर text रुका रहे, तो पूरा connection एक ही object की तरह buffer होने की संभावना कम होती है।
4. पहला और आखिरी message। User experience के लिए यही महत्वपूर्ण हैं। first_s, last_s और span_s की तुलना करें। बहुत से events का बहुत छोटे interval में आना final burst का measurable evidence है।

Successful Chatflow में पहले message_end, फिर workflow_finished और उसका data.status=succeeded होना चाहिए। आधिकारिक SSE contract में workflow_finished.data.status के लिए succeeded, failed, partial-succeeded और stopped दिए गए हैं; इन्हें अलग रिकॉर्ड करें। Human-input pause में इसके बजाय human_input_required के बाद अलग workflow_paused event आता है, जिससे script run_outcome=paused रिकॉर्ड करता है। Run record की status enum को workflow_finished event का contract न मानें। error event में status, error type और message top level पर होते हैं; script status, code और redacted message दिखाता है। Terminal event के बिना connection बंद हो या transport/JSON parsing टूटे, तो summary incomplete होगा। कुछ message events मिलने पर भी उनकी span से सुधार सिद्ध नहीं होता जब तक complete_for_timing_acceptance true न हो।

Result के अनुसार ही configuration बदलें

LLM control भी रुकता है: पहले upstream layer ठीक करें

Agent पर लौटने से पहले Model ID, endpoint, provider streaming और intermediate gateway जाँचें। यदि BetterToken को optional OpenAI-compatible model API baseline की तरह उपयोग करते हैं, तो उसकी Dify setup guide पहले tools के बिना छोटा plain-text request चलाने और चुने हुए model तथा endpoint का support साबित होने के बाद ही tool calls चालू करने की सलाह देती है। इससे upstream model layer अलग हो सकती है, लेकिन Dify nodes buffering नहीं करेंगे—यह साबित नहीं होता।

सभी events देर से आते हैं: transport path जाँचें

CDN, reverse proxy या अतिरिक्त API gateway layers कम करके वही request दोहराएँ। यदि Dify से direct connection पर events क्रम से आते हैं और किसी intermediary के वापस आने पर फिर grouped हो जाते हैं, तो उसी layer पर SSE forwarding और response buffering जाँचें। साथ में model न बदलें।

केवल Fail Branch पर message grouped होता है

निष्कर्ष को अपनी Dify version, Agent plugin और workflow configuration तक सीमित रखें। Production fallback default रूप से बनाए रखें। यदि live delivery अधिक महत्वपूर्ण है, तो migration से पहले दूसरी recovery path design और verify करें। एक control run alerts, backup calls या error logs हटाने का पर्याप्त कारण नहीं है।

Dify या Agent plugin upgrade से पहले एक complete normal run और एक controlled-failure verification save करें। Upgrade के बाद test copy publish करके वही prompt, model, sanitized inputs और version-recording प्रक्रिया दोहराएँ। केवल failure timeline save होना completion नहीं है: target node वास्तव में fail होना चाहिए, संबंधित Fail Branch और fallback Answer चलने चाहिए, और user को sanitized fallback दिखना चाहिए।

Answer reorder करने पर streaming लौटती है

संभव है upstream model हमेशा stream कर रहा था और Answer किसी पहले variable की प्रतीक्षा कर रहा था। नया order या staged Answer nodes रखें, फिर नीचे दिया failure verification चलाएँ। error_type और error_message routing तथा internal logs के लिए उपयोग हो सकते हैं, लेकिन user-facing Answer में raw error details की जगह sanitized wording रखें।

Controlled failure केवल test copy में verify करें

पहले ऐसा existing, repeatable condition चुनें जिसमें real user data न हो और review किया जा रहा exact target node node_finished में अपने execution की त्रुटि दिखाए; Fail Branch से handled error में data.status=exception मान्य है, इसलिए failed अनिवार्य नहीं होना चाहिए। Tool का internal error, कोई agent_log, या model log अपने-आप Agent node failure का प्रमाण नहीं है।

संबंधित version के source में graphon 0.7.0, FAIL_BRANCH से handled failure को NodeRunExceptionEvent और exception status में बदलता है। Dify 1.17.1 इसे node_finished.data.status=exception के रूप में भेजता है, जबकि unhandled node failure का status failed होता है। Script का target_node_failed, exception या failed को target node की अपनी त्रुटि का प्रमाण मानता है—पूरे workflow की विफलता या सफल fallback का नहीं। Validation के लिए संबंधित fallback node का सफल होना, user को दिखने वाला sanitized fallback और नीचे दिए गए अनुमत final outcomes में से एक अब भी ज़रूरी हैं।

यदि ऐसा condition न हो, test copy की diagnostic path में अस्थायी HTTP Request node जोड़ें, URL https://fail-branch-check.invalid/ रखें और छोटा timeout दें। .invalid जानबूझकर invalid names के लिए reserved है। इस node की error handling को Fail Branch करें और orange branch को review वाले fallback Answer से जोड़ें। यह temporary HTTP node की Fail Branch/Answer wiring verify करता है; यदि acceptance target Agent node है, तो Agent-level controlled failure की जगह इसे न मानें।

1. Temporary बदलाव के बाद test copy publish करें। Published version, target node ID, fallback Answer node ID और केवल fallback में आने वाली एक छोटी non-sensitive phrase लिखें।
2. Copy का Base और Key, वही sanitized inputs और वही published version लें। तीनों validation values set करके वही 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. 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. Temporary failure node हटाएँ या original controlled condition restore करें, normal connections/configuration वापस रखें और copy फिर publish करें। तीनों validation variables साफ करके normal request दोबारा चलाएँ। Restored run में complete_for_timing_acceptance: true होना चाहिए; उसके बाद ही message के first/last time की तुलना करें।

Streaming और fallback दोनों pass होने पर ही fix स्वीकार करें

सिर्फ पहला chunk जल्दी आने को success न मानें। दो summaries रखें: एक complete normal request जिसमें generation के दौरान message chunks फैले हों, और एक controlled failure जिसमें target node का failure, संबंधित Fail Branch/fallback Answer का execution और user को sanitized fallback मिलना सिद्ध हो। केवल tool error या Agent log, target-node failure का अभाव, user-visible fallback का अभाव, या test बदलाव restore करने के बाद normal rerun न होना—इनमें से कोई भी acceptance fail करता है।