diff --git a/README.md b/README.md index ab567b5f3..315864035 100644 --- a/README.md +++ b/README.md @@ -1456,6 +1456,8 @@ DeerFlow is model-agnostic — it works with any LLM that implements the OpenAI- ## 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: Thread IDs may be supplied by callers and do not have to be UUIDs. Explicit diff --git a/backend/packages/harness/deerflow/AGENTS.md b/backend/packages/harness/deerflow/AGENTS.md index 279542f2d..99e47940e 100644 --- a/backend/packages/harness/deerflow/AGENTS.md +++ b/backend/packages/harness/deerflow/AGENTS.md @@ -53,7 +53,7 @@ drift. **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; 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` - `"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) diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index 745ec6892..fe0f48f0d 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/backend/packages/harness/deerflow/client.py @@ -127,7 +127,7 @@ class StreamEvent: """A single event from the streaming agent response. 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). - ``"end"``: Stream finished. @@ -853,7 +853,7 @@ class DeerFlowClient: Yields: 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="messages-tuple" data={"type": "ai", "content": , "id": str} - type="messages-tuple" data={"type": "ai", "content": , "id": str, "usage_metadata": {...}} @@ -1102,6 +1102,7 @@ class DeerFlowClient: type="values", data={ "title": chunk.get("title"), + "summary_text": chunk.get("summary_text"), "messages": [self._serialize_message(m) for m in messages], "artifacts": chunk.get("artifacts", []), }, diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py index 51a8baca0..cf48ef981 100644 --- a/backend/tests/test_client.py +++ b/backend/tests/test_client.py @@ -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.""" agent = MagicMock() agent.stream.return_value = iter(chunks) @@ -496,6 +496,33 @@ class TestStream: assert values_events[-1].data["title"] == "Greeting" 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): """Messages with the same id are not emitted twice.""" ai = AIMessage(content="Hello!", id="ai-1") @@ -902,6 +929,7 @@ class TestStream: "values", { "title": None, + "summary_text": None, "messages": [ {"type": "human", "content": "hi", "id": "h-1"}, {"type": "ai", "content": "Hello", "id": "ai-1", "usage_metadata": usage},