Dify Chatflow responde só no fim: diagnostique Agent, Fail Branch e Answer

Uma resposta do Dify Chatflow que aparece inteira no fim não prova que o modelo deixou de transmitir. Também não prova que trocar o provedor resolverá o problema. Mantenha fixos um prompt curto, um modelo e a configuração do Agent, e compare quatro relógios: primeiro token upstream, eventos SSE, agent_log e blocos message entregues ao usuário.

O objetivo é descobrir em qual camada o texto espera ou fica armazenado. Você não precisa remover a proteção de produção: desativar Fail Branch é apenas um controle em uma cópia de teste.

Identifique a camada pelo padrão dos eventos

O que você observa	Suspeita principal	Próxima verificação
O controle LLM → Answer também não mostra texto cedo	Modelo, provider, gateway ou parâmetros	Valide uma solicitação curta de texto e o streaming do modelo
workflow_started, node_started e outros eventos chegam todos atrasados	Proxy, gateway ou transporte SSE	Repita a mesma solicitação diretamente no Dify e procure buffering
agent_log chega durante a execução, mas message vem em uma rajada final	Caminho Agent → Answer; Fail Branch como variável de controle	Preserve o fallback de produção e compare o tempo em um fluxo clonado
message só começa depois que outra variável termina	Variável anterior ainda não resolvida no Answer	Reordene variáveis ou use vários nós Answer por etapa

A documentação do Answer diz que o nó transmite conforme as variáveis ficam disponíveis. Ele entrega o conteúdo até a primeira variável não resolvida, espera e continua depois. Portanto, a ordem das variáveis no Answer — não a ordem de conclusão dos nós upstream — controla o que aparece primeiro.

Faça quatro controles alterando apenas um fator

Duplique o Chatflow atual para testes. Registre versão publicada, Model ID, temperature, estratégia do Agent, ferramentas, máximo de iterações, memória e tratamento de erros. Use exatamente o mesmo prompt curto em todas as execuções, por exemplo:

Escreva 12 verificações numeradas para depurar uma resposta em streaming, com uma frase por item.

O texto precisa ser longo o bastante para criar um intervalo observável, mas não deve conter dados reais de usuários.

Antes dos controles A, B, C e D, repita sempre a mesma sequência: aplique apenas a mudança daquela rodada, publique o estado atual da cópia de teste, registre um rótulo de execução e a versão publicada do workflow, e use o Service API Base exibido em API Access dessa cópia (o caminho completo normalmente termina em /v1) com o App API Key dela. Não reutilize a chave do app original. Sem workflow_id, o Dify executa a versão publicada mais recente; chamar antes de publicar pode fazer várias rodadas testarem a mesma versão antiga.

Controle A: somente LLM → Answer

Na cópia, envie a saída do mesmo modelo do LLM diretamente ao Answer, sem Agent nem ferramentas. O Dify documenta que nós LLM transmitem por padrão. Se message ainda vier só no final, não culpe o Agent ainda: verifique primeiro provider, gateway OpenAI-compatible, proxy reverso e ordem de variáveis do Answer.

Controle B: Agent atual com Fail Branch preservado

Restaure o caminho real do Agent e o fallback de produção. Essa é a linha de base. Não elimine a recuperação apenas para acelerar a interface: Fail Branch executa um fluxo separado após falha, como registrar, alertar, tentar outra abordagem ou usar um serviço de backup.

Controle C: desative Fail Branch apenas na cópia

Faça esse teste somente se o Agent atual expuser a mesma opção. Retorne o tratamento de erros ao padrão na cópia, sem mudar modelo, prompt, ferramentas ou outros parâmetros. O resultado apenas mostra se a ramificação se correlaciona com a espera no seu ambiente; não demonstra que produção pode ficar sem fallback.

Um relato de usuário em Dify 1.17.1, langgenius/agent 0.0.33 e graphon 0.7.0 registrou 805 eventos message entre 29,819 e 30,035 segundos com Fail Branch, intervalo de 0,216 segundo. No controle padrão, 866 mensagens ficaram entre 3,397 e 40,991 segundos. agent_log continuou chegando durante a execução. É evidência de um ambiente específico, não uma regra para todas as versões nem uma confirmação de versão corrigida.

Controle D: altere somente a ordem do Answer

Primeiro deixe o Answer referenciar apenas o text do Agent. Depois adicione as outras variáveis uma por vez. Se uma variável lenta vier antes do texto, o Answer pode esperar mesmo com o Agent gerando. Coloque a variável de texto antes ou use vários nós Answer, suportados oficialmente para entregar conteúdo em fases.

Registre o horário real com uma chamada mínima

Chame a API publicada do Chatflow em vez de avaliar apenas a animação do navegador. O guia de streaming do Dify define response_mode: "streaming" como uma resposta SSE. No Chatflow, a resposta visível é a concatenação ordenada dos eventos message. agent_log é um log informativo de etapa do Agent: comprova que o canal está ativo, não que o texto chegou ao usuário.

Em cada execução, obtenha o Service API Base e o App API Key na página API Access da cópia de teste. Informe o Base completo, normalmente com /v1; o script acrescenta /chat-messages. Quando for possível fixar a versão, coloque em DIFY_WORKFLOW_ID o ID da versão publicada naquela rodada. Deixe em branco apenas para usar intencionalmente a versão publicada mais recente. Em ambos os casos, publique antes e inclua a versão em DIFY_RUN_LABEL.

Use {} em DIFY_TEST_INPUTS_JSON apenas se a cópia não tiver entradas personalizadas obrigatórias. Se Start/User Input definir campos obrigatórios, envie nas quatro rodadas os mesmos valores JSON sanitizados, com os nomes e tipos exatos dos parâmetros.

No Bash, leia o App API Key sem exibi-lo e informe os parâmetros da rodada:

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

Salve o conteúdo abaixo 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)

O script não grava o texto da resposta. Ele registra horários, ID/título do nó, data.status, tamanho do texto e, em um evento error, os campos superiores status e code, substituindo message por um marcador de tamanho. Ele diferencia succeeded, failed, partial-succeeded, stopped, paused, um error superior e um stream encerrado sem evento terminal.

Os nomes dos arquivos usam DIFY_RUN_LABEL. Se a conexão for interrompida ou o JSON do SSE não puder ser analisado, os horários já capturados continuam salvos, mas run_outcome vira incomplete e complete_for_timing_acceptance fica false. Esses intervalos servem apenas para diagnóstico. Um controle normal só passa na aceitação temporal após message_end e workflow_finished.data.status=succeeded. Também é capturada a exceção http.client.IncompleteRead causada pelo truncamento de SSE com transferência chunked em HTTP/1.1: os horários dos eventos recebidos permanecem no resumo, acceptance_checks_passed=false e o corpo parcial da resposta contido na exceção não é exibido.

Compare quatro horários

1. Primeiro token upstream. Registre a métrica do provider ou gateway próprio quando existir. Sem ela, o controle LLM → Answer separa a rota de forma prática.
2. Primeiro evento SSE com dados. Um workflow_started cedo confirma que a execução foi aceita, não que o texto esteja sendo entregue aos poucos.
3. Primeiro agent_log. Logs antecipados mostram que as etapas do Agent atravessam a conexão. Se o log chega e o texto espera, provavelmente o canal inteiro não está retido como um único objeto.
4. Primeiro e último message. Esses tempos representam a experiência do usuário. Compare first_s, last_s e span_s. Muitos eventos comprimidos em um intervalo mínimo comprovam a rajada final.

Um Chatflow bem-sucedido deve emitir message_end e depois workflow_finished com data.status=succeeded. O contrato oficial de SSE lista succeeded, failed, partial-succeeded e stopped para workflow_finished.data.status; registre-os separadamente. Uma pausa para entrada humana, por outro lado, emite human_input_required e depois o evento separado workflow_paused, a partir do qual o script registra run_outcome=paused. Não use a enumeração de status de um registro de execução como contrato do evento workflow_finished. Em um evento error, status, tipo e mensagem ficam no nível superior; o script mostra status, code e um message sanitizado. Se a conexão terminar sem evento final ou houver interrupção de transporte/análise JSON, o resumo será incomplete. Mesmo com alguns message, o intervalo não comprova melhora enquanto complete_for_timing_acceptance não for true.

Escolha a mudança correspondente ao resultado

O controle LLM também espera: corrija primeiro o upstream

Confirme Model ID, endpoint, streaming do provider e gateways intermediários antes de voltar ao Agent. Se BetterToken for sua linha de base opcional OpenAI-compatible, o guia de Dify recomenda validar primeiro um texto curto sem ferramentas e ativar tool calls apenas após confirmar suporte do modelo e endpoint. Isso isola a camada do modelo, mas não garante que nós do Dify não agrupem a saída.

Todos os eventos chegam tarde: inspecione o transporte

Repita a mesma solicitação reduzindo CDN, proxy reverso e gateways extras. Se a conexão direta ao Dify transmite progressivamente e um intermediário volta a agrupar eventos, verifique repasse SSE e buffering nessa camada. Não troque o modelo ao mesmo tempo.

Apenas Fail Branch agrupa os message

Restrinja a conclusão à sua versão do Dify, plugin Agent e configuração. Preserve o fallback de produção por padrão. Se a entrega ao vivo tiver prioridade, projete e valide outra recuperação antes de migrar; um único controle não justifica remover alertas, chamadas de backup ou logs de erro.

Antes de atualizar o Dify ou o plugin Agent, salve uma execução normal completa e uma verificação de falha controlada. Depois da atualização, publique a cópia e repita com o mesmo prompt, modelo, inputs sanitizados e método de registro da versão. Uma linha do tempo de falha, sozinha, não conclui o teste: o nó alvo deve realmente falhar, o Fail Branch e o Answer de fallback devem executar, e o usuário deve receber uma mensagem sanitizada.

Reordenar o Answer recupera o streaming

O modelo upstream pode ter feito streaming o tempo todo enquanto o Answer aguardava uma variável anterior. Mantenha a nova ordem ou os Answers em etapas e faça a verificação abaixo. error_type e error_message podem orientar rotas e logs internos, mas o Answer visível deve usar texto sanitizado, não detalhes brutos.

Verifique uma falha controlada somente na cópia de teste

Prefira uma condição existente e repetível, sem dados reais, que faça o nó alvo exato emitir um resultado de erro em node_finished; um erro tratado por Fail Branch pode ter data.status=exception, portanto não se deve exigir failed. Um erro interno de ferramenta, um agent_log ou log de modelo não prova, por si só, que o nó Agent falhou.

No código dessas versões, graphon 0.7.0 transforma uma falha tratada por FAIL_BRANCH em NodeRunExceptionEvent com status exception. Dify 1.17.1 a emite como node_finished.data.status=exception, enquanto uma falha de nó não tratada tem status failed. O campo target_node_failed do script aceita exception ou failed como evidência de erro no próprio nó alvo, não de falha de todo o workflow nem de sucesso do fallback. A validação continua exigindo que o nó de fallback correspondente tenha sucesso, que o usuário receba uma resposta de fallback sanitizada e que o resultado final seja um dos permitidos abaixo.

Se não houver condição adequada, adicione temporariamente um nó HTTP Request em uma rota de diagnóstico da cópia, use https://fail-branch-check.invalid/ e defina timeout curto. .invalid é reservado para nomes certamente inválidos. Configure o tratamento desse nó como Fail Branch e ligue a ramificação laranja ao Answer de fallback. Isso valida a ligação Fail Branch/Answer do nó HTTP temporário; não substitui uma falha do próprio Agent quando ele é o alvo da aceitação.

1. Publique a cópia após a mudança temporária. Registre a versão publicada, o ID do nó alvo, o ID do Answer de fallback e uma frase curta não sensível presente apenas no fallback.
2. Use o Base e a Key da cópia, os mesmos inputs sanitizados e essa versão. Defina os três valores e execute o mesmo 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. O resumo deve mostrar target_node_failed: true, fallback_node_succeeded: true, expected_fallback_seen: true e failure_validation_passed: true. O workflow pode terminar como succeeded ou partial-succeeded, mas não failed, stopped, paused nem incomplete.
4. Remova o nó temporário ou restaure a condição original, as conexões e a configuração normal; publique novamente. Limpe as três variáveis e repita a solicitação normal. A execução restaurada deve mostrar complete_for_timing_acceptance: true antes de comparar novamente o primeiro e o último message.

Aceite a correção somente quando streaming e fallback passarem

Não aceite apenas porque o primeiro trecho ficou mais rápido. Guarde dois resumos: uma solicitação normal completa, com message distribuídos durante a geração, e uma falha controlada que prove a falha do nó alvo, a execução do Fail Branch/Answer de fallback e a entrega de um fallback sanitizado. Erro de ferramenta ou log do Agent isolado, ausência da falha do nó, ausência de fallback visível ou falta de nova execução normal após restaurar a cópia impedem a aceitação.