diff --git a/README.md b/README.md index 025531244e..8feddb3728 100644 --- a/README.md +++ b/README.md @@ -2786,6 +2786,9 @@ are checked recursively as well. ## Embedded Python Client +`chat()` and `stream()` preserve text appended by model middleware before the +first state snapshot, emitting only the suffix after the streamed text. + For `DeerFlowClient(agent_name="researcher")`, the named agent's `mcp_plugins` selection applies to both the lead agent and its `task` / `batch_task` delegations: `null` inherits all enabled MCP plugins, `[]` selects none, and diff --git a/backend/packages/harness/deerflow/AGENTS.md b/backend/packages/harness/deerflow/AGENTS.md index 25879c0cb1..3b213e2774 100644 --- a/backend/packages/harness/deerflow/AGENTS.md +++ b/backend/packages/harness/deerflow/AGENTS.md @@ -67,6 +67,7 @@ drift. **Agent Conversation**: - `chat(message, thread_id)` — synchronous; returns final AI text. +- The first `values` snapshot reconciles any suffix appended to already-streamed AI text by a model wrapper; unchanged text and non-prefix replacements are not re-emitted. Later snapshots retain the same append-only rule. - `stream(message, thread_id)` — subscribes to LangGraph `stream_mode=["values", "messages", "custom"]` and yields `StreamEvent`: - `"values"` — state snapshot (title, messages, artifacts, summary_text). Always forward `summary_text` (current summary or `None`), including unchanged values/resets. Never re-emit AI text delivered via `messages`; serialized `ToolMessage` entries retain non-`None` native `artifact` - `"messages-tuple"` — current-turn AI text **deltas** by `id` and each tool call/result once; excludes resumed history and preserves result `artifact` diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index be54b91425..a520f23226 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -1237,13 +1237,19 @@ class DeerFlowClient: if msg_id: seen_messages[msg_id] = msg - # Already streamed via ``messages`` mode; only (defensively) - # capture usage here and skip re-synthesizing the event. + # Already streamed via ``messages`` mode; emit only text added + # before this first snapshot, plus unsent tool calls/metadata. if msg_id and msg_id in streamed_ids: if isinstance(msg, AIMessage): _account_usage(msg_id, getattr(msg, "usage_metadata", None)) additional_kwargs = self._serialize_additional_kwargs(msg) additional_kwargs_delta = _unsent_additional_kwargs(msg_id, additional_kwargs) + text = self._extract_text(msg.content) + sent_text = sent_text_by_id.get(msg_id, "") + if len(text) > len(sent_text) and text.startswith(sent_text): + sent_text_by_id[msg_id] = text + yield self._ai_text_event(msg_id, text[len(sent_text) :], None, additional_kwargs_delta) + additional_kwargs_delta = None if msg_id in pending_tool_call_ids and msg.tool_calls: pending_tool_call_ids.discard(msg_id) yield self._ai_tool_calls_event(msg_id, msg.tool_calls, additional_kwargs_delta) diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py index fcdfb43ada..75916600a2 100644 --- a/backend/tests/test_client.py +++ b/backend/tests/test_client.py @@ -811,6 +811,28 @@ class TestStream: assert [event.data["content"] for event in _ai_events(events)] == ["Checking again.", "\n\n[FORCED STOP] Repeated tool calls exceeded the safety limit."] + @pytest.mark.parametrize("snapshot_text, expected_delta", [("Hello\n\n附加说明", "\n\n附加说明"), ("Hello", ""), ("Rewritten answer", "")]) + def test_first_streamed_snapshot_preserves_text_and_metadata_once(self, client, snapshot_text, expected_delta): + call = {"name": "bash", "args": {"command": "ls"}, "id": "call-1"} + message = AIMessage(content=snapshot_text, id="ai-1", tool_calls=[call], additional_kwargs={"note": "added"}) + agent = _make_agent_mock( + [ + ("messages", (AIMessageChunk(content="Hello", id="ai-1"), {})), + ("messages", (AIMessageChunk(content="", id="ai-1", tool_call_chunks=[{"name": "bash", "args": '{"command":"ls"}', "id": "call-1", "index": 0}]), {})), + ("values", {"messages": [message]}), + ("values", {"messages": [message.model_copy()]}), + ] + ) + with patch.object(client, "_ensure_agent"), patch.object(client, "_agent", agent): + events = list(client.stream("hi", thread_id="t-first-streamed-snapshot")) + + ai_events = [event.data for event in events if event.type == "messages-tuple" and event.data.get("type") == "ai"] + assert "".join(event["content"] for event in ai_events) == "Hello" + expected_delta + first_snapshot_index = next(index for index, event in enumerate(events) if event.type == "values") + assert "".join(event.data.get("content", "") for event in events[:first_snapshot_index]) == "Hello" + expected_delta + assert [event["additional_kwargs"] for event in ai_events if event.get("additional_kwargs")] == [{"note": "added"}] + assert [event["tool_calls"] for event in ai_events if event.get("tool_calls")] == [[call]] + def test_stream_does_not_resend_a_replacement_that_does_not_extend_the_sent_text(self, client): """Only appended text is sent; a rewrite of what was already sent would duplicate output.""" sent = AIMessage(content="Let me look.", id="ai-1") @@ -1260,6 +1282,26 @@ class TestStream: class TestChat: + @pytest.mark.parametrize("suffix", ["", "\n\n附加说明"]) + def test_returns_text_appended_before_the_first_values_snapshot(self, client, suffix): + """A model wrapper can extend streamed text before the model node commits.""" + from langchain.agents import create_agent + from langchain.agents.middleware import AgentMiddleware, ModelResponse + from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel + + class AppendTextMiddleware(AgentMiddleware): + def wrap_model_call(self, request, handler): + response = handler(request) + message = response.result[0] + return ModelResponse(result=[message.model_copy(update={"content": message.content + suffix})]) + + # A stable ID connects streamed text to the first committed snapshot; + # id-less responses would bypass reconciliation and mask a regression. + model = FakeMessagesListChatModel(responses=[AIMessage(content="Hello", id="ai-1")]) + graph = create_agent(model=model, middleware=[AppendTextMiddleware()], state_schema=ThreadState) + with patch.object(client, "_ensure_agent"), patch.object(client, "_agent", graph): + assert client.chat("hi", thread_id="t-first-snapshot") == "Hello" + suffix + @pytest.mark.parametrize( ("events", "expected"), [