mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
refactor: share one message->text helper for journal + thread messages (#3747)
Part of #3742. RunJournal._message_text and the gateway thread-messages helper (thread_runs._message_text) reimplemented the same 'extract display text from a message' logic — str / list of string|{text}|nested{content} blocks joined without a separator / mapping with text|content key. They differed only in two ways: journal reads a BaseMessage attribute while thread_runs reads dict-shaped run_events rows, and journal falls back to message.text. Add deerflow.utils.messages.message_to_text(message, *, text_attribute_fallback=False) that handles both message shapes (attribute or mapping content access) and gates the .text fallback behind a flag, and have both call sites delegate. journal passes text_attribute_fallback=True; thread_runs uses the default. Behavior is unchanged at both sites. Verified behavior-preserving with an equivalence harness running both original implementations vs the shared helper over 98 inputs (BaseMessage and dict messages; str/list/mapping/None/numeric content; mixed blocks; .text attribute present/absent/non-str) -> 0 mismatches. Added tests/test_utils_messages.py; the journal last_ai_message extraction tests still pass.
This commit is contained in:
parent
67dd75db62
commit
cc1df2d038
@ -25,7 +25,7 @@ from app.gateway.deps import get_checkpointer, get_current_user, get_feedback_re
|
|||||||
from app.gateway.pagination import trim_run_message_page
|
from app.gateway.pagination import trim_run_message_page
|
||||||
from app.gateway.services import sse_consumer, start_run, wait_for_run_completion
|
from app.gateway.services import sse_consumer, start_run, wait_for_run_completion
|
||||||
from deerflow.runtime import RunRecord, RunStatus, serialize_channel_values_for_api
|
from deerflow.runtime import RunRecord, RunStatus, serialize_channel_values_for_api
|
||||||
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, get_original_user_content_text
|
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, get_original_user_content_text, message_to_text
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
router = APIRouter(prefix="/api/threads", tags=["runs"])
|
router = APIRouter(prefix="/api/threads", tags=["runs"])
|
||||||
@ -196,29 +196,7 @@ def _message_content(message: Any) -> Any:
|
|||||||
|
|
||||||
|
|
||||||
def _message_text(message: Any) -> str:
|
def _message_text(message: Any) -> str:
|
||||||
content = _message_content(message)
|
return message_to_text(message)
|
||||||
if isinstance(content, str):
|
|
||||||
return content
|
|
||||||
if isinstance(content, list):
|
|
||||||
parts: list[str] = []
|
|
||||||
for block in content:
|
|
||||||
if isinstance(block, str):
|
|
||||||
parts.append(block)
|
|
||||||
elif isinstance(block, dict):
|
|
||||||
text = block.get("text")
|
|
||||||
if isinstance(text, str):
|
|
||||||
parts.append(text)
|
|
||||||
else:
|
|
||||||
nested = block.get("content")
|
|
||||||
if isinstance(nested, str):
|
|
||||||
parts.append(nested)
|
|
||||||
return "".join(parts)
|
|
||||||
if isinstance(content, dict):
|
|
||||||
for key in ("text", "content"):
|
|
||||||
value = content.get(key)
|
|
||||||
if isinstance(value, str):
|
|
||||||
return value
|
|
||||||
return ""
|
|
||||||
|
|
||||||
|
|
||||||
def _message_additional_kwargs(message: Any) -> dict[str, Any]:
|
def _message_additional_kwargs(message: Any) -> dict[str, Any]:
|
||||||
|
|||||||
@ -29,6 +29,8 @@ from langchain_core.callbacks import BaseCallbackHandler
|
|||||||
from langchain_core.messages import AIMessage, AnyMessage, BaseMessage, HumanMessage, ToolMessage
|
from langchain_core.messages import AIMessage, AnyMessage, BaseMessage, HumanMessage, ToolMessage
|
||||||
from langgraph.types import Command
|
from langgraph.types import Command
|
||||||
|
|
||||||
|
from deerflow.utils.messages import message_to_text
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from deerflow.runtime.events.store.base import RunEventStore
|
from deerflow.runtime.events.store.base import RunEventStore
|
||||||
|
|
||||||
@ -104,33 +106,7 @@ class RunJournal(BaseCallbackHandler):
|
|||||||
@staticmethod
|
@staticmethod
|
||||||
def _message_text(message: BaseMessage) -> str:
|
def _message_text(message: BaseMessage) -> str:
|
||||||
"""Extract displayable text from a message's mixed content shape."""
|
"""Extract displayable text from a message's mixed content shape."""
|
||||||
content = getattr(message, "content", None)
|
return message_to_text(message, text_attribute_fallback=True)
|
||||||
if isinstance(content, str):
|
|
||||||
return content
|
|
||||||
if isinstance(content, list):
|
|
||||||
parts: list[str] = []
|
|
||||||
for block in content:
|
|
||||||
if isinstance(block, str):
|
|
||||||
parts.append(block)
|
|
||||||
elif isinstance(block, Mapping):
|
|
||||||
text = block.get("text")
|
|
||||||
if isinstance(text, str):
|
|
||||||
parts.append(text)
|
|
||||||
else:
|
|
||||||
nested = block.get("content")
|
|
||||||
if isinstance(nested, str):
|
|
||||||
parts.append(nested)
|
|
||||||
return "".join(parts)
|
|
||||||
if isinstance(content, Mapping):
|
|
||||||
for key in ("text", "content"):
|
|
||||||
value = content.get(key)
|
|
||||||
if isinstance(value, str):
|
|
||||||
return value
|
|
||||||
|
|
||||||
text = getattr(message, "text", None)
|
|
||||||
if isinstance(text, str):
|
|
||||||
return text
|
|
||||||
return ""
|
|
||||||
|
|
||||||
def _record_message_summary(self, message: BaseMessage, *, caller: str | None = None) -> None:
|
def _record_message_summary(self, message: BaseMessage, *, caller: str | None = None) -> None:
|
||||||
"""Update run-level convenience fields for persisted run rows."""
|
"""Update run-level convenience fields for persisted run rows."""
|
||||||
|
|||||||
@ -23,6 +23,49 @@ def message_content_to_text(content: Any) -> str:
|
|||||||
return str(content)
|
return str(content)
|
||||||
|
|
||||||
|
|
||||||
|
def message_to_text(message: Any, *, text_attribute_fallback: bool = False) -> str:
|
||||||
|
"""Extract display text from a whole message (``BaseMessage`` or dict-shaped).
|
||||||
|
|
||||||
|
Reads ``content`` from either an attribute (``BaseMessage``) or a mapping key
|
||||||
|
(``run_events`` rows are dicts), then walks the mixed ``content`` shapes:
|
||||||
|
plain string; a list of string / ``{"text": ...}`` / nested ``{"content": ...}``
|
||||||
|
blocks joined without a separator; or a mapping with a ``text``/``content`` key.
|
||||||
|
Set ``text_attribute_fallback=True`` to fall back to ``message.text`` when
|
||||||
|
content yields nothing (matches ``RunJournal._message_text``).
|
||||||
|
|
||||||
|
Unlike :func:`message_content_to_text` (which takes raw ``content`` and joins
|
||||||
|
list blocks with newlines), this keeps the no-separator join and the broader
|
||||||
|
shape handling that several call sites had each reimplemented.
|
||||||
|
"""
|
||||||
|
content = message.get("content") if isinstance(message, Mapping) else getattr(message, "content", None)
|
||||||
|
if isinstance(content, str):
|
||||||
|
return content
|
||||||
|
if isinstance(content, list):
|
||||||
|
parts: list[str] = []
|
||||||
|
for block in content:
|
||||||
|
if isinstance(block, str):
|
||||||
|
parts.append(block)
|
||||||
|
elif isinstance(block, Mapping):
|
||||||
|
text = block.get("text")
|
||||||
|
if isinstance(text, str):
|
||||||
|
parts.append(text)
|
||||||
|
else:
|
||||||
|
nested = block.get("content")
|
||||||
|
if isinstance(nested, str):
|
||||||
|
parts.append(nested)
|
||||||
|
return "".join(parts)
|
||||||
|
if isinstance(content, Mapping):
|
||||||
|
for key in ("text", "content"):
|
||||||
|
value = content.get(key)
|
||||||
|
if isinstance(value, str):
|
||||||
|
return value
|
||||||
|
if text_attribute_fallback:
|
||||||
|
text = getattr(message, "text", None)
|
||||||
|
if isinstance(text, str):
|
||||||
|
return text
|
||||||
|
return ""
|
||||||
|
|
||||||
|
|
||||||
def get_original_user_content_text(content: Any, additional_kwargs: Mapping[str, Any] | None) -> str:
|
def get_original_user_content_text(content: Any, additional_kwargs: Mapping[str, Any] | None) -> str:
|
||||||
"""Return pre-middleware user text when available, otherwise content text."""
|
"""Return pre-middleware user text when available, otherwise content text."""
|
||||||
original_content = (additional_kwargs or {}).get(ORIGINAL_USER_CONTENT_KEY)
|
original_content = (additional_kwargs or {}).get(ORIGINAL_USER_CONTENT_KEY)
|
||||||
|
|||||||
72
backend/tests/test_utils_messages.py
Normal file
72
backend/tests/test_utils_messages.py
Normal file
@ -0,0 +1,72 @@
|
|||||||
|
"""Tests for deerflow.utils.messages text extraction.
|
||||||
|
|
||||||
|
``message_to_text`` is the shared extractor that ``RunJournal._message_text``
|
||||||
|
(BaseMessage, with ``.text`` fallback) and the gateway thread-messages helper
|
||||||
|
(dict-shaped run_events rows, no fallback) now delegate to — see the
|
||||||
|
"consolidate message->text helpers" tracking issue.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from types import SimpleNamespace
|
||||||
|
|
||||||
|
from deerflow.utils.messages import message_content_to_text, message_to_text
|
||||||
|
|
||||||
|
# ---------- message_to_text: content shapes ----------
|
||||||
|
|
||||||
|
|
||||||
|
def test_plain_string_content():
|
||||||
|
assert message_to_text(SimpleNamespace(content="hello")) == "hello"
|
||||||
|
assert message_to_text({"content": "hi"}) == "hi"
|
||||||
|
assert message_to_text(SimpleNamespace(content="")) == ""
|
||||||
|
|
||||||
|
|
||||||
|
def test_list_content_joins_without_separator():
|
||||||
|
content = ["a", {"text": "B"}, {"content": "C"}, {"other": 1}, 42]
|
||||||
|
expected = "aBC" # strings + dict["text"] + nested dict["content"]; non-text dropped
|
||||||
|
assert message_to_text(SimpleNamespace(content=content)) == expected
|
||||||
|
assert message_to_text({"content": content}) == expected
|
||||||
|
|
||||||
|
|
||||||
|
def test_mapping_content_text_then_content_key():
|
||||||
|
assert message_to_text(SimpleNamespace(content={"text": "T"})) == "T"
|
||||||
|
assert message_to_text(SimpleNamespace(content={"content": "N"})) == "N"
|
||||||
|
assert message_to_text(SimpleNamespace(content={"other": "x"})) == ""
|
||||||
|
|
||||||
|
|
||||||
|
def test_dict_message_without_content_key():
|
||||||
|
assert message_to_text({}) == ""
|
||||||
|
assert message_to_text({"role": "user"}) == ""
|
||||||
|
|
||||||
|
|
||||||
|
def test_non_text_content_returns_empty():
|
||||||
|
assert message_to_text(SimpleNamespace(content=None)) == ""
|
||||||
|
assert message_to_text(SimpleNamespace(content=123)) == ""
|
||||||
|
|
||||||
|
|
||||||
|
# ---------- text_attribute_fallback (journal behavior) ----------
|
||||||
|
|
||||||
|
|
||||||
|
def test_text_attribute_fallback_only_when_enabled():
|
||||||
|
# Content yields nothing, but the message has a ``.text`` attribute.
|
||||||
|
msg = SimpleNamespace(content=None, text="from-attr")
|
||||||
|
assert message_to_text(msg, text_attribute_fallback=True) == "from-attr"
|
||||||
|
assert message_to_text(msg) == "" # default: no fallback
|
||||||
|
|
||||||
|
|
||||||
|
def test_empty_string_content_is_not_overridden_by_fallback():
|
||||||
|
# Empty-string content matches the str branch and wins over the fallback.
|
||||||
|
msg = SimpleNamespace(content="", text="from-attr")
|
||||||
|
assert message_to_text(msg, text_attribute_fallback=True) == ""
|
||||||
|
|
||||||
|
|
||||||
|
def test_non_string_text_attribute_ignored():
|
||||||
|
msg = SimpleNamespace(content=None, text=lambda: "callable-not-str")
|
||||||
|
assert message_to_text(msg, text_attribute_fallback=True) == ""
|
||||||
|
|
||||||
|
|
||||||
|
# ---------- message_content_to_text unchanged (newline join, takes content) ----------
|
||||||
|
|
||||||
|
|
||||||
|
def test_message_content_to_text_still_joins_with_newline():
|
||||||
|
assert message_content_to_text(["a", {"text": "b"}]) == "a\nb"
|
||||||
Loading…
x
Reference in New Issue
Block a user