diff --git a/plugins/Runner/dify-agent/README.md b/plugins/Runner/dify-agent/README.md index 4b62497fc..fad5ed7b5 100644 --- a/plugins/Runner/dify-agent/README.md +++ b/plugins/Runner/dify-agent/README.md @@ -35,6 +35,23 @@ Configuration is **static** and should not contain runtime state. Only the follo - `multimodal_input`: yes - `interactions`: yes +## Dify 1.18 Agent Applications + +Use the Dify Agent Runner with `app-type: agent` and the published application's +Service API URL and key. LocalAgent is a separate, model-backed Runner; a Dify +application's Service API is not an OpenAI-compatible model endpoint. + +Dify's new Agent application emits process text as `agent_message` and the +terminal answer as `message`. The Runner treats these as separate streams: once +`message` arrives, its answer replaces the process-text preview instead of being +appended to it. Older agents that emit only `agent_message` continue to work. +Repeated chunks within the same stream are not removed by content equality. + +Legacy agents' empty `agent_thought` events do not flush the answer buffer; +`message_end` still closes the response. For workflows, final `summary` outputs +or explicit Answer-node outputs take precedence over buffered `text_chunk` +events, which are used only when no explicit answer was produced. + ## Workflow Human Input When Dify emits `workflow_paused` with a `human_input_required` reason, the diff --git a/plugins/Runner/dify-agent/components/runner/default.py b/plugins/Runner/dify-agent/components/runner/default.py index 92b9c42b1..ddb946be1 100644 --- a/plugins/Runner/dify-agent/components/runner/default.py +++ b/plugins/Runner/dify-agent/components/runner/default.py @@ -962,6 +962,7 @@ class DefaultRunner(Runner): pending_content = "" mode = "basic" # basic or workflow mode in chat message_count = 0 + received_message = False has_response = False final_conversation_id = conversation_id usage: dict[str, typing.Any] | None = None @@ -1039,6 +1040,12 @@ class DefaultRunner(Runner): return elif event_type == "message" or event_type == "agent_message": + if event_type == "message" and not received_message: + pending_content = "" + message_count = 0 + received_message = True + elif event_type == "agent_message" and received_message: + continue # Accumulate text chunks answer = event.get("answer", "") # Dify deployments send either deltas or cumulative snapshots. @@ -1071,7 +1078,7 @@ class DefaultRunner(Runner): observation = event.get("observation", "") # Skip tool result observations - if tool and observation: + if not tool or observation: continue # Yield accumulated content before tool call @@ -1310,7 +1317,7 @@ class DefaultRunner(Runner): ) # Handle remaining pending content - if pending_content: + if pending_content and not has_response: content, _ = process_thinking_content(pending_content, remove_think) if content: has_response = True diff --git a/plugins/Runner/dify-agent/manifest.yaml b/plugins/Runner/dify-agent/manifest.yaml index 400d66c01..72361226f 100644 --- a/plugins/Runner/dify-agent/manifest.yaml +++ b/plugins/Runner/dify-agent/manifest.yaml @@ -3,7 +3,7 @@ kind: Plugin metadata: author: langbot-team name: DifyAgent - version: 0.2.4 + version: 0.2.5 repository: https://github.com/langbot-app/LangBot/tree/master/plugins/Runner/dify-agent label: en_US: Dify Agent diff --git a/plugins/Runner/tests/test_dify_modes.py b/plugins/Runner/tests/test_dify_modes.py new file mode 100644 index 000000000..340a96d8e --- /dev/null +++ b/plugins/Runner/tests/test_dify_modes.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +import pytest + +from tests.test_native_parity_external import collect, ctx, kind, load_runner + + +def final_texts(results): + return [ + result.data["chunk"]["content"] + for result in results + if kind(result) == "message.delta" and result.data["chunk"].get("is_final") + ] + + +def test_legacy_agent_empty_thought_keeps_answer_until_message_end(): + with load_runner("dify-agent") as (_, runner): + + class Client: + async def chat_messages(self, **kwargs): + yield {"event": "agent_thought", "tool": "", "observation": ""} + yield {"event": "agent_message", "answer": "23"} + yield {"event": "agent_thought", "tool": "", "observation": ""} + yield {"event": "message_end"} + + results = collect(runner._run_chat_or_agent(ctx(), Client(), {}, "hi", "user", "", [], "agent", False)) + assert final_texts(results) == ["23"] + + +@pytest.mark.parametrize("summary", ["23", "final answer", None]) +def test_workflow_terminal_summary_owns_streamed_text(summary): + with load_runner("dify-agent") as (_, runner): + + class Client: + async def workflow_run(self, **kwargs): + yield {"event": "workflow_started", "data": {"workflow_run_id": "workflow"}} + yield {"event": "text_chunk", "data": {"text": "2"}} + yield {"event": "text_chunk", "data": {"text": "3"}} + yield {"event": "workflow_finished", "data": {"outputs": {"summary": summary}}} + + results = collect(runner._run_workflow(ctx(), Client(), {}, "hi", "user", [], False)) + assert final_texts(results) == [summary or "23"] + assert kind(results[-1]) == "run.completed" + + +def test_workflow_answer_node_owns_streamed_text(): + with load_runner("dify-agent") as (_, runner): + + class Client: + async def workflow_run(self, **kwargs): + yield {"event": "text_chunk", "data": {"text": "23"}} + yield {"event": "node_finished", "data": {"node_type": "answer", "outputs": {"answer": "23"}}} + yield {"event": "workflow_finished", "data": {"outputs": {}}} + + results = collect(runner._run_workflow(ctx(), Client(), {}, "hi", "user", [], False)) + assert final_texts(results) == ["23"] + + +@pytest.mark.parametrize("app_type", ["chat", "chatflow"]) +def test_chatflow_answer_snapshot_and_message_end_complete_once(app_type): + with load_runner("dify-agent") as (_, runner): + + class Client: + async def chat_messages(self, **kwargs): + yield {"event": "workflow_started", "data": {"workflow_run_id": "workflow"}} + yield {"event": "message", "answer": "2"} + yield {"event": "message", "answer": "3"} + yield {"event": "node_finished", "data": {"node_type": "answer", "outputs": {"answer": "23"}}} + yield {"event": "message_end"} + yield {"event": "workflow_finished", "data": {"outputs": {"answer": "23"}}} + + results = collect(runner._run_chat_or_agent(ctx(), Client(), {}, "hi", "user", "", [], app_type, False)) + assert final_texts(results) == ["23"] diff --git a/plugins/Runner/tests/test_native_parity_external.py b/plugins/Runner/tests/test_native_parity_external.py index 32c6c6918..fa8bdeee3 100644 --- a/plugins/Runner/tests/test_native_parity_external.py +++ b/plugins/Runner/tests/test_native_parity_external.py @@ -149,6 +149,72 @@ def test_dify_native_cumulative_chat_chunks_do_not_duplicate(): assert contents(results)[-1] == "hello world" +@pytest.mark.parametrize("app_type", ["chat", "agent"]) +@pytest.mark.parametrize( + ("events", "expected"), + [ + ([("agent_message", "23"), ("message", "23")], "23"), + ([("agent_message", "Thinking"), ("message", "Final"), ("message", " answer")], "Final answer"), + ([("agent_message", "2"), ("agent_message", "3"), ("message", "23")], "23"), + ([("agent_message", "ha"), ("agent_message", "ha")], "haha"), + ([("message", "ha"), ("message", "ha")], "haha"), + ([("message", "Final"), ("agent_message", "process")], "Final"), + ([("agent_message", "hel"), ("agent_message", "hello")], "hello"), + ], +) +def test_dify_agent_process_and_final_answer_are_separate(app_type, events, expected): + with load_runner("dify-agent") as (_, runner): + + class Client: + async def chat_messages(self, **kwargs): + for event_type, answer in events: + yield {"event": event_type, "answer": answer, "conversation_id": "external-conversation"} + yield {"event": "message_end"} + + results = collect(runner._run_chat_or_agent(ctx(), Client(), {}, "hi", "user", "", [], app_type, False)) + assert contents(results) == [expected] + updates = [result for result in results if kind(result) == "state.updated"] + assert updates[-1].data["value"] == "external-conversation" + + +def test_dify_agent_final_replaces_streamed_process_snapshot(): + with load_runner("dify-agent") as (_, runner): + + class Client: + async def chat_messages(self, **kwargs): + for _ in range(8): + yield {"event": "agent_message", "answer": "process "} + yield {"event": "message", "answer": "23"} + yield {"event": "message_end"} + + results = collect(runner._run_chat_or_agent(ctx(), Client(), {}, "hi", "user", "", [], "agent", False)) + messages = [result.data["chunk"] for result in results if kind(result) == "message.delta"] + assert messages[0]["is_final"] is False + assert messages[-1]["content"] == "23" + assert sum(bool(message["is_final"]) for message in messages) == 1 + + +def test_dify_118_live_event_order_does_not_repeat_terminal_answer(): + with load_runner("dify-agent") as (_, runner): + + class Client: + async def chat_messages(self, **kwargs): + yield {"event": "agent_thought", "tool": "", "thought": "", "observation": ""} + yield {"event": "agent_message", "answer": "2"} + yield {"event": "agent_thought", "tool": "", "thought": "", "observation": ""} + yield {"event": "agent_message", "answer": "3"} + yield {"event": "message", "answer": "23"} + yield {"event": "message_end"} + + results = collect(runner._run_chat_or_agent(ctx(), Client(), {}, "hi", "user", "", [], "agent", False)) + final = [ + result.data["chunk"]["content"] + for result in results + if kind(result) == "message.delta" and result.data["chunk"]["is_final"] + ] + assert final == ["23"] + + def test_dify_unclosed_think_never_visible_on_pause(): with load_runner("dify-agent") as (m, runner): diff --git a/plugins/catalog.json b/plugins/catalog.json index d829d2b0c..9331ef413 100644 --- a/plugins/catalog.json +++ b/plugins/catalog.json @@ -92,7 +92,7 @@ }, { "id": "langbot-team/DifyAgent", - "version": "0.2.4", + "version": "0.2.5", "category": "Runner", "path": "Runner/dify-agent", "repository": "https://github.com/langbot-app/LangBot/tree/master/plugins/Runner/dify-agent" diff --git a/tests/fixtures/legacy_plugin_identity_sources.json b/tests/fixtures/legacy_plugin_identity_sources.json index 34d3940d7..e820e59e0 100644 --- a/tests/fixtures/legacy_plugin_identity_sources.json +++ b/tests/fixtures/legacy_plugin_identity_sources.json @@ -3,7 +3,7 @@ "revision": "in-tree", "contract": "Current dedicated-workspace and trusted shared-installation identity; see each plugin SHARED_RUNTIME.md.", "files": { - "Runner/dify-agent/components/runner/default.py": "8ca64052fb8df872272726aa060e21ba3b4afa454346b6babcc7e9d2df600980", + "Runner/dify-agent/components/runner/default.py": "1a5901036d73fdeed8966660a539824a8aa37cb4843409cc5de37105ca36aa04", "Runner/dify-agent/pkg/scoped_identity.py": "5d03d0c56aaf3301bc5d594905ecf862151fa2abc73daba1a6596f0fdc16374a", "Runner/dify-agent/pkg/dify_client.py": "607ed8d610d916e5b64ed20699e3de6e2216b238a5fecc9cabcdb75533b8dc56", "Runner/n8n-agent/components/runner/default.py": "2719ec0a8ff779e90e61bd0209653fb9a1eab4b552877cfcbd4d3c9ce803fcf1",