mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
fix(client): exclude prior turns from resumed stream events (#5811)
* fix(client): exclude prior turns from embedded stream events * docs(client): tighten streaming guidance --------- Co-authored-by: xbzz1018 <186267668+xbzz1018@users.noreply.github.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
xbzz1018
GitHub
parent
83c42a61e7
commit
29dbce45a1
@ -2018,6 +2018,8 @@ client.get_goal("thread-1") # {"goal": {...}} or {"goal": None}
|
||||
client.clear_goal("thread-1")
|
||||
```
|
||||
|
||||
When continuing a thread, `values` still contains the full conversation state, while `messages-tuple` and `end.usage` cover only the current turn.
|
||||
|
||||
The HTTP Gateway accepts `values`, `messages-tuple`, `updates`, `debug`, `tasks`, `checkpoints`, and `custom` stream modes. Unsupported modes such as `messages` and `events`, unsupported non-default run options such as webhooks, delayed execution, or `multitask_strategy="enqueue"`, and undeclared SDK options such as checkpoint durability overrides return `422` before execution instead of being silently ignored or downgraded.
|
||||
|
||||
All dict-returning methods are validated against Gateway Pydantic response models in CI (`TestGatewayConformance`), ensuring the embedded client stays in sync with the HTTP API schemas. See `backend/packages/harness/deerflow/client.py` for full API documentation.
|
||||
|
||||
@ -8,7 +8,7 @@
|
||||
|
||||
- DeerFlow 有**两条并行**的流式路径：**Gateway 路径**（async / HTTP SSE / JSON 序列化）服务浏览器和 IM 渠道；**DeerFlowClient 路径**（sync / in-process / 原生 LangChain 对象）服务 Jupyter、脚本、测试。它们**无法合并**——消费者模型不同。
|
||||
- 两条路径都从 `create_agent()` 工厂出发，核心都是订阅 LangGraph 的 `stream_mode=["values", "messages", "custom"]`。`values` 是节点级 state 快照，`messages` 是 LLM token 级 delta，`custom` 是显式 `StreamWriter` 事件。DeerFlow 内置 custom 事件同时通过 callback dispatch 暴露为 `astream_events(version="v2")` 的 `on_custom_event`，供 AG-UI 等 callback 型消费者使用。**这些接口不是详细程度的梯度，而是独立事件源**，消费者必须订阅自己需要的接口。
|
||||
- 嵌入式 client 为每个 `stream()` 调用维护三个 `set[str]`：`seen_ids` / `streamed_ids` / `counted_usage_ids`。三者看起来相似但管理**三个独立的不变式**，不能合并。
|
||||
- 嵌入式 client 为每个 `stream()` 调用维护四个 `set[str]`：`seen_ids` / `streamed_ids` / `counted_usage_ids` / `historical_message_ids`。前三者分别负责去重与用量幂等；最后一个隔离续聊时的历史消息。
|
||||
|
||||
---
|
||||
|
||||
@ -219,27 +219,29 @@ return "".join(chunks.get(last_id, ()))
|
||||
|
||||
---
|
||||
|
||||
## 三个 id set 为什么不能合并
|
||||
## 消息 id set 为什么不能合并
|
||||
|
||||
`DeerFlowClient.stream()` 在一次调用生命周期内维护三个 `set[str]`：
|
||||
`DeerFlowClient.stream()` 在一次调用生命周期内维护四个 `set[str]`：
|
||||
|
||||
```python
|
||||
seen_ids: set[str] = set() # values 路径内部 dedup
|
||||
streamed_ids: set[str] = set() # messages → values 跨模式 dedup
|
||||
counted_usage_ids: set[str] = set() # usage_metadata 幂等计数
|
||||
historical_message_ids: set[str] = set() # 续聊时隔离先前回合
|
||||
```
|
||||
|
||||
乍看像是"三份几乎一样的东西"，实际每个管**不同的不变式**。
|
||||
乍看用途相近，实际每个管**不同的不变式**。
|
||||
|
||||
| Set | 负责的不变式 | 被谁填充 | 被谁查询 |
|
||||
|---|---|---|---|
|
||||
| `seen_ids` | 连续两个 `values` 快照里同一条 message 只生成一个 `messages-tuple` 事件 | values 分支每处理一条消息就加入 | values 分支处理下一条消息前检查 |
|
||||
| `streamed_ids` | 如果一条消息已经通过 `messages` 模式 token 级流过，values 快照到达时**不要**再合成一次完整 `messages-tuple` | messages 分支每发一个 AI/tool 事件就加入 | values 分支看到消息时检查 |
|
||||
| `counted_usage_ids` | 同一个 `usage_metadata` 在 messages 末尾 chunk 和 values 快照的 final AIMessage 里各带一份，**累计总量只算一次** | `_account_usage()` 每次接受 usage 就加入 | `_account_usage()` 每次调用时检查 |
|
||||
| `historical_message_ids` | 续聊时旧消息不再作为本轮增量事件，也不计入本轮 `end.usage` | values 快照里当前用户消息之前的 id | messages 和 values 分支发射前检查 |
|
||||
|
||||
### 为什么不能只用一个 set
|
||||
|
||||
关键观察：**同一个 message id 在这三个 set 里的加入时机不同**。
|
||||
关键观察：`seen_ids`、`streamed_ids`、`counted_usage_ids` 对**本轮消息**的加入时机不同；`historical_message_ids` 专门划定旧回合边界。
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
|
||||
@ -58,9 +58,9 @@ drift.
|
||||
- `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"` — 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"` — AI text **deltas** (concatenate per `id`); emit tool calls/results once each, preserving non-`None` native result `artifact`
|
||||
- `"messages-tuple"` — current-turn AI text **deltas** by `id` and each tool call/result once; excludes resumed history and preserves result `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)
|
||||
- `"end"` — current-turn cumulative `usage`, counted once per message id
|
||||
- **Custom-event invariant** — use `emit_custom_event` / `aemit_custom_event`, never `StreamWriter` alone. Built-in payloads require a non-empty string `type`; typeless payloads stay writer-only, absent from `astream_events`. The writer runs first and is authoritative for Gateway/Web UI/embedded clients; best-effort callbacks must not break it. Async graph hooks must await the async helper, never dispatch synchronously on a running event loop.
|
||||
- Lazy graph creation uses `create_agent()` + `build_middlewares()`.
|
||||
- Cache graphs by storage `user_id` and the unordered set of named-agent `mcp_plugins`. `stream()` materializes `user_id` before worker/loop boundaries in every auth mode.
|
||||
|
||||
@ -858,8 +858,10 @@ class DeerFlowClient:
|
||||
|
||||
Tool calls and tool results are still emitted once per logical
|
||||
message. ``values`` events continue to carry full state snapshots
|
||||
after each graph node finishes; AI text already delivered via the
|
||||
``messages`` stream is **not** re-synthesized from the snapshot to
|
||||
after each graph node finishes. On resumed threads, historical messages
|
||||
remain in those snapshots, but are not emitted again as per-message
|
||||
events or counted in this turn's usage. AI text already delivered via
|
||||
the ``messages`` stream is **not** re-synthesized from the snapshot to
|
||||
avoid duplicate deliveries. When a later node replaces a delivered AI
|
||||
message under the same id and appends to its text (a guard's stop
|
||||
notice), only the appended text is emitted, as one more delta for that
|
||||
@ -999,6 +1001,9 @@ class DeerFlowClient:
|
||||
# Cross-mode handoff: ids already streamed via LangGraph ``messages``
|
||||
# mode so the ``values`` path skips re-synthesis of the same message.
|
||||
streamed_ids: set[str] = set()
|
||||
# A resumed thread's values snapshots include prior turns. They remain
|
||||
# in the full-state event, but must not become new deltas or usage.
|
||||
historical_message_ids: set[str] = set()
|
||||
# AI messages whose tool calls arrived as streamed fragments. The
|
||||
# arguments only parse once the message is complete, so their
|
||||
# tool_calls event is emitted from the values snapshot instead.
|
||||
@ -1076,6 +1081,8 @@ class DeerFlowClient:
|
||||
msg_chunk = chunk
|
||||
|
||||
msg_id = getattr(msg_chunk, "id", None)
|
||||
if msg_id and msg_id in historical_message_ids:
|
||||
continue
|
||||
|
||||
if isinstance(msg_chunk, AIMessage):
|
||||
text = self._extract_text(msg_chunk.content)
|
||||
@ -1120,8 +1127,17 @@ class DeerFlowClient:
|
||||
# mode == "values"
|
||||
messages = chunk.get("messages", [])
|
||||
|
||||
for msg in messages:
|
||||
current_user_index = next(
|
||||
(index for index, msg in enumerate(messages) if isinstance(msg, HumanMessage) and (getattr(msg, "additional_kwargs", None) or {}).get("run_id") == run_id),
|
||||
None,
|
||||
)
|
||||
if current_user_index is not None:
|
||||
historical_message_ids.update(msg_id for msg in messages[:current_user_index] if (msg_id := getattr(msg, "id", None)))
|
||||
|
||||
for index, msg in enumerate(messages):
|
||||
msg_id = getattr(msg, "id", None)
|
||||
if (current_user_index is not None and index < current_user_index) or (msg_id and msg_id in historical_message_ids):
|
||||
continue
|
||||
if msg_id and msg_id in seen_messages:
|
||||
if seen_messages[msg_id] is msg:
|
||||
continue
|
||||
|
||||
@ -437,6 +437,44 @@ class TestStream:
|
||||
assert first_args[0]["messages"][0].additional_kwargs["run_id"] == first_run_id
|
||||
assert second_args[0]["messages"][0].additional_kwargs["run_id"] == second_run_id
|
||||
|
||||
def test_resumed_stream_does_not_reemit_history_or_count_old_usage(self, client):
|
||||
"""Only messages generated in this turn belong in the delta stream and usage."""
|
||||
old_usage = {"input_tokens": 7, "output_tokens": 3, "total_tokens": 10}
|
||||
new_usage = {"input_tokens": 11, "output_tokens": 4, "total_tokens": 15}
|
||||
old_user = HumanMessage(content="first turn", id="h-old", additional_kwargs={"run_id": "old-run"})
|
||||
old_ai = AIMessage(
|
||||
content="",
|
||||
id="ai-old",
|
||||
tool_calls=[{"name": "ls", "args": {"path": "/mnt/user-data/workspace"}, "id": "call-old"}],
|
||||
usage_metadata=old_usage,
|
||||
)
|
||||
old_tool = ToolMessage(content="old result", id="tool-old", name="ls", tool_call_id="call-old")
|
||||
new_ai = AIMessage(content="new answer", id="ai-new", usage_metadata=new_usage)
|
||||
|
||||
def stream_turn(state, *, context, **_kwargs):
|
||||
current_user = HumanMessage(
|
||||
content=state["messages"][0].content,
|
||||
id="h-current",
|
||||
additional_kwargs={"run_id": context["run_id"]},
|
||||
)
|
||||
history = [old_user, old_ai, old_tool, current_user]
|
||||
return iter(
|
||||
[
|
||||
("values", {"messages": history}),
|
||||
("messages", (AIMessageChunk(content="new answer", id="ai-new", usage_metadata=new_usage), {})),
|
||||
("values", {"messages": [*history, new_ai]}),
|
||||
]
|
||||
)
|
||||
|
||||
agent = MagicMock()
|
||||
agent.stream.side_effect = stream_turn
|
||||
with patch.object(client, "_ensure_agent"), patch.object(client, "_agent", agent):
|
||||
events = list(client.stream("second turn", thread_id="t-resumed"))
|
||||
|
||||
assert {event.data.get("id") for event in events if event.type == "messages-tuple"} == {"ai-new"}
|
||||
assert events[-1].data["usage"] == new_usage
|
||||
assert [message["id"] for message in next(event.data for event in events if event.type == "values")["messages"]] == ["h-old", "ai-old", "tool-old", "h-current"]
|
||||
|
||||
def test_custom_mode_is_normalized_to_string(self, client):
|
||||
"""stream() forwards custom events even when the mode is not a plain string."""
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user