mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-23 16:10:27 +08:00
feat(schema): project new content_blocks back to the v1 wire shape
The in-memory Message and the v2 (AG-UI) path use the new content_blocks union (groups tagged "group", every node carries id/contents, the agent answer is a trailing top-level TextContent). The v1 API keeps emitting the pre-1.11.0 shape via a pure legacy_render projection that runs only at the v1 boundaries: the v1 read/response models, the memories endpoint, the v1 build SSE stream, the webhook events SSE stream, and the /run response and stream. The build, webhook, and /run projections recurse so the Data mirror (data.data.content_blocks) is projected alongside the top-level copy. v2 serializes the live Message and keeps the new shape.
This commit is contained in:
ogabrielluiz
@ -9,6 +9,7 @@ from fastapi import BackgroundTasks, HTTPException, Response
|
||||
from lfx.graph.graph.base import Graph
|
||||
from lfx.graph.utils import log_vertex_build
|
||||
from lfx.log.logger import logger
|
||||
from lfx.schema.legacy_render import project_payload_to_v1
|
||||
from lfx.schema.schema import InputValueRequest
|
||||
from sqlmodel import select
|
||||
|
||||
@ -55,6 +56,34 @@ from langflow.services.telemetry.schema import (
|
||||
STREAMING_ACTIVITY_REFRESH_S = 10.0
|
||||
|
||||
|
||||
def _project_event_to_v1(raw: str) -> str:
|
||||
"""Render an ``add_message`` event's content_blocks/text to the v1 shape.
|
||||
|
||||
The v1 build stream ships ``message.model_dump()`` raw, which carries the new
|
||||
content_blocks union. Re-project it to the release-1.11.0 shape before it
|
||||
reaches a v1 client, so the v1 SSE stays byte-compatible. The v2 (AG-UI) path
|
||||
drains a different queue and never passes through here. The substring guard
|
||||
skips the JSON round-trip for every non-add_message event (tokens, vertex
|
||||
builds, end, ...).
|
||||
"""
|
||||
if "add_message" not in raw:
|
||||
return raw
|
||||
body = raw.rstrip("\n")
|
||||
try:
|
||||
event = json.loads(body)
|
||||
except (ValueError, TypeError):
|
||||
return raw
|
||||
data = event.get("data")
|
||||
if event.get("event") != "add_message" or not isinstance(data, dict) or "content_blocks" not in data:
|
||||
return raw
|
||||
# ``message.model_dump()`` carries content_blocks at BOTH ``data.content_blocks``
|
||||
# and ``data.data.content_blocks`` (Message subclasses Data, so the dump mirrors
|
||||
# it). Recurse over the whole payload so every copy is projected to the v1 shape
|
||||
# and the co-located non-string text is coerced, matching what /run already does.
|
||||
event["data"] = project_payload_to_v1(data)
|
||||
return json.dumps(event) + "\n\n"
|
||||
|
||||
|
||||
def _output_meta_for_vertex(graph: Graph, vertex_id: str) -> dict:
|
||||
"""Authoritative per-output metadata for the v2 ``output`` stream event.
|
||||
|
||||
@ -219,7 +248,7 @@ async def get_flow_events_response(
|
||||
# Include the end event
|
||||
events.append(None)
|
||||
break
|
||||
events.append(value.decode("utf-8"))
|
||||
events.append(_project_event_to_v1(value.decode("utf-8")))
|
||||
|
||||
# If no events were available, wait for one (with timeout)
|
||||
if not events:
|
||||
@ -230,7 +259,7 @@ async def get_flow_events_response(
|
||||
event_task.cancel()
|
||||
event_manager.on_end(data={})
|
||||
else:
|
||||
events.append(value.decode("utf-8"))
|
||||
events.append(_project_event_to_v1(value.decode("utf-8")))
|
||||
|
||||
# Return as NDJSON format - each line is a complete JSON object
|
||||
content = "\n".join([event for event in events if event is not None])
|
||||
@ -310,7 +339,7 @@ async def create_flow_response(
|
||||
if value is None:
|
||||
break
|
||||
get_time = time.time()
|
||||
yield value.decode("utf-8")
|
||||
yield _project_event_to_v1(value.decode("utf-8"))
|
||||
await logger.adebug(f"Event {event_id} consumed in {get_time - put_time:.4f}s")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
await logger.aexception(f"Error consuming event: {exc}")
|
||||
|
||||
@ -12,7 +12,7 @@ import orjson
|
||||
import sqlalchemy as sa
|
||||
from fastapi import APIRouter, BackgroundTasks, Body, Depends, HTTPException, Request, UploadFile, status
|
||||
from fastapi.encoders import jsonable_encoder
|
||||
from fastapi.responses import StreamingResponse
|
||||
from fastapi.responses import JSONResponse, StreamingResponse
|
||||
from lfx.custom.custom_component.component import Component
|
||||
from lfx.custom.utils import (
|
||||
add_code_field_to_build_config,
|
||||
@ -24,6 +24,7 @@ from lfx.graph.graph.base import Graph
|
||||
from lfx.graph.schema import RunOutputs
|
||||
from lfx.interface.components import component_cache
|
||||
from lfx.log.logger import logger
|
||||
from lfx.schema.legacy_render import project_payload_to_v1
|
||||
from lfx.schema.schema import InputValueRequest
|
||||
from lfx.services.settings.service import SettingsService
|
||||
from lfx.utils.flow_validation import (
|
||||
@ -338,7 +339,7 @@ async def simple_run_flow(
|
||||
except (RuntimeError, ValueError, OSError):
|
||||
await logger.awarning("Memory base hook scheduling failed for flow %s", flow.id, exc_info=True)
|
||||
|
||||
return RunResponse(outputs=task_result, session_id=session_id)
|
||||
return _v1_run_response(RunResponse(outputs=task_result, session_id=session_id))
|
||||
|
||||
except sa.exc.StatementError as exc:
|
||||
raise ValueError(str(exc)) from exc
|
||||
@ -440,6 +441,36 @@ async def simple_run_flow_task(
|
||||
return None
|
||||
|
||||
|
||||
def _v1_run_response(response: RunResponse) -> JSONResponse:
|
||||
"""Serialize a RunResponse with content_blocks projected to the v1 shape.
|
||||
|
||||
The /run result holds Messages whose content_blocks serialize through the
|
||||
shared (v2) Message serializer, so project them at this v1 boundary to keep
|
||||
the release-1.11.0 wire shape. The live objects and the v2 path are untouched.
|
||||
"""
|
||||
return JSONResponse(content=project_payload_to_v1(jsonable_encoder(response)))
|
||||
|
||||
|
||||
def _project_run_event(value):
|
||||
"""Project a /run stream event's content_blocks to the v1 shape.
|
||||
|
||||
Covers add_message events and the final ``end`` result (which nests messages)
|
||||
before they reach a v1 client. The v2 path drains a different queue and never
|
||||
passes through here. The substring guard skips events that carry no
|
||||
content_blocks (tokens, ...).
|
||||
"""
|
||||
if not isinstance(value, (bytes, bytearray)):
|
||||
return value
|
||||
raw = value.decode("utf-8")
|
||||
if "content_blocks" not in raw:
|
||||
return value
|
||||
try:
|
||||
event = json.loads(raw.rstrip("\n"))
|
||||
except (ValueError, TypeError):
|
||||
return value
|
||||
return (json.dumps(project_payload_to_v1(event)) + "\n\n").encode("utf-8")
|
||||
|
||||
|
||||
async def consume_and_yield(queue: asyncio.Queue, client_consumed_queue: asyncio.Queue) -> AsyncGenerator:
|
||||
"""Consumes events from a queue and yields them to the client while tracking timing metrics.
|
||||
|
||||
@ -465,7 +496,7 @@ async def consume_and_yield(queue: asyncio.Queue, client_consumed_queue: asyncio
|
||||
if value is None:
|
||||
break
|
||||
get_time = time.time()
|
||||
yield value
|
||||
yield _project_run_event(value)
|
||||
get_time_yield = time.time()
|
||||
client_consumed_queue.put_nowait(event_id)
|
||||
await logger.adebug(
|
||||
@ -932,7 +963,14 @@ async def webhook_events_stream(
|
||||
try:
|
||||
event = await asyncio.wait_for(queue.get(), timeout=SSE_HEARTBEAT_TIMEOUT_SECONDS)
|
||||
event_type = event["event"]
|
||||
event_data = json.dumps(event["data"])
|
||||
payload = event["data"]
|
||||
# add_message carries message.model_dump(), which holds the new
|
||||
# content_blocks union at both the top level and the nested Data
|
||||
# mirror. Project both to the v1 shape for this v1 SSE, matching
|
||||
# the build stream and /run.
|
||||
if event_type == "add_message":
|
||||
payload = project_payload_to_v1(payload)
|
||||
event_data = json.dumps(payload)
|
||||
yield f"event: {event_type}\ndata: {event_data}\n\n"
|
||||
except asyncio.TimeoutError:
|
||||
yield f"event: heartbeat\ndata: {json.dumps({'timestamp': time.time()})}\n\n"
|
||||
@ -1185,7 +1223,7 @@ async def experimental_run_flow(
|
||||
except (RuntimeError, ValueError, OSError):
|
||||
await logger.awarning("Memory base hook scheduling failed for flow %s", flow.id, exc_info=True)
|
||||
|
||||
return RunResponse(outputs=task_result, session_id=session_id)
|
||||
return _v1_run_response(RunResponse(outputs=task_result, session_id=session_id))
|
||||
|
||||
|
||||
@router.post(
|
||||
|
||||
@ -28,6 +28,7 @@ from typing import Annotated, Any
|
||||
from fastapi import APIRouter, Body, Depends, HTTPException
|
||||
from fastapi_pagination import Page, Params
|
||||
from fastapi_pagination.ext.sqlmodel import apaginate
|
||||
from lfx.schema.legacy_render import render_v1_content_blocks
|
||||
from pydantic import BaseModel
|
||||
from sqlmodel import select
|
||||
|
||||
@ -255,7 +256,7 @@ async def list_memory_base_messages(
|
||||
sender_name=msg.sender_name,
|
||||
session_id=msg.session_id,
|
||||
text=msg.text,
|
||||
content_blocks=msg.content_blocks or [],
|
||||
content_blocks=render_v1_content_blocks(msg.content_blocks) or [],
|
||||
job_id=mir.job_id,
|
||||
ingested_at=mir.ingested_at,
|
||||
)
|
||||
|
||||
@ -7,6 +7,7 @@ from uuid import UUID, uuid4
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
from fastapi.encoders import jsonable_encoder
|
||||
from lfx.schema.legacy_render import render_v1_content_blocks
|
||||
from pydantic import ConfigDict, field_serializer, field_validator
|
||||
from sqlalchemy import Index, Text, text
|
||||
from sqlmodel import JSON, Column, Field, SQLModel
|
||||
@ -287,10 +288,38 @@ class MessageRead(MessageBase):
|
||||
flow_id: UUID | None = None
|
||||
session_metadata: dict | None = None
|
||||
run_id: UUID | None = None
|
||||
# v1 read shape: hold content_blocks as legacy dicts and render the
|
||||
# release-1.11.0 shape, so editing/reading a stored message (new-shape rows
|
||||
# or pre-1.11.0 untagged rows) keeps the v1 wire shape and never trips the
|
||||
# new union validator (the union_tag_not_found 500 on PUT /messages/{id}).
|
||||
content_blocks: list[dict] = Field(default_factory=list) # type: ignore[assignment]
|
||||
|
||||
@field_validator("content_blocks", mode="before")
|
||||
@classmethod
|
||||
def render_legacy_content_blocks(cls, value):
|
||||
return render_v1_content_blocks(value) or []
|
||||
|
||||
|
||||
def _coerce_content_block_dicts(value):
|
||||
"""Normalize content_blocks items to plain dicts (request-parse path).
|
||||
|
||||
Accepts legacy/new dicts and ContentType model instances without forcing
|
||||
them through the new discriminated union, and stores them as-sent (no v1
|
||||
legacy projection: these are input models, not the v1 response).
|
||||
"""
|
||||
if isinstance(value, list):
|
||||
return [block.model_dump() if hasattr(block, "model_dump") else block for block in value]
|
||||
return value
|
||||
|
||||
|
||||
class MessageCreate(MessageBase):
|
||||
session_metadata: dict | None = None
|
||||
content_blocks: list[dict] = Field(default_factory=list) # type: ignore[assignment]
|
||||
|
||||
@field_validator("content_blocks", mode="before")
|
||||
@classmethod
|
||||
def coerce_content_blocks(cls, value):
|
||||
return _coerce_content_block_dicts(value) or []
|
||||
|
||||
|
||||
class MessageUpdate(SQLModel):
|
||||
@ -305,4 +334,9 @@ class MessageUpdate(SQLModel):
|
||||
properties: Properties | None = None
|
||||
session_metadata: dict | None = None
|
||||
category: str | None = None
|
||||
content_blocks: list[ContentType] | None = None
|
||||
content_blocks: list[dict] | None = None
|
||||
|
||||
@field_validator("content_blocks", mode="before")
|
||||
@classmethod
|
||||
def coerce_content_blocks(cls, value):
|
||||
return _coerce_content_block_dicts(value)
|
||||
|
||||
@ -405,27 +405,30 @@ async def test_aupdate_message_with_content_blocks(created_message):
|
||||
assert updated[0].text == "Message with content blocks"
|
||||
assert len(updated[0].content_blocks) == 1
|
||||
|
||||
# Verify the content block structure
|
||||
# MessageRead renders content_blocks as the legacy v1 shape: plain dicts,
|
||||
# a single group {title, contents, allow_markdown, media_url} whose leaves
|
||||
# carry no id/contents. The setter-appended top-level text is carried by
|
||||
# ``text`` (dropped from content_blocks, since the group is not "Agent Steps").
|
||||
updated_block = updated[0].content_blocks[0]
|
||||
assert updated_block.title == "Test Block"
|
||||
assert updated_block["title"] == "Test Block"
|
||||
expected_len = 2
|
||||
assert len(updated_block.contents) == expected_len
|
||||
assert len(updated_block["contents"]) == expected_len
|
||||
|
||||
# Verify text content
|
||||
text_content = updated_block.contents[0]
|
||||
assert text_content.type == "text"
|
||||
assert text_content.text == "Test content"
|
||||
text_content = updated_block["contents"][0]
|
||||
assert text_content["type"] == "text"
|
||||
assert text_content["text"] == "Test content"
|
||||
duration = 5
|
||||
assert text_content.duration == duration
|
||||
assert text_content.header["title"] == "Test Header"
|
||||
assert text_content["duration"] == duration
|
||||
assert text_content["header"]["title"] == "Test Header"
|
||||
|
||||
# Verify tool content
|
||||
tool_content = updated_block.contents[1]
|
||||
assert tool_content.type == "tool_use"
|
||||
assert tool_content.name == "test_tool"
|
||||
assert tool_content.tool_input == {"param": "value"}
|
||||
tool_content = updated_block["contents"][1]
|
||||
assert tool_content["type"] == "tool_use"
|
||||
assert tool_content["name"] == "test_tool"
|
||||
assert tool_content["tool_input"] == {"param": "value"}
|
||||
duration = 10
|
||||
assert tool_content.duration == duration
|
||||
assert tool_content["duration"] == duration
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("client")
|
||||
@ -509,9 +512,12 @@ async def test_aupdate_message_with_dataframe_in_tool_output(created_message):
|
||||
assert updated[0].text == "Agent response after Memory Base retrieval"
|
||||
assert len(updated[0].content_blocks) == 1
|
||||
|
||||
stored_tool = updated[0].content_blocks[0].contents[0]
|
||||
assert stored_tool.type == "tool_use"
|
||||
assert stored_tool.name == "MemoryBase"
|
||||
# MessageRead renders the legacy v1 shape (plain dicts). For an "Agent Steps"
|
||||
# group the answer is folded back in as a trailing Output leaf, so the tool
|
||||
# stays at contents[0].
|
||||
stored_tool = updated[0].content_blocks[0]["contents"][0]
|
||||
assert stored_tool["type"] == "tool_use"
|
||||
assert stored_tool["name"] == "MemoryBase"
|
||||
|
||||
# The output must be free of pandas / numpy types and fully JSON-encodable.
|
||||
def _assert_json_native(value):
|
||||
@ -525,9 +531,9 @@ async def test_aupdate_message_with_dataframe_in_tool_output(created_message):
|
||||
assert not isinstance(value, np.generic), f"numpy scalar leaked: {value!r}"
|
||||
assert not isinstance(value, pd.DataFrame), "DataFrame leaked into persisted output"
|
||||
|
||||
_assert_json_native(stored_tool.output)
|
||||
_assert_json_native(stored_tool["output"])
|
||||
# Strict JSON dump must succeed without a fallback encoder.
|
||||
json.dumps(stored_tool.output, allow_nan=False)
|
||||
json.dumps(stored_tool["output"], allow_nan=False)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
|
||||
138
src/lfx/src/lfx/schema/legacy_render.py
Normal file
138
src/lfx/src/lfx/schema/legacy_render.py
Normal file
@ -0,0 +1,138 @@
|
||||
"""Project the new content_blocks model back onto the release-1.11.0 (v1) wire shape.
|
||||
|
||||
The in-memory ``Message`` is the source of truth and carries the new content_blocks
|
||||
union: groups are tagged ``type="group"``, every node carries ``id``/``contents``,
|
||||
and an agent's final answer is a trailing top-level ``TextContent``. The v1 API must
|
||||
keep emitting the exact pre-1.11.0 shape, so these pure helpers run only at the v1
|
||||
boundary (the v1 read models, ``MessageResponse.from_message``, the v1 event stream,
|
||||
and the v1 openai-responses consumer). The v2 (AG-UI / workflows) path serializes the
|
||||
live ``Message`` directly and keeps the new shape, so it must never call these.
|
||||
|
||||
Baseline v1 invariants reproduced here:
|
||||
- ``content_blocks`` holds groups only (baseline never stored a top-level text leaf).
|
||||
- each group is ``{title, contents, allow_markdown, media_url}`` in that order.
|
||||
- leaves carry no ``id`` / ``contents`` keys.
|
||||
- an agent's answer lives inside the "Agent Steps" group as a trailing "Output" leaf.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
_OUTPUT_HEADER = {"title": "Output", "icon": "MessageSquare"}
|
||||
_AGENT_STEPS_TITLE = "Agent Steps"
|
||||
|
||||
|
||||
def legacy_text(message) -> str:
|
||||
"""The v1 ``text`` value: always a string. A pending stream/iterator renders as ""."""
|
||||
text = message.text
|
||||
return text if isinstance(text, str) else ""
|
||||
|
||||
|
||||
def _is_group(block) -> bool:
|
||||
if not isinstance(block, dict):
|
||||
return False
|
||||
if block.get("type") == "group":
|
||||
return True
|
||||
# A legacy group dict read back from the DB carries no discriminator.
|
||||
return "title" in block and "contents" in block and "type" not in block
|
||||
|
||||
|
||||
def _legacy_leaf(node: dict) -> dict:
|
||||
# Drop the two additive keys the new model puts on every leaf. ``id`` sat right
|
||||
# after ``type`` and ``contents`` right after ``header`` in the new node, so plain
|
||||
# deletion leaves the remaining keys in baseline order (type, duration, header, ...).
|
||||
return {k: v for k, v in node.items() if k not in ("id", "contents")}
|
||||
|
||||
|
||||
def _legacy_group(block: dict) -> dict:
|
||||
return {
|
||||
"title": block.get("title"),
|
||||
"contents": [_legacy_leaf(c) for c in block.get("contents") or []],
|
||||
"allow_markdown": block.get("allow_markdown", True),
|
||||
"media_url": block.get("media_url"),
|
||||
}
|
||||
|
||||
|
||||
def legacy_content_blocks(blocks) -> list[dict]:
|
||||
"""Project new-model content_blocks (serialized dicts) onto the v1 shape.
|
||||
|
||||
Groups are kept and rendered legacy; top-level text leaves are dropped because
|
||||
baseline never stored one. For an agent message (a group titled "Agent Steps")
|
||||
the relocated answer is folded back into that group as the trailing "Output"
|
||||
leaf, reproducing baseline. In every other case the answer text is carried by
|
||||
the message's ``text`` / ``data["text"]`` and is not re-injected, which keeps
|
||||
generic ``.text=`` / multi-group / custom-titled-group messages untouched.
|
||||
"""
|
||||
blocks = blocks or []
|
||||
groups = [b for b in blocks if _is_group(b)]
|
||||
answer = "".join(
|
||||
b.get("text", "") for b in blocks if isinstance(b, dict) and not _is_group(b) and b.get("type") == "text"
|
||||
)
|
||||
|
||||
rendered: list[dict] = []
|
||||
folded = False
|
||||
for block in groups:
|
||||
group = _legacy_group(block)
|
||||
if not folded and answer and block.get("title") == _AGENT_STEPS_TITLE:
|
||||
# ``duration`` is None on purpose: the new model relocates the answer to a
|
||||
# bare top-level TextContent (the text setter builds ``TextContent(text=...)``
|
||||
# with no duration), so the answer's timing baseline once stamped on the
|
||||
# in-group leaf is simply not present to carry through. This is a faithful
|
||||
# projection of what the message holds, not a dropped field.
|
||||
group["contents"].append({"type": "text", "duration": None, "header": dict(_OUTPUT_HEADER), "text": answer})
|
||||
folded = True
|
||||
rendered.append(group)
|
||||
return rendered
|
||||
|
||||
|
||||
def project_payload_to_v1(obj):
|
||||
"""Recursively project every message ``content_blocks`` in a nested payload.
|
||||
|
||||
Used at v1 boundaries that serialize whole run results or stream events (the
|
||||
``/run`` endpoint and its streaming ``end`` event) where messages sit at
|
||||
arbitrary depth. Pure: returns a new structure, the live objects are
|
||||
untouched. Wherever a dict carries a ``content_blocks`` list it is rendered
|
||||
to the legacy v1 shape, and a co-located non-string ``text`` is coerced to "".
|
||||
"""
|
||||
if isinstance(obj, dict):
|
||||
out = {}
|
||||
for key, value in obj.items():
|
||||
if key == "content_blocks" and isinstance(value, list):
|
||||
out[key] = legacy_content_blocks(value)
|
||||
else:
|
||||
out[key] = project_payload_to_v1(value)
|
||||
if isinstance(out.get("content_blocks"), list) and "text" in out and not isinstance(out["text"], str):
|
||||
# Reached only when text is non-string (the guard above); v1 text is
|
||||
# always a string, so a co-located non-string projects to "".
|
||||
out["text"] = ""
|
||||
return out
|
||||
if isinstance(obj, list):
|
||||
return [project_payload_to_v1(item) for item in obj]
|
||||
return obj
|
||||
|
||||
|
||||
def render_v1_content_blocks(value):
|
||||
"""Normalize any content_blocks value to the legacy v1 shape.
|
||||
|
||||
Accepts ``ContentType`` model instances (live ``Message``), serialized dicts
|
||||
(a DB-row read), a JSON string, a list mixing those, or ``None``. ``None`` is
|
||||
returned unchanged; everything else is normalized to dicts and projected with
|
||||
:func:`legacy_content_blocks`. This is the single entry point every v1 wire
|
||||
model uses so the projection cannot drift between them.
|
||||
"""
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, str):
|
||||
value = json.loads(value)
|
||||
if not isinstance(value, list):
|
||||
return value
|
||||
dicts = []
|
||||
for block in value:
|
||||
if hasattr(block, "model_dump"):
|
||||
dicts.append(block.model_dump())
|
||||
elif isinstance(block, str):
|
||||
dicts.append(json.loads(block))
|
||||
else:
|
||||
dicts.append(block)
|
||||
return legacy_content_blocks(dicts)
|
||||
@ -35,6 +35,7 @@ from lfx.schema.content_block import ContentBlock, ContentType
|
||||
from lfx.schema.content_types import ErrorContent, TextContent
|
||||
from lfx.schema.data import Data
|
||||
from lfx.schema.image import Image, get_file_paths, is_image_file
|
||||
from lfx.schema.legacy_render import legacy_text, render_v1_content_blocks
|
||||
from lfx.schema.properties import Properties, Source
|
||||
from lfx.schema.validators import str_to_timestamp_validator, timestamp_to_str, timestamp_to_str_validator
|
||||
from lfx.utils.constants import MESSAGE_SENDER_AI, MESSAGE_SENDER_NAME_AI, MESSAGE_SENDER_NAME_USER, MESSAGE_SENDER_USER
|
||||
@ -120,11 +121,14 @@ class Message(Data):
|
||||
|
||||
@computed_field
|
||||
@property
|
||||
def text(self) -> str:
|
||||
def text(self) -> str | AsyncIterator | Iterator:
|
||||
"""Extract text from content_blocks, or fall back to data dict.
|
||||
|
||||
Always returns a string. For streaming access, use `text_stream` property.
|
||||
"""
|
||||
stream = self.__dict__.get("_text_stream")
|
||||
if stream is not None:
|
||||
return stream
|
||||
# If content_blocks has TextContent, derive from there
|
||||
text_from_blocks = "".join(b.text for b in self.content_blocks if isinstance(b, TextContent))
|
||||
if text_from_blocks:
|
||||
@ -298,6 +302,15 @@ class Message(Data):
|
||||
existing = self.data[self.text_key]
|
||||
if existing is not None and not isinstance(existing, str):
|
||||
self.data[self.text_key] = ""
|
||||
else:
|
||||
# Bare-default message (``text`` never provided): restore the
|
||||
# release-1.11.0 default of ``text == ""`` so a sender-only message
|
||||
# still persists via ``from_message`` and v1 consumers keep finding
|
||||
# ``data["text"]``. This matches baseline for both v1 and v2 (``text``
|
||||
# was a stored field defaulting to ""); only the ``text`` value is
|
||||
# seeded, ``content_blocks`` is untouched. An explicit ``text=None``
|
||||
# keeps the key present with ``None`` (handled by the elif above).
|
||||
self.data[self.text_key] = ""
|
||||
|
||||
def set_flow_id(self, flow_id: str) -> None:
|
||||
self.flow_id = flow_id
|
||||
@ -740,29 +753,21 @@ class MessageResponse(DefaultModel):
|
||||
|
||||
properties: Properties | None = None
|
||||
category: str | None = None
|
||||
content_blocks: list[ContentType] | None = None
|
||||
# v1 wire shape: content_blocks is held as plain legacy dicts, not the new
|
||||
# ContentType union, so the v1 API keeps emitting the release-1.11.0 shape.
|
||||
# The new (v2) shape never flows through MessageResponse.
|
||||
content_blocks: list[dict] | None = None
|
||||
session_metadata: dict | None = None
|
||||
|
||||
@field_validator("content_blocks", mode="before")
|
||||
@classmethod
|
||||
def validate_content_blocks(cls, v):
|
||||
if isinstance(v, str):
|
||||
v = json.loads(v)
|
||||
if isinstance(v, list):
|
||||
return [cls.validate_content_blocks(block) for block in v]
|
||||
if isinstance(v, dict):
|
||||
# Route every tagged dict through the discriminated-union adapter.
|
||||
# Stored blocks always serialize ``contents: []`` (default_factory),
|
||||
# so an "absence of contents" guard would misfire and force flat
|
||||
# ContentTypes into ContentBlock.model_validate (which requires a
|
||||
# title) on the DB read path. The grouped ContentBlock carries
|
||||
# type="group" and is itself a member of the union.
|
||||
if "type" in v:
|
||||
return _CONTENT_TYPE_ADAPTER.validate_python(v)
|
||||
if "title" in v and "contents" in v:
|
||||
return ContentBlock.model_validate(v)
|
||||
return _CONTENT_TYPE_ADAPTER.validate_python(v)
|
||||
return v
|
||||
# Project the new content_blocks model onto the legacy (release-1.11.0)
|
||||
# v1 shape. Accepts ContentType models (from ``from_message``), serialized
|
||||
# dicts (from a DB-row ``model_validate``), or a JSON string. The agent
|
||||
# answer is folded back into the "Agent Steps" group; the new (v2) shape
|
||||
# never flows through MessageResponse.
|
||||
return render_v1_content_blocks(v)
|
||||
|
||||
@field_validator("properties", mode="before")
|
||||
@classmethod
|
||||
@ -803,9 +808,7 @@ class MessageResponse(DefaultModel):
|
||||
if no_content or not message.sender or not message.sender_name:
|
||||
msg = "The message does not have the required fields (text, sender, sender_name)."
|
||||
raise ValueError(msg)
|
||||
text = message.text
|
||||
if not isinstance(text, str):
|
||||
text = ""
|
||||
text = legacy_text(message)
|
||||
return cls(
|
||||
sender=message.sender,
|
||||
sender_name=message.sender_name,
|
||||
|
||||
Reference in New Issue
Block a user