mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-30 07:00:37 +00:00
fix(runtime): normalize non-finite floats in canonical serializer (#7670)
serialize_lc_object() passed floats through unchanged, so a non-finite
value (NaN/inf/-inf) survived serialization and broke the two display
paths that share it with opposite JSON policies: the SSE wire
(format_sse / Redis bridge use the lenient default and emit bare
NaN/Infinity tokens the langgraph-sdk/useStream decoder rejects,
breaking the live stream) and the REST surface (Starlette JSONResponse
uses allow_nan=False and raises, turning a thread read into HTTP 500).
Non-finite floats reach the serializer from ordinary runtime data:
json.loads("1e400") yields inf for a tool parsing external JSON, a
provider logprob of -inf for a zero-probability token, or a divide-by-
zero ratio. They round-trip inside Python and checkpoint persistence and
only fail at the JSON boundary.
Normalize non-finite floats to None and recurse into model_dump()/dict()
output so nested values (e.g. a message's response_metadata) are
normalized too. This is a display/wire normalization only; persisted
checkpoint state is never routed through this serializer. Finite floats
are unchanged.
Signed-off-by: CorgiBoyG <111257566+CorgiBoyG@users.noreply.github.com>
This commit is contained in:
parent
9e7448cabd
commit
9e6ce2f98e
@ -10,29 +10,47 @@ Consumers: ``deerflow.runtime.runs.worker`` (SSE publishing) and
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from typing import Any
|
||||
|
||||
|
||||
def serialize_lc_object(obj: Any) -> Any:
|
||||
"""Recursively serialize a LangChain object to a JSON-serialisable dict."""
|
||||
"""Recursively serialize a LangChain object to a JSON-serialisable dict.
|
||||
|
||||
Non-finite floats (``NaN`` / ``inf`` / ``-inf``) are normalized to ``None``.
|
||||
They are not valid JSON (RFC 8259) and reach this serializer from ordinary
|
||||
runtime data — e.g. ``json.loads("1e400")`` yields ``inf`` for a tool parsing
|
||||
external JSON, or a provider logprob of ``-inf`` for a zero-probability token.
|
||||
Left untouched they break both display paths that share this serializer: the
|
||||
SSE wire (``format_sse`` / the Redis bridge encode with the lenient default,
|
||||
emitting bare ``NaN`` / ``Infinity`` tokens the ``langgraph-sdk`` / useStream
|
||||
decoder rejects) and the REST surface (Starlette's ``JSONResponse`` uses
|
||||
``allow_nan=False`` and raises, turning a thread read into HTTP 500). This is
|
||||
a display/wire normalization only; persisted checkpoint state is never routed
|
||||
through here.
|
||||
"""
|
||||
if obj is None:
|
||||
return None
|
||||
if isinstance(obj, (str, int, float, bool)):
|
||||
# ``bool`` is a subclass of ``int``, so it passes through unchanged here.
|
||||
if isinstance(obj, (str, int)):
|
||||
return obj
|
||||
if isinstance(obj, float):
|
||||
return obj if math.isfinite(obj) else None
|
||||
if isinstance(obj, dict):
|
||||
return {k: serialize_lc_object(v) for k, v in obj.items()}
|
||||
if isinstance(obj, (list, tuple)):
|
||||
return [serialize_lc_object(item) for item in obj]
|
||||
# Pydantic v2
|
||||
# Pydantic v2 — recurse so non-finite floats nested in the dump (e.g. a
|
||||
# message's ``response_metadata``) are normalized too, not returned raw.
|
||||
if hasattr(obj, "model_dump"):
|
||||
try:
|
||||
return obj.model_dump()
|
||||
return serialize_lc_object(obj.model_dump())
|
||||
except Exception:
|
||||
pass
|
||||
# Pydantic v1 / older objects
|
||||
if hasattr(obj, "dict"):
|
||||
try:
|
||||
return obj.dict()
|
||||
return serialize_lc_object(obj.dict())
|
||||
except Exception:
|
||||
pass
|
||||
# Interrupt is a __slots__ class — no model_dump/dict/__dict__, so it
|
||||
@ -124,10 +142,22 @@ def serialize_channel_values_for_api(channel_values: dict[str, Any]) -> dict[str
|
||||
|
||||
|
||||
def serialize_messages_tuple(obj: Any) -> Any:
|
||||
"""Serialize a messages-mode tuple ``(chunk, metadata)``."""
|
||||
"""Serialize a messages-mode tuple ``(chunk, metadata)``.
|
||||
|
||||
The metadata dict rides the same wire as the chunk, so it goes through the
|
||||
same canonical serializer: a non-finite float in metadata (LangGraph merges
|
||||
caller-supplied run metadata, which can carry one) would otherwise reach
|
||||
``format_sse`` / the Redis bridge encode and emit a bare ``NaN`` /
|
||||
``Infinity`` token the SSE decoder rejects. Real LangGraph metadata also
|
||||
contains tuples (``langgraph_triggers`` / ``langgraph_path``); they become
|
||||
lists here, their JSON wire shape anyway.
|
||||
"""
|
||||
if isinstance(obj, tuple) and len(obj) == 2:
|
||||
chunk, metadata = obj
|
||||
return [serialize_lc_object(chunk), metadata if isinstance(metadata, dict) else {}]
|
||||
return [
|
||||
serialize_lc_object(chunk),
|
||||
serialize_lc_object(metadata) if isinstance(metadata, dict) else {},
|
||||
]
|
||||
return serialize_lc_object(obj)
|
||||
|
||||
|
||||
|
||||
@ -162,6 +162,149 @@ def test_serialize_dispatcher_default_mode():
|
||||
assert result == {"key": "v1"}
|
||||
|
||||
|
||||
# ── non-finite float normalization ───────────────────────────────────────────
|
||||
|
||||
|
||||
class _FakePydanticWithNonFinite:
|
||||
"""Object whose model_dump output carries a nested non-finite float.
|
||||
|
||||
Mirrors a live LangChain message whose ``response_metadata`` holds a
|
||||
provider logprob of ``-inf`` for a zero-probability token.
|
||||
"""
|
||||
|
||||
def model_dump(self):
|
||||
return {"response_metadata": {"logprobs": {"token": float("-inf")}}}
|
||||
|
||||
|
||||
def test_serialize_finite_floats_unchanged():
|
||||
from deerflow.runtime.serialization import serialize_lc_object
|
||||
|
||||
assert serialize_lc_object(3.14) == 3.14
|
||||
assert serialize_lc_object([1.5, {"x": -2.0}]) == [1.5, {"x": -2.0}]
|
||||
|
||||
|
||||
def test_serialize_non_finite_floats_normalized_to_none():
|
||||
import math
|
||||
|
||||
from deerflow.runtime.serialization import serialize_lc_object
|
||||
|
||||
assert serialize_lc_object(float("nan")) is None
|
||||
assert serialize_lc_object(float("inf")) is None
|
||||
assert serialize_lc_object(float("-inf")) is None
|
||||
result = serialize_lc_object({"a": float("inf"), "b": [float("nan"), 1.0]})
|
||||
assert result == {"a": None, "b": [None, 1.0]}
|
||||
assert all(not isinstance(v, float) or math.isfinite(v) for v in _iter_floats(result))
|
||||
|
||||
|
||||
def test_serialize_recurses_into_model_dump_output():
|
||||
"""A non-finite float inside a model_dump result must also be normalized."""
|
||||
from deerflow.runtime.serialization import serialize_lc_object
|
||||
|
||||
result = serialize_lc_object(_FakePydanticWithNonFinite())
|
||||
assert result == {"response_metadata": {"logprobs": {"token": None}}}
|
||||
|
||||
|
||||
def test_serialize_channel_values_json_safe_for_both_wires():
|
||||
"""Serialized channel values must survive a strict (allow_nan=False) encoder."""
|
||||
import json
|
||||
|
||||
from deerflow.runtime.serialization import serialize_channel_values_for_api
|
||||
|
||||
raw = {
|
||||
"messages": [{"type": "ai", "content": "hi", "response_metadata": {"logprobs": {"t": float("-inf")}}}],
|
||||
"some_metric": float("nan"),
|
||||
}
|
||||
result = serialize_channel_values_for_api(raw)
|
||||
# Strict encoding mirrors Starlette's JSONResponse.render (REST path).
|
||||
json.dumps(result, allow_nan=False)
|
||||
# Strict decoding mirrors the JS useStream / langgraph-sdk SSE parser.
|
||||
encoded = json.dumps(result)
|
||||
json.loads(encoded, parse_constant=_reject_constant)
|
||||
|
||||
|
||||
def test_serialize_messages_mode_json_safe():
|
||||
import json
|
||||
|
||||
from deerflow.runtime.serialization import serialize
|
||||
|
||||
# Both halves of the tuple must be JSON-safe: the chunk carries the
|
||||
# non-finite in model_dump output, the metadata in caller-supplied run
|
||||
# metadata (merged into messages-mode frames by LangGraph).
|
||||
payload = serialize(
|
||||
(_FakePydanticWithNonFinite(), {"node": "model", "metric": float("-inf")}),
|
||||
mode="messages",
|
||||
)
|
||||
assert payload[1]["metric"] is None
|
||||
json.dumps(payload, allow_nan=False)
|
||||
json.loads(json.dumps(payload), parse_constant=_reject_constant)
|
||||
|
||||
|
||||
def test_serialize_messages_tuple_metadata_non_finite_floats_normalized():
|
||||
"""The metadata half of a messages-mode tuple rides the same wire as the
|
||||
chunk: non-finite floats nested there must be normalized too (#7669), not
|
||||
emitted as bare ``NaN`` / ``Infinity`` tokens."""
|
||||
import json
|
||||
|
||||
from deerflow.runtime.serialization import serialize_messages_tuple
|
||||
|
||||
metadata = {
|
||||
"langgraph_node": "agent",
|
||||
"langgraph_step": 1,
|
||||
"stats": {"score": float("inf"), "delta": float("nan")},
|
||||
}
|
||||
result = serialize_messages_tuple(({"content": "hi"}, metadata))
|
||||
assert result[1] == {
|
||||
"langgraph_node": "agent",
|
||||
"langgraph_step": 1,
|
||||
"stats": {"score": None, "delta": None},
|
||||
}
|
||||
# Strict encoding mirrors Starlette's JSONResponse; strict decoding
|
||||
# mirrors the JS useStream / langgraph-sdk SSE parser.
|
||||
json.dumps(result, allow_nan=False)
|
||||
json.loads(json.dumps(result), parse_constant=_reject_constant)
|
||||
|
||||
|
||||
def test_serialize_messages_mode_json_safe_metadata():
|
||||
"""At the dispatcher level metadata gets the same canonical treatment as
|
||||
the chunk: tuples become lists (their JSON shape anyway) and nested
|
||||
objects are recursively serialized."""
|
||||
import json
|
||||
|
||||
from deerflow.runtime.serialization import serialize
|
||||
|
||||
payload = serialize(
|
||||
(
|
||||
{"content": "hi"},
|
||||
{
|
||||
"langgraph_node": "agent",
|
||||
"langgraph_triggers": ("branch:to:agent",),
|
||||
"nested": {"obj": _FakePydanticV2(), "floor": float("-inf")},
|
||||
},
|
||||
),
|
||||
mode="messages",
|
||||
)
|
||||
assert payload[1]["langgraph_node"] == "agent"
|
||||
assert payload[1]["langgraph_triggers"] == ["branch:to:agent"]
|
||||
assert payload[1]["nested"] == {"obj": {"key": "v2"}, "floor": None}
|
||||
json.dumps(payload, allow_nan=False)
|
||||
json.loads(json.dumps(payload), parse_constant=_reject_constant)
|
||||
|
||||
|
||||
def _reject_constant(token):
|
||||
raise ValueError(f"non-finite JSON token: {token}")
|
||||
|
||||
|
||||
def _iter_floats(obj):
|
||||
if isinstance(obj, float):
|
||||
yield obj
|
||||
elif isinstance(obj, dict):
|
||||
for v in obj.values():
|
||||
yield from _iter_floats(v)
|
||||
elif isinstance(obj, (list, tuple)):
|
||||
for v in obj:
|
||||
yield from _iter_floats(v)
|
||||
|
||||
|
||||
# ── strip_data_url_image_blocks ──────────────────────────────────────────────
|
||||
|
||||
|
||||
|
||||
@ -28,3 +28,44 @@ def test_sse_error_format():
|
||||
parsed = json.loads(frame.split("data: ")[1].split("\n")[0])
|
||||
assert parsed["message"] == "boom"
|
||||
assert parsed["name"] == "ValueError"
|
||||
|
||||
|
||||
def test_sse_frame_rejects_non_finite_floats():
|
||||
"""A non-finite float must not reach the wire as a bare NaN/Infinity token.
|
||||
|
||||
The canonical serializer normalizes non-finite floats before they reach
|
||||
``format_sse``; this guards the wire boundary directly so a regression in
|
||||
either layer is caught. A strict ``parse_constant`` mirrors the JS
|
||||
``useStream`` / ``langgraph-sdk`` SSE decoder, which rejects such tokens.
|
||||
"""
|
||||
from deerflow.runtime.serialization import serialize
|
||||
|
||||
payload = serialize({"messages": [], "metric": float("inf")}, mode="values")
|
||||
frame = _format_sse("values", payload)
|
||||
data_line = frame.split("data: ")[1].split("\n")[0]
|
||||
|
||||
def _reject_constant(token):
|
||||
raise ValueError(f"non-finite JSON token: {token}")
|
||||
|
||||
json.loads(data_line, parse_constant=_reject_constant)
|
||||
|
||||
|
||||
def test_sse_messages_frame_rejects_non_finite_floats_in_metadata():
|
||||
"""A non-finite float inside messages-mode metadata must not reach the
|
||||
wire as a bare NaN/Infinity token either (#7669)."""
|
||||
from deerflow.runtime.serialization import serialize
|
||||
|
||||
payload = serialize(
|
||||
(
|
||||
{"type": "AIMessageChunk", "content": "hi"},
|
||||
{"langgraph_node": "model", "score": float("inf")},
|
||||
),
|
||||
mode="messages",
|
||||
)
|
||||
frame = _format_sse("messages", payload)
|
||||
data_line = frame.split("data: ")[1].split("\n")[0]
|
||||
|
||||
def _reject_constant(token):
|
||||
raise ValueError(f"non-finite JSON token: {token}")
|
||||
|
||||
json.loads(data_line, parse_constant=_reject_constant)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user