mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-10 14:08:52 +00:00
fix: expose summary_text in embedded client values events (#5249)
* fix: expose context summary in embedded values events * test: cover summary values in mode-tagged streams --------- Co-authored-by: Sami Belhareth <6599699+belharethsami@users.noreply.github.com>
This commit is contained in:
parent
d8ed8160c9
commit
a3848ef155
@ -1456,6 +1456,8 @@ DeerFlow is model-agnostic — it works with any LLM that implements the OpenAI-
|
|||||||
|
|
||||||
## Embedded Python Client
|
## Embedded Python Client
|
||||||
|
|
||||||
|
`DeerFlowClient.stream()` includes `summary_text` in each `values` event. This is the current compacted context summary, or `None` when absent. Consumers can record changes without reading checkpoint internals; repeated snapshots may carry the same summary, and an initial snapshot may already contain one from an earlier turn.
|
||||||
|
|
||||||
DeerFlow can be used as an embedded Python library without running the full HTTP services. The `DeerFlowClient` provides direct in-process access to all agent and Gateway capabilities, returning the same response schemas as the HTTP Gateway API. The HTTP Gateway also exposes `DELETE /api/threads/{thread_id}` to remove DeerFlow-managed local thread data after the LangGraph thread itself has been deleted:
|
DeerFlow can be used as an embedded Python library without running the full HTTP services. The `DeerFlowClient` provides direct in-process access to all agent and Gateway capabilities, returning the same response schemas as the HTTP Gateway API. The HTTP Gateway also exposes `DELETE /api/threads/{thread_id}` to remove DeerFlow-managed local thread data after the LangGraph thread itself has been deleted:
|
||||||
|
|
||||||
Thread IDs may be supplied by callers and do not have to be UUIDs. Explicit
|
Thread IDs may be supplied by callers and do not have to be UUIDs. Explicit
|
||||||
|
|||||||
@ -53,7 +53,7 @@ drift.
|
|||||||
**Agent Conversation**:
|
**Agent Conversation**:
|
||||||
- `chat(message, thread_id)` — synchronous, accumulates streaming deltas per message-id and returns the final AI text
|
- `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`:
|
- `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; serialized `ToolMessage` entries preserve a non-`None` native `artifact`
|
- `"values"` — state snapshot (title, messages, artifacts, summary_text); `summary_text` is the current summary or `None` when absent and is forwarded on every snapshot, including unchanged summaries and resets. 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`
|
- `"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`
|
- `"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"` — stream finished (carries cumulative `usage` counted once per message id)
|
||||||
|
|||||||
@ -127,7 +127,7 @@ class StreamEvent:
|
|||||||
"""A single event from the streaming agent response.
|
"""A single event from the streaming agent response.
|
||||||
|
|
||||||
Event types align with the LangGraph SSE protocol:
|
Event types align with the LangGraph SSE protocol:
|
||||||
- ``"values"``: Full state snapshot (title, messages, artifacts).
|
- ``"values"``: State snapshot (title, messages, artifacts, summary_text).
|
||||||
- ``"messages-tuple"``: Per-message update (AI text, tool calls, tool results).
|
- ``"messages-tuple"``: Per-message update (AI text, tool calls, tool results).
|
||||||
- ``"end"``: Stream finished.
|
- ``"end"``: Stream finished.
|
||||||
|
|
||||||
@ -853,7 +853,7 @@ class DeerFlowClient:
|
|||||||
|
|
||||||
Yields:
|
Yields:
|
||||||
StreamEvent with one of:
|
StreamEvent with one of:
|
||||||
- type="values" data={"title": str|None, "messages": [...], "artifacts": [...]}
|
- type="values" data={"title": str|None, "messages": [...], "artifacts": [...], "summary_text": str|None}
|
||||||
- type="custom" data={...}
|
- type="custom" data={...}
|
||||||
- type="messages-tuple" data={"type": "ai", "content": <delta>, "id": str}
|
- type="messages-tuple" data={"type": "ai", "content": <delta>, "id": str}
|
||||||
- type="messages-tuple" data={"type": "ai", "content": <delta>, "id": str, "usage_metadata": {...}}
|
- type="messages-tuple" data={"type": "ai", "content": <delta>, "id": str, "usage_metadata": {...}}
|
||||||
@ -1102,6 +1102,7 @@ class DeerFlowClient:
|
|||||||
type="values",
|
type="values",
|
||||||
data={
|
data={
|
||||||
"title": chunk.get("title"),
|
"title": chunk.get("title"),
|
||||||
|
"summary_text": chunk.get("summary_text"),
|
||||||
"messages": [self._serialize_message(m) for m in messages],
|
"messages": [self._serialize_message(m) for m in messages],
|
||||||
"artifacts": chunk.get("artifacts", []),
|
"artifacts": chunk.get("artifacts", []),
|
||||||
},
|
},
|
||||||
|
|||||||
@ -247,7 +247,7 @@ class TestConfigQueries:
|
|||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
def _make_agent_mock(chunks: list[dict]):
|
def _make_agent_mock(chunks: list[dict | tuple[str, dict]]):
|
||||||
"""Create a mock agent whose .stream() yields the given chunks."""
|
"""Create a mock agent whose .stream() yields the given chunks."""
|
||||||
agent = MagicMock()
|
agent = MagicMock()
|
||||||
agent.stream.return_value = iter(chunks)
|
agent.stream.return_value = iter(chunks)
|
||||||
@ -496,6 +496,33 @@ class TestStream:
|
|||||||
assert values_events[-1].data["title"] == "Greeting"
|
assert values_events[-1].data["title"] == "Greeting"
|
||||||
assert "messages" in values_events[-1].data
|
assert "messages" in values_events[-1].data
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("mode_tagged", [False, True], ids=["bare-dict", "mode-tuple"])
|
||||||
|
def test_values_events_preserve_summary_text_updates(self, client, mode_tagged):
|
||||||
|
messages = [HumanMessage(content="hi", id="h-1"), AIMessage(content="ok", id="ai-1")]
|
||||||
|
summaries = [None, "first summary", "first summary", "revised summary", "", None]
|
||||||
|
chunks = [{"messages": messages, "summary_text": summary} for summary in summaries]
|
||||||
|
agent = _make_agent_mock([("values", chunk) for chunk in chunks] if mode_tagged else chunks)
|
||||||
|
|
||||||
|
with patch.object(client, "_ensure_agent"), patch.object(client, "_agent", agent):
|
||||||
|
events = list(client.stream("hi", thread_id="summary-stream"))
|
||||||
|
|
||||||
|
values_events = [event for event in events if event.type == "values"]
|
||||||
|
assert [event.data["summary_text"] for event in values_events] == summaries
|
||||||
|
assert all(len(event.data["messages"]) == 2 for event in values_events)
|
||||||
|
assert len(_ai_events(events)) == 1
|
||||||
|
assert events[-1].type == "end"
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("mode_tagged", [False, True], ids=["bare-dict", "mode-tuple"])
|
||||||
|
def test_values_events_without_summary_expose_none(self, client, mode_tagged):
|
||||||
|
chunk = {"messages": [HumanMessage(content="hi", id="h-1")]}
|
||||||
|
agent = _make_agent_mock([("values", chunk) if mode_tagged else chunk])
|
||||||
|
|
||||||
|
with patch.object(client, "_ensure_agent"), patch.object(client, "_agent", agent):
|
||||||
|
events = list(client.stream("hi", thread_id="no-summary"))
|
||||||
|
|
||||||
|
values_events = [event for event in events if event.type == "values"]
|
||||||
|
assert values_events[0].data["summary_text"] is None
|
||||||
|
|
||||||
def test_deduplication(self, client):
|
def test_deduplication(self, client):
|
||||||
"""Messages with the same id are not emitted twice."""
|
"""Messages with the same id are not emitted twice."""
|
||||||
ai = AIMessage(content="Hello!", id="ai-1")
|
ai = AIMessage(content="Hello!", id="ai-1")
|
||||||
@ -902,6 +929,7 @@ class TestStream:
|
|||||||
"values",
|
"values",
|
||||||
{
|
{
|
||||||
"title": None,
|
"title": None,
|
||||||
|
"summary_text": None,
|
||||||
"messages": [
|
"messages": [
|
||||||
{"type": "human", "content": "hi", "id": "h-1"},
|
{"type": "human", "content": "hi", "id": "h-1"},
|
||||||
{"type": "ai", "content": "Hello", "id": "ai-1", "usage_metadata": usage},
|
{"type": "ai", "content": "Hello", "id": "ai-1", "usage_metadata": usage},
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user