From 0f0955bf7b2ae64ecb5099551b86049c2091a80a Mon Sep 17 00:00:00 2001 From: ShitK <80999801+ShitK@users.noreply.github.com> Date: Sat, 25 Jul 2026 09:47:58 +0800 Subject: [PATCH] fix(client): preserve ToolMessage artifacts (#4422) Co-authored-by: Willem Jiang --- README.md | 4 ++ backend/AGENTS.md | 4 +- backend/docs/STREAMING.md | 3 +- backend/packages/harness/deerflow/client.py | 23 ++++--- backend/tests/test_client.py | 70 +++++++++++++++++++++ 5 files changed, 91 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 33c6fcd12..5ad7acd11 100644 --- a/README.md +++ b/README.md @@ -865,6 +865,10 @@ response = client.chat("Analyze this paper for me", thread_id="my-thread") for event in client.stream("hello"): if event.type == "messages-tuple" and event.data.get("type") == "ai": print(event.data["content"]) + elif event.type == "messages-tuple" and event.data.get("type") == "tool" and "artifact" in event.data: + # Structured tool artifacts (for example, ask_clarification cards) + # are preserved when the ToolMessage provides one. + print(event.data["artifact"]) # Configuration & management — returns Gateway-aligned dicts models = client.list_models() # {"models": [...]} diff --git a/backend/AGENTS.md b/backend/AGENTS.md index ebfb86bc5..923651d61 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -980,8 +980,8 @@ Gateway API endpoints and `DeerFlowClient` methods can modify MCP servers and sk **Agent Conversation**: - `chat(message, thread_id)` — synchronous, accumulates streaming deltas per message-id and returns the final AI text - `stream(message, thread_id)` — subscribes to LangGraph `stream_mode=["values", "messages", "custom"]` and yields `StreamEvent`: - - `"values"` — full state snapshot (title, messages, artifacts); AI text already delivered via `messages` mode is **not** re-synthesized here to avoid duplicate deliveries - - `"messages-tuple"` — per-chunk update: for AI text this is a **delta** (concat per `id` to rebuild the full message); tool calls and tool results are emitted once each + - `"values"` — full state snapshot (title, messages, artifacts); AI text already delivered via `messages` mode is **not** re-synthesized here to avoid duplicate deliveries; serialized `ToolMessage` entries preserve a non-`None` native `artifact` + - `"messages-tuple"` — per-chunk update: for AI text this is a **delta** (concat per `id` to rebuild the full message); tool calls and tool results are emitted once each, and tool results preserve a non-`None` native `artifact` - `"custom"` — forwarded from `StreamWriter`; DeerFlow-built-in custom events are dual-emitted through `deerflow.utils.custom_events`, so `astream_events(version="v2")` consumers also receive one `on_custom_event` with `name=payload["type"]` and the unchanged payload as `data` - `"end"` — stream finished (carries cumulative `usage` counted once per message id) - **Custom-event invariant** — production DeerFlow emitters must use `emit_custom_event` / `aemit_custom_event`, not call `StreamWriter` alone. Every built-in payload must carry a non-empty string `type`; typeless payloads remain writer-only and are intentionally absent from `astream_events`. The writer runs first and remains authoritative for Gateway, Web UI, and embedded-client compatibility; callback dispatch is best-effort and must not break that path. Async graph hooks must await the async helper rather than invoking synchronous dispatch on a running event loop. diff --git a/backend/docs/STREAMING.md b/backend/docs/STREAMING.md index 77b5e2925..5b748ce78 100644 --- a/backend/docs/STREAMING.md +++ b/backend/docs/STREAMING.md @@ -173,7 +173,7 @@ sequenceDiagram - 没有 `RunManager` —— 一次 `stream()` 调用对应一次生命周期，只生成轻量 `run_id` 供 runtime context、tracing 和 per-run middleware 使用。 - 没有 `StreamBridge` —— 直接 `yield`，生产和消费在同一个 Python 调用栈，不需要跨 task 中介。 -- 没有 JSON 序列化 —— `StreamEvent.data` 直接装原生 LangChain 对象（`AIMessage.content`、`usage_metadata` 的 `UsageMetadata` TypedDict）。Jupyter 用户拿到的是真正的类型，不是匿名 dict。 +- 没有 JSON 序列化 —— `StreamEvent.data` 直接装原生 LangChain 值（`AIMessage.content`、`usage_metadata` 的 `UsageMetadata` TypedDict，以及非 `None` 的 `ToolMessage.artifact`）。`messages-tuple` 工具结果和 `values` 快照里的工具消息都会保留 artifact；没有 artifact 的工具消息维持原有字段形状。Jupyter 用户拿到的是真正的类型，不是经过网络序列化后的匿名值。 - 没有 asyncio —— 调用者可以直接 `for event in ...`，不必写 `async for`。 --- @@ -348,6 +348,7 @@ assert "messages" in agent.stream.call_args.kwargs["stream_mode"] | 关心什么 | 看这里 | |---|---| | DeerFlowClient 嵌入式流 | `packages/harness/deerflow/client.py::DeerFlowClient.stream` | +| Embedded ToolMessage artifact 序列化 | `packages/harness/deerflow/client.py::_tool_message_event` / `_serialize_message` | | `chat()` 的 delta 累加器 | `packages/harness/deerflow/client.py::DeerFlowClient.chat` | | Gateway async 流 | `packages/harness/deerflow/runtime/runs/worker.py::run_agent` | | HTTP SSE 帧输出 | `app/gateway/services.py::sse_consumer` / `format_sse` | diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index 757ce6ee7..50a28438d 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -431,16 +431,16 @@ class DeerFlowClient: @staticmethod def _tool_message_event(msg: ToolMessage) -> "StreamEvent": """Build a ``messages-tuple`` tool-result event from a ToolMessage.""" - return StreamEvent( - type="messages-tuple", - data={ - "type": "tool", - "content": DeerFlowClient._extract_text(msg.content), - "name": msg.name, - "tool_call_id": msg.tool_call_id, - "id": msg.id, - }, - ) + data: dict[str, Any] = { + "type": "tool", + "content": DeerFlowClient._extract_text(msg.content), + "name": msg.name, + "tool_call_id": msg.tool_call_id, + "id": msg.id, + } + if (artifact := getattr(msg, "artifact", None)) is not None: + data["artifact"] = artifact + return StreamEvent(type="messages-tuple", data=data) @staticmethod def _serialize_message(msg) -> dict: @@ -464,6 +464,8 @@ class DeerFlowClient: } if additional_kwargs := DeerFlowClient._serialize_additional_kwargs(msg): d["additional_kwargs"] = additional_kwargs + if (artifact := getattr(msg, "artifact", None)) is not None: + d["artifact"] = artifact return d if isinstance(msg, HumanMessage): d = {"type": "human", "content": msg.content, "id": getattr(msg, "id", None)} @@ -803,6 +805,7 @@ class DeerFlowClient: - type="messages-tuple" data={"type": "ai", "content": "", "id": str, "tool_calls": [...]} - type="messages-tuple" data={"type": "ai", "content": "", "id": str, "additional_kwargs": {...}} - type="messages-tuple" data={"type": "tool", "content": str, "name": str, "tool_call_id": str, "id": str} + Tool results also include ``"artifact"`` when the source ToolMessage has a non-None artifact. - type="end" data={"usage": {"input_tokens": int, "output_tokens": int, "total_tokens": int}} """ if thread_id is None: diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py index fab6187c2..ea5718bad 100644 --- a/backend/tests/test_client.py +++ b/backend/tests/test_client.py @@ -691,6 +691,47 @@ class TestStream: assert tool_events[0].data["content"] == "file.txt" assert tool_events[0].data["name"] == "bash" assert tool_events[0].data["tool_call_id"] == "tc-1" + assert "artifact" not in tool_events[0].data + + def test_messages_mode_tool_message_preserves_human_input_artifact(self, client): + """Structured clarification data survives both embedded stream paths.""" + artifact = { + "human_input": { + "request_id": "request-1", + "tool_call_id": "tc-1", + "question": "Which environment should be used?", + "options": [{"label": "Production", "value": "production"}], + } + } + tool_message = ToolMessage( + content="Which environment should be used?", + id="tm-1", + tool_call_id="tc-1", + name="ask_clarification", + artifact=artifact, + ) + agent = MagicMock() + agent.stream.return_value = iter( + [ + ("messages", (tool_message, {})), + ("values", {"messages": [HumanMessage(content="deploy", id="h-1"), tool_message]}), + ] + ) + + with ( + patch.object(client, "_ensure_agent"), + patch.object(client, "_agent", agent), + ): + events = list(client.stream("deploy", thread_id="t-tool-artifact")) + + tool_event = next(event for event in events if event.type == "messages-tuple" and event.data.get("type") == "tool") + values_event = next(event for event in events if event.type == "values") + serialized_tool_message = next(message for message in values_event.data["messages"] if message["type"] == "tool") + + assert tool_event.data["artifact"] == artifact + assert serialized_tool_message["artifact"] == artifact + assert tool_event.data["artifact"] is artifact + assert serialized_tool_message["artifact"] is artifact def test_list_content_blocks(self, client): """stream() handles AIMessage with list-of-blocks content.""" @@ -3658,6 +3699,35 @@ class TestSerializeMessage: result = DeerFlowClient._serialize_message(msg) assert result["type"] == "tool" assert isinstance(result["content"], str) + assert "artifact" not in result + + def test_tool_message_event_preserves_native_artifact(self): + marker = object() + msg = ToolMessage( + content="result", + id="tm-1", + tool_call_id="tc-1", + name="tool", + artifact={"payload": marker}, + ) + + result = DeerFlowClient._tool_message_event(msg) + + assert result.data["artifact"] is msg.artifact + + def test_tool_message_values_serialization_preserves_native_artifact(self): + marker = object() + msg = ToolMessage( + content="result", + id="tm-1", + tool_call_id="tc-1", + name="tool", + artifact={"payload": marker}, + ) + + result = DeerFlowClient._serialize_message(msg) + + assert result["artifact"] is msg.artifact # ===========================================================================