mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-27 08:28:00 +00:00
391 lines
15 KiB
Python
391 lines
15 KiB
Python
"""Pure view-state reducer for the DeerFlow TUI.
|
|
|
|
This module has **no** Textual / rendering dependency. It models the visible
|
|
conversation as an immutable list of typed rows and a small set of actions,
|
|
and exposes a single pure ``reduce(state, action) -> state`` function.
|
|
|
|
Keeping this layer pure makes the interesting behaviour (streaming deltas,
|
|
tool cards, error rows) testable with plain ``pytest`` and a handful of
|
|
synthetic actions, independent of any terminal.
|
|
|
|
The runtime bridge (``deerflow.tui.runtime``) is responsible for translating
|
|
``DeerFlowClient`` ``StreamEvent`` objects into these actions; the Textual app
|
|
renders ``ViewState`` into widgets. Both sides depend on this module, not on
|
|
each other.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field, replace
|
|
from typing import Literal
|
|
|
|
from .message_format import format_tool_detail, format_tool_result, summarize_tool_title
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Rows — the immutable units the transcript is built from.
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UserRow:
|
|
text: str
|
|
kind: Literal["user"] = "user"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AssistantRow:
|
|
text: str
|
|
id: str | None = None
|
|
error: bool = False
|
|
kind: Literal["assistant"] = "assistant"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ToolRow:
|
|
tool_call_id: str
|
|
tool_name: str
|
|
title: str
|
|
detail: str = ""
|
|
result: str = ""
|
|
status: Literal["running", "ok", "error"] = "running"
|
|
kind: Literal["tool"] = "tool"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SystemRow:
|
|
text: str
|
|
tone: Literal["info", "error"] = "info"
|
|
kind: Literal["system"] = "system"
|
|
|
|
|
|
Row = UserRow | AssistantRow | ToolRow | SystemRow
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Actions — the only ways the state can change.
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UserSubmitted:
|
|
text: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RunStarted:
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RunEnded:
|
|
usage: dict | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AssistantDelta:
|
|
id: str
|
|
text: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AssistantError:
|
|
text: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ToolStarted:
|
|
tool_call_id: str
|
|
tool_name: str
|
|
args: dict = field(default_factory=dict)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ToolResult:
|
|
tool_call_id: str
|
|
content: str
|
|
is_error: bool = False
|
|
tool_name: str = ""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SystemMessage:
|
|
text: str
|
|
tone: Literal["info", "error"] = "info"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ThreadTitle:
|
|
title: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ClearRows:
|
|
pass
|
|
|
|
|
|
Action = UserSubmitted | RunStarted | RunEnded | AssistantDelta | AssistantError | ToolStarted | ToolResult | SystemMessage | ThreadTitle | ClearRows
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# State.
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ViewState:
|
|
rows: tuple[Row, ...] = ()
|
|
streaming: bool = False
|
|
usage: dict | None = None
|
|
title: str | None = None
|
|
# Id of the message currently being generated this turn. Only this row renders
|
|
# as plain text while streaming; everything else (history) stays Markdown.
|
|
streaming_id: str | None = None
|
|
# Row index of the *anonymous* (empty-id) assistant row receiving deltas this
|
|
# turn, if any. A genuine id is a reliable cross-chunk key (see
|
|
# `_apply_assistant_delta`'s whole-transcript id scan), but an empty id ("" —
|
|
# see `runtime._as_str`) is shared by every id-less chunk from every turn, so
|
|
# it cannot be matched the same way: scanning for `row.id == ""` would fold a
|
|
# brand new turn's text into whatever earlier turn's row happened to be
|
|
# id-less too. This index instead pins "this turn's" anonymous row by
|
|
# position, reset alongside `streaming_id` at the start/end of every turn.
|
|
streaming_anonymous_row_index: int | None = None
|
|
|
|
|
|
def initial_state(rows: tuple[Row, ...] = ()) -> ViewState:
|
|
return ViewState(rows=tuple(rows))
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Reducer.
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _append(state: ViewState, row: Row) -> ViewState:
|
|
return replace(state, rows=state.rows + (row,))
|
|
|
|
|
|
def reduce(state: ViewState, action: Action) -> ViewState:
|
|
"""Return a new ``ViewState`` after applying ``action``. Pure."""
|
|
|
|
if isinstance(action, UserSubmitted):
|
|
return _append(state, UserRow(text=action.text))
|
|
|
|
if isinstance(action, RunStarted):
|
|
# New turn: no message is actively streaming yet (the client re-emits
|
|
# prior messages first; those must not be treated as the active one).
|
|
return replace(state, streaming=True, streaming_id=None, streaming_anonymous_row_index=None)
|
|
|
|
if isinstance(action, RunEnded):
|
|
return replace(
|
|
state,
|
|
streaming=False,
|
|
streaming_id=None,
|
|
streaming_anonymous_row_index=None,
|
|
usage=action.usage if action.usage is not None else state.usage,
|
|
)
|
|
|
|
if isinstance(action, AssistantDelta):
|
|
return _apply_assistant_delta(state, action)
|
|
|
|
if isinstance(action, AssistantError):
|
|
return _append(state, AssistantRow(text=action.text, error=True))
|
|
|
|
if isinstance(action, ToolStarted):
|
|
return _apply_tool_started(state, action)
|
|
|
|
if isinstance(action, ToolResult):
|
|
return _apply_tool_result(state, action)
|
|
|
|
if isinstance(action, SystemMessage):
|
|
return _append(state, SystemRow(text=action.text, tone=action.tone))
|
|
|
|
if isinstance(action, ThreadTitle):
|
|
return replace(state, title=action.title)
|
|
|
|
if isinstance(action, ClearRows):
|
|
return replace(state, rows=(), streaming_id=None, streaming_anonymous_row_index=None)
|
|
|
|
return state
|
|
|
|
|
|
def _apply_assistant_delta(state: ViewState, action: AssistantDelta) -> ViewState:
|
|
"""Update the assistant row for this delta, or start a new one.
|
|
|
|
A genuine (non-empty) id is matched anywhere in the transcript, not just
|
|
the most recent assistant row: on a thread with history, the client
|
|
re-emits every prior message on each new turn (its dedup is per-turn), and
|
|
a re-emitted *older* message can arrive after a newer one has started — so
|
|
matching only the tail row would duplicate prior answers.
|
|
|
|
An empty id ("" — some providers/paths never stamp per-chunk ids, see
|
|
``runtime._as_str``) is NOT a reliable key for that same scan: unlike a
|
|
genuine id, it is shared by every id-less chunk from *every* turn, so
|
|
matching `row.id == ""` across the whole transcript would fold a brand
|
|
new turn's text into whatever earlier turn's row happened to be id-less
|
|
too — silently vanishing the new turn's answer into a stale row. Empty-id
|
|
deltas are therefore routed to `_apply_assistant_delta_anonymous`, which
|
|
tracks "this turn's" row by position instead of by id.
|
|
|
|
Updates also merge by content rather than blindly concatenating, to absorb
|
|
full re-sends / cumulative snapshots vs. genuine incremental deltas:
|
|
|
|
* new text == accumulated, or starts with it -> cumulative/re-send: replace
|
|
* accumulated starts with new text -> stale/shorter re-send: keep
|
|
* otherwise -> a real delta: append
|
|
"""
|
|
if not action.id:
|
|
return _apply_assistant_delta_anonymous(state, action)
|
|
|
|
rows = list(state.rows)
|
|
for i, row in enumerate(rows):
|
|
# ``not row.error``: error rows are appended without an id, so they never
|
|
# match here anyway — the guard is belt-and-suspenders to keep an error
|
|
# row from being merged into if a future change ever gives it an id.
|
|
if isinstance(row, AssistantRow) and row.id == action.id and not row.error:
|
|
# Exact re-send of the same full text (e.g. a values snapshot
|
|
# re-emitting history after reconnection): no-op. Only multi-char
|
|
# matches are treated as re-sends so single-char deltas that happen
|
|
# to equal the buffer (CJK reduplication) are NOT mistaken for no-ops.
|
|
if row.text == action.text and len(action.text) > 1:
|
|
return state
|
|
merged = _merge_stream_text(row.text, action.text)
|
|
rows[i] = replace(row, text=merged)
|
|
return _mark_streaming(replace(state, rows=tuple(rows)), action.id)
|
|
return _mark_streaming(_append(state, AssistantRow(text=action.text, id=action.id)), action.id)
|
|
|
|
|
|
def _apply_assistant_delta_anonymous(state: ViewState, action: AssistantDelta) -> ViewState:
|
|
"""Handle an ``AssistantDelta`` whose id is empty (see `_apply_assistant_delta`).
|
|
|
|
Multiple id-less chunks legitimately arrive for a single turn — a provider
|
|
that never stamps per-chunk ids still streams token by token, e.g.
|
|
``"Hel"`` then ``"lo"`` — so the first empty-id delta of a turn starts a
|
|
new row, and later empty-id deltas keep appending to that row (tracked by
|
|
``state.streaming_anonymous_row_index``, reset on every
|
|
``RunStarted``/``RunEnded``/``ClearRows``, not by id, so a later turn
|
|
always starts its own new row instead of matching the previous turn's
|
|
leftover id-less row — the bug this split exists to avoid).
|
|
|
|
The tracked row is only reused while it is still the LAST row in the
|
|
transcript. A genuine id naturally changes across a tool round-trip
|
|
(LangGraph gives the post-tool continuation a new AIMessage id), which is
|
|
why an interleaved ``ToolStarted``/``ToolResult`` already starts a new row
|
|
in the id-keyed path (see `test_assistant_delta_with_new_id_after_tool_
|
|
creates_separate_row`). An empty id has no such natural signal — it is
|
|
always ``""`` before and after the tool call — so this function uses row
|
|
*position* as the substitute: once anything else has been appended (a
|
|
tool card, in practice), the anonymous row is no longer the tail, and the
|
|
next empty-id delta starts a fresh row rather than reaching backward past
|
|
the tool card into stale text.
|
|
"""
|
|
index = state.streaming_anonymous_row_index
|
|
if index is not None and index == len(state.rows) - 1:
|
|
row = state.rows[index]
|
|
if isinstance(row, AssistantRow) and not row.error:
|
|
# Same no-op / merge semantics as the id-keyed path above.
|
|
if row.text == action.text and len(action.text) > 1:
|
|
return state
|
|
rows = list(state.rows)
|
|
merged = _merge_stream_text(row.text, action.text)
|
|
rows[index] = replace(row, text=merged)
|
|
return _mark_streaming_anonymous(replace(state, rows=tuple(rows)), index)
|
|
|
|
new_state = _append(state, AssistantRow(text=action.text, id=action.id))
|
|
return _mark_streaming_anonymous(new_state, len(new_state.rows) - 1)
|
|
|
|
|
|
def _mark_streaming(state: ViewState, message_id: str) -> ViewState:
|
|
"""Record the actively-streaming message id (only while a run is active)."""
|
|
if state.streaming:
|
|
return replace(state, streaming_id=message_id)
|
|
return state
|
|
|
|
|
|
def _mark_streaming_anonymous(state: ViewState, index: int) -> ViewState:
|
|
"""Record the active turn's anonymous-row index (only while a run is active).
|
|
|
|
Deliberately leaves ``streaming_id`` at ``None`` rather than ``""``: unlike
|
|
a genuine id, ``""`` would be shared by every anonymous row across every
|
|
turn, so using it as the render layer's "is this the row actively
|
|
streaming" key (``render.render_transcript``) would flag every past
|
|
anonymous row as actively streaming too, the moment a new one starts. The
|
|
cost is purely cosmetic — an anonymous row never gets the
|
|
raw-text-while-streaming treatment other rows get, only the id-less
|
|
fallback path is affected — in exchange for not reintroducing a cross-turn
|
|
ambiguity into the render layer that this fix removes from ``rows``.
|
|
"""
|
|
if state.streaming:
|
|
return replace(state, streaming_id=None, streaming_anonymous_row_index=index)
|
|
return state
|
|
|
|
|
|
def _merge_stream_text(existing: str, incoming: str) -> str:
|
|
if not existing:
|
|
return incoming
|
|
# Cumulative re-delivery: incoming strictly extends existing.
|
|
if len(incoming) > len(existing) and incoming.startswith(existing):
|
|
return incoming
|
|
# Stale/shorter re-send: existing already contains incoming as a prefix
|
|
# (e.g. a values snapshot re-emitting history that has already been
|
|
# accumulated from deltas). Only treat as stale when strictly shorter.
|
|
if len(existing) > len(incoming) and existing.startswith(incoming):
|
|
return existing
|
|
return existing + incoming # genuine incremental delta
|
|
|
|
|
|
def _apply_tool_started(state: ViewState, action: ToolStarted) -> ViewState:
|
|
"""Create or update a tool card, de-duplicated by ``tool_call_id``.
|
|
|
|
Streaming tool calls arrive as multiple chunks for one call id (name first,
|
|
then growing arguments), and the client may re-emit the call via a values
|
|
snapshot. Chunks with no id are argument-fragment noise and are dropped.
|
|
"""
|
|
if not action.tool_call_id:
|
|
return state
|
|
|
|
rows = list(state.rows)
|
|
for i, row in enumerate(rows):
|
|
if isinstance(row, ToolRow) and row.tool_call_id == action.tool_call_id:
|
|
name = action.tool_name or row.tool_name
|
|
detail = format_tool_detail(name, action.args) or row.detail
|
|
rows[i] = replace(row, tool_name=name, title=summarize_tool_title(name), detail=detail)
|
|
return replace(state, rows=tuple(rows))
|
|
|
|
return _append(
|
|
state,
|
|
ToolRow(
|
|
tool_call_id=action.tool_call_id,
|
|
tool_name=action.tool_name,
|
|
title=summarize_tool_title(action.tool_name),
|
|
detail=format_tool_detail(action.tool_name, action.args),
|
|
status="running",
|
|
),
|
|
)
|
|
|
|
|
|
def _apply_tool_result(state: ViewState, action: ToolResult) -> ViewState:
|
|
if not action.tool_call_id:
|
|
return state
|
|
|
|
rows = list(state.rows)
|
|
for i, row in enumerate(rows):
|
|
if isinstance(row, ToolRow) and row.tool_call_id == action.tool_call_id:
|
|
rows[i] = replace(
|
|
row,
|
|
status="error" if action.is_error else "ok",
|
|
result=format_tool_result(action.content),
|
|
)
|
|
return replace(state, rows=tuple(rows))
|
|
|
|
# No matching tool card (started chunks missed) -> surface the result anyway.
|
|
return _append(
|
|
state,
|
|
ToolRow(
|
|
tool_call_id=action.tool_call_id,
|
|
tool_name=action.tool_name,
|
|
title=summarize_tool_title(action.tool_name),
|
|
status="error" if action.is_error else "ok",
|
|
result=format_tool_result(action.content),
|
|
),
|
|
)
|