diff --git a/README.md b/README.md index de7e0a650f..34f58d5313 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/backend/docs/STREAMING.md b/backend/docs/STREAMING.md index f438b506be..4146306310 100644 --- a/backend/docs/STREAMING.md +++ b/backend/docs/STREAMING.md @@ -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 diff --git a/backend/packages/harness/deerflow/AGENTS.md b/backend/packages/harness/deerflow/AGENTS.md index 921b66b8d1..d1abe445e9 100644 --- a/backend/packages/harness/deerflow/AGENTS.md +++ b/backend/packages/harness/deerflow/AGENTS.md @@ -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. diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index 38bce69db3..08f691be7d 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -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 diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py index f53d7c39d2..9e243b1573 100644 --- a/backend/tests/test_client.py +++ b/backend/tests/test_client.py @@ -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."""