mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-05-23 16:35:59 +00:00
Huixin615
8785658a2e
* fix(agents): preserve todos state across node updates ThreadState.todos had no reducer, so any downstream node returning a partial state without todos was implicitly setting it to None, which LangGraph then used to overwrite the previously streamed value. This caused the to-do list to render correctly during streaming but vanish once streaming completed. Add a merge_todos reducer that keeps the last non-None value, mirroring the merge_artifacts pattern already used in the same file. An explicit empty list is still respected so that 'user cleared todos' works. Tests: 10 new unit tests in tests/test_thread_state_reducers.py covering merge_todos plus regression coverage for merge_artifacts and merge_viewed_images. All 69 thread-related tests pass locally. Closes #3123 * test(agents): add annotation binding regression guard Address Copilot review feedback on #3123: - Add TestThreadStateAnnotations asserting that ThreadState.todos is Annotated with merge_todos. Without this guard, reverting the Annotated[list | None, merge_todos] binding would silently regress #3123 while all existing reducer unit tests continue to pass. - Align test imports to 'from deerflow.agents.thread_state import ...' matching the rest of the backend test suite.
69 lines
2.2 KiB
Python
69 lines
2.2 KiB
Python
from typing import Annotated, NotRequired, TypedDict
|
|
|
|
from langchain.agents import AgentState
|
|
|
|
|
|
class SandboxState(TypedDict):
|
|
sandbox_id: NotRequired[str | None]
|
|
|
|
|
|
class ThreadDataState(TypedDict):
|
|
workspace_path: NotRequired[str | None]
|
|
uploads_path: NotRequired[str | None]
|
|
outputs_path: NotRequired[str | None]
|
|
|
|
|
|
class ViewedImageData(TypedDict):
|
|
base64: str
|
|
mime_type: str
|
|
|
|
|
|
def merge_artifacts(existing: list[str] | None, new: list[str] | None) -> list[str]:
|
|
"""Reducer for artifacts list - merges and deduplicates artifacts."""
|
|
if existing is None:
|
|
return new or []
|
|
if new is None:
|
|
return existing
|
|
# Use dict.fromkeys to deduplicate while preserving order
|
|
return list(dict.fromkeys(existing + new))
|
|
|
|
|
|
def merge_viewed_images(existing: dict[str, ViewedImageData] | None, new: dict[str, ViewedImageData] | None) -> dict[str, ViewedImageData]:
|
|
"""Reducer for viewed_images dict - merges image dictionaries.
|
|
|
|
Special case: If new is an empty dict {}, it clears the existing images.
|
|
This allows middlewares to clear the viewed_images state after processing.
|
|
"""
|
|
if existing is None:
|
|
return new or {}
|
|
if new is None:
|
|
return existing
|
|
# Special case: empty dict means clear all viewed images
|
|
if len(new) == 0:
|
|
return {}
|
|
# Merge dictionaries, new values override existing ones for same keys
|
|
return {**existing, **new}
|
|
|
|
|
|
def merge_todos(existing: list | None, new: list | None) -> list | None:
|
|
"""Reducer for todos list - keeps the last non-None value.
|
|
|
|
Semantics:
|
|
- If `new` is None (node didn't touch todos), preserve `existing`.
|
|
- If `new` is provided (even empty list), it represents an explicit
|
|
update and wins over `existing`.
|
|
"""
|
|
if new is None:
|
|
return existing
|
|
return new
|
|
|
|
|
|
class ThreadState(AgentState):
|
|
sandbox: NotRequired[SandboxState | None]
|
|
thread_data: NotRequired[ThreadDataState | None]
|
|
title: NotRequired[str | None]
|
|
artifacts: Annotated[list[str], merge_artifacts]
|
|
todos: Annotated[list | None, merge_todos]
|
|
uploaded_files: NotRequired[list[dict] | None]
|
|
viewed_images: Annotated[dict[str, ViewedImageData], merge_viewed_images] # image_path -> {base64, mime_type}
|