mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-24 04:13:36 +08:00
feat(agent): interleaved text + tool_use rendering and tabbed tool-output visualizer
This commit is contained in:
ogabrielluiz
4654
.secrets.baseline
4654
.secrets.baseline
File diff suppressed because it is too large
Load Diff
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@ -15,6 +15,7 @@ separate "wrapper" shape; the wrapper *is* a ContentType.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Annotated, Any, Literal
|
||||
|
||||
from fastapi.encoders import jsonable_encoder
|
||||
@ -165,6 +166,24 @@ class ToolContent(BaseContent):
|
||||
error: Any | None = None
|
||||
duration: int | None = None
|
||||
|
||||
@field_validator("tool_input", mode="before")
|
||||
@classmethod
|
||||
def _coerce_tool_input(cls, v: Any) -> dict[str, Any]:
|
||||
# LangChain ``AgentAction.tool_input`` is ``str | dict`` during
|
||||
# streaming, and ``event["data"].get("input") or {}`` passes a
|
||||
# non-empty string straight through. The dict-typed field used to
|
||||
# raise ValidationError on it. Parse JSON objects; wrap any other
|
||||
# string under an ``input`` key so the field always validates.
|
||||
if isinstance(v, str):
|
||||
try:
|
||||
parsed = json.loads(v)
|
||||
except (ValueError, TypeError):
|
||||
return {"input": v}
|
||||
return parsed if isinstance(parsed, dict) else {"input": parsed}
|
||||
if v is None:
|
||||
return {}
|
||||
return v
|
||||
|
||||
|
||||
class _MediaContentMixin:
|
||||
"""Shared validation for media content types (image, audio, video)."""
|
||||
|
||||
@ -4,7 +4,7 @@ from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
from langchain_core.agents import AgentFinish
|
||||
from langchain_core.messages import AIMessageChunk
|
||||
from langchain_core.messages import AIMessage, AIMessageChunk
|
||||
from lfx.base.agents.events import (
|
||||
_extract_output_text,
|
||||
handle_on_chain_end,
|
||||
@ -15,7 +15,6 @@ from lfx.base.agents.events import (
|
||||
handle_on_tool_start,
|
||||
process_agent_events,
|
||||
)
|
||||
from lfx.schema.content_block import ContentBlock
|
||||
from lfx.schema.content_types import ToolContent
|
||||
from lfx.schema.message import Message
|
||||
from lfx.utils.constants import MESSAGE_SENDER_AI
|
||||
@ -59,15 +58,17 @@ async def test_chain_start_event():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
session_id="test_session_id",
|
||||
)
|
||||
|
||||
result = await process_agent_events(create_event_iterator(events), agent_message, send_message)
|
||||
|
||||
assert result.properties.icon == "Bot"
|
||||
assert len(result.content_blocks) == 1
|
||||
assert result.content_blocks[0].title == "Agent Steps"
|
||||
# handle_on_chain_start is a no-op in the flat content_blocks design --
|
||||
# the user's input is already rendered above the agent reply, so we no
|
||||
# longer echo it as a synthetic "Input" TextContent.
|
||||
assert result.content_blocks == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@ -85,7 +86,7 @@ async def test_chain_end_event():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
session_id="test_session_id",
|
||||
)
|
||||
|
||||
@ -124,16 +125,15 @@ async def test_tool_start_event():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
session_id="test_session_id",
|
||||
)
|
||||
result = await process_agent_events(create_event_iterator(events), agent_message, send_message)
|
||||
|
||||
assert result.properties.icon == "Bot"
|
||||
# Flat content_blocks: the ToolContent is the only entry, at index 0.
|
||||
assert len(result.content_blocks) == 1
|
||||
assert result.content_blocks[0].title == "Agent Steps"
|
||||
assert len(result.content_blocks[0].contents) > 0
|
||||
tool_content = result.content_blocks[0].contents[-1]
|
||||
tool_content = result.content_blocks[-1]
|
||||
assert isinstance(tool_content, ToolContent)
|
||||
assert tool_content.name == "test_tool"
|
||||
assert tool_content.tool_input == {"query": "tool input"}, tool_content
|
||||
@ -164,13 +164,13 @@ async def test_tool_end_event():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
session_id="test_session_id",
|
||||
)
|
||||
result = await process_agent_events(create_event_iterator(events), agent_message, send_message)
|
||||
|
||||
assert len(result.content_blocks) == 1
|
||||
tool_content = result.content_blocks[0].contents[-1]
|
||||
tool_content = result.content_blocks[-1]
|
||||
assert tool_content.name == "test_tool"
|
||||
assert tool_content.output == "tool output"
|
||||
|
||||
@ -200,13 +200,13 @@ async def test_tool_error_event():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
session_id="test_session_id",
|
||||
)
|
||||
|
||||
result = await process_agent_events(create_event_iterator(events), agent_message, send_message)
|
||||
|
||||
tool_content = result.content_blocks[0].contents[-1]
|
||||
tool_content = result.content_blocks[-1]
|
||||
assert tool_content.name == "test_tool"
|
||||
assert tool_content.error == "error message"
|
||||
assert tool_content.header["title"] == "Error using **test_tool**"
|
||||
@ -222,7 +222,7 @@ async def test_chain_stream_event():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
session_id="test_session_id",
|
||||
)
|
||||
result = await process_agent_events(create_event_iterator(events), agent_message, send_message)
|
||||
@ -263,14 +263,22 @@ async def test_multiple_events():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
)
|
||||
|
||||
result = await process_agent_events(create_event_iterator(events), agent_message, send_message)
|
||||
|
||||
assert result.properties.state == "complete"
|
||||
assert result.properties.icon == "Bot"
|
||||
# No on_chat_model_end in the stream, so the interleaved-text path
|
||||
# doesn't fire; we get just the ToolContent that on_tool_start
|
||||
# appended (via its fallback path) and that on_tool_end filled in.
|
||||
# handle_on_chain_end stashes the final text in data["text"] but
|
||||
# doesn't append a synthetic TextContent — Message.text's getter
|
||||
# falls back to data["text"] when content_blocks has no TextContent.
|
||||
assert len(result.content_blocks) == 1
|
||||
assert isinstance(result.content_blocks[0], ToolContent)
|
||||
assert result.content_blocks[0].output == "tool output"
|
||||
assert result.text == "final output"
|
||||
|
||||
|
||||
@ -282,7 +290,7 @@ async def test_unknown_event():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])], # Initialize with empty content block
|
||||
content_blocks=[], # Initialize with empty content block
|
||||
)
|
||||
|
||||
events = [{"event": "unknown_event", "data": {"some": "data"}, "start_time": 0}]
|
||||
@ -291,9 +299,8 @@ async def test_unknown_event():
|
||||
|
||||
# Should complete without error and maintain default state
|
||||
assert result.properties.state == "complete"
|
||||
# Content blocks should be empty but present
|
||||
assert len(result.content_blocks) == 1
|
||||
assert len(result.content_blocks[0].contents) == 0
|
||||
# Unknown events should not touch content_blocks.
|
||||
assert result.content_blocks == []
|
||||
|
||||
|
||||
# Additional tests for individual handler functions
|
||||
@ -307,15 +314,15 @@ async def test_handle_on_chain_start_with_input():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
)
|
||||
event = {"event": "on_chain_start", "data": {"input": {"input": "test input", "chat_history": []}}, "start_time": 0}
|
||||
|
||||
updated_message, start_time = await handle_on_chain_start(event, agent_message, send_message, None, 0.0)
|
||||
|
||||
assert updated_message.properties.icon == "Bot"
|
||||
assert len(updated_message.content_blocks) == 1
|
||||
assert updated_message.content_blocks[0].title == "Agent Steps"
|
||||
# No-op handler: content_blocks stays untouched.
|
||||
assert updated_message.content_blocks == []
|
||||
assert isinstance(start_time, float)
|
||||
|
||||
|
||||
@ -327,15 +334,15 @@ async def test_handle_on_chain_start_no_input():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
)
|
||||
event = {"event": "on_chain_start", "data": {}, "start_time": 0}
|
||||
|
||||
updated_message, start_time = await handle_on_chain_start(event, agent_message, send_message, None, 0.0)
|
||||
|
||||
assert updated_message.properties.icon == "Bot"
|
||||
assert len(updated_message.content_blocks) == 1
|
||||
assert len(updated_message.content_blocks[0].contents) == 0
|
||||
# No-op handler: content_blocks stays untouched.
|
||||
assert updated_message.content_blocks == []
|
||||
assert isinstance(start_time, float)
|
||||
|
||||
|
||||
@ -347,7 +354,7 @@ async def test_handle_on_chain_end_with_output():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
)
|
||||
|
||||
output = AgentFinish(return_values={"output": "final output"}, log="test log")
|
||||
@ -369,7 +376,7 @@ async def test_handle_on_chain_end_no_output():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
)
|
||||
event = {"event": "on_chain_end", "data": {}, "start_time": 0}
|
||||
|
||||
@ -389,7 +396,7 @@ async def test_handle_on_chain_end_empty_data():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
)
|
||||
event = {"event": "on_chain_end", "data": {"output": None}, "start_time": 0}
|
||||
|
||||
@ -409,7 +416,7 @@ async def test_handle_on_chain_end_with_empty_return_values():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
)
|
||||
|
||||
class MockOutputEmptyReturnValues:
|
||||
@ -435,7 +442,7 @@ async def test_handle_on_tool_start():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
)
|
||||
event = {
|
||||
"event": "on_tool_start",
|
||||
@ -448,9 +455,9 @@ async def test_handle_on_tool_start():
|
||||
updated_message, start_time = await handle_on_tool_start(event, agent_message, tool_blocks_map, send_message, 0.0)
|
||||
|
||||
assert len(updated_message.content_blocks) == 1
|
||||
assert len(updated_message.content_blocks[0].contents) > 0
|
||||
assert len(updated_message.content_blocks) > 0
|
||||
tool_key = f"{event['name']}_{event['run_id']}"
|
||||
tool_content = updated_message.content_blocks[0].contents[-1]
|
||||
tool_content = updated_message.content_blocks[-1]
|
||||
assert tool_content == tool_blocks_map.get(tool_key)
|
||||
assert isinstance(tool_content, ToolContent)
|
||||
assert tool_content.name == "test_tool"
|
||||
@ -468,7 +475,7 @@ async def test_handle_on_tool_end():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
)
|
||||
|
||||
start_event = {
|
||||
@ -490,7 +497,7 @@ async def test_handle_on_tool_end():
|
||||
updated_message, start_time = await handle_on_tool_end(end_event, agent_message, tool_blocks_map, send_message, 0.0)
|
||||
|
||||
f"{end_event['name']}_{end_event['run_id']}"
|
||||
tool_content = updated_message.content_blocks[0].contents[-1]
|
||||
tool_content = updated_message.content_blocks[-1]
|
||||
assert tool_content.name == "test_tool"
|
||||
assert tool_content.output == "tool output"
|
||||
assert isinstance(tool_content.duration, int)
|
||||
@ -506,7 +513,7 @@ async def test_handle_on_tool_error():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
)
|
||||
|
||||
start_event = {
|
||||
@ -529,7 +536,7 @@ async def test_handle_on_tool_error():
|
||||
error_event, agent_message, tool_blocks_map, send_message, 0.0
|
||||
)
|
||||
|
||||
tool_content = updated_message.content_blocks[0].contents[-1]
|
||||
tool_content = updated_message.content_blocks[-1]
|
||||
assert tool_content.name == "test_tool"
|
||||
assert tool_content.error == "error message"
|
||||
assert tool_content.header["title"] == "Error using **test_tool**"
|
||||
@ -545,7 +552,7 @@ async def test_handle_on_chain_stream_with_output():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
)
|
||||
event = {
|
||||
"event": "on_chain_stream",
|
||||
@ -567,7 +574,7 @@ async def test_handle_on_chain_stream_no_output():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
session_id="test_session_id",
|
||||
)
|
||||
event = {
|
||||
@ -793,7 +800,7 @@ async def test_agent_streaming_no_text_accumulation():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
session_id="test_session_id",
|
||||
)
|
||||
# Add an ID to the message (normally set when persisted to DB)
|
||||
@ -863,7 +870,7 @@ async def test_agent_streaming_without_event_manager():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
session_id="test_session_id",
|
||||
)
|
||||
|
||||
@ -909,7 +916,7 @@ async def test_agent_streaming_skips_empty_chunks():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
session_id="test_session_id",
|
||||
)
|
||||
# Add an ID to the message (normally set when persisted to DB)
|
||||
@ -973,7 +980,7 @@ async def test_agent_streaming_preserves_message_id():
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
content_blocks=[],
|
||||
session_id="test_session_id",
|
||||
)
|
||||
|
||||
@ -1002,3 +1009,331 @@ async def test_agent_streaming_preserves_message_id():
|
||||
assert token_events[1]["id"] == "test-persisted-id"
|
||||
assert result.properties.state == "complete"
|
||||
assert result.text == "Hello world"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Flat-emission design tests
|
||||
#
|
||||
# These pin the chronological "stream of events" shape the new design calls
|
||||
# for: tool calls land flat in content_blocks in the order they fire, and
|
||||
# message.text triggers the setter to append a TextContent at the end. No
|
||||
# wrapping ContentBlock("Agent Steps", ...) is emitted.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_wrapper_group_for_tool_then_text():
|
||||
"""Two tools fired in order, then a final text -- everything stays flat.
|
||||
|
||||
With the on_chat_model_end-driven interleaving, the final text
|
||||
arrives through that handler (an AIMessage whose .content holds
|
||||
just the text). on_chain_end stashes the same string in
|
||||
data["text"] but doesn't append a TextContent — the model-end
|
||||
handler is the source of truth for the text blocks.
|
||||
"""
|
||||
send_message = create_mock_send_message()
|
||||
|
||||
final_ai_message = AIMessage(content=[{"type": "text", "text": "all done"}])
|
||||
output = AgentFinish(return_values={"output": "all done"}, log="")
|
||||
|
||||
events = [
|
||||
{
|
||||
"event": "on_tool_start",
|
||||
"name": "tool_a",
|
||||
"run_id": "run-a",
|
||||
"data": {"input": {"q": "1"}},
|
||||
"start_time": 0,
|
||||
},
|
||||
{
|
||||
"event": "on_tool_end",
|
||||
"name": "tool_a",
|
||||
"run_id": "run-a",
|
||||
"data": {"output": "a result"},
|
||||
"start_time": 0,
|
||||
},
|
||||
{
|
||||
"event": "on_tool_start",
|
||||
"name": "tool_b",
|
||||
"run_id": "run-b",
|
||||
"data": {"input": {"q": "2"}},
|
||||
"start_time": 0,
|
||||
},
|
||||
{
|
||||
"event": "on_tool_end",
|
||||
"name": "tool_b",
|
||||
"run_id": "run-b",
|
||||
"data": {"output": "b result"},
|
||||
"start_time": 0,
|
||||
},
|
||||
{"event": "on_chat_model_end", "data": {"output": final_ai_message}, "start_time": 0},
|
||||
{"event": "on_chain_end", "data": {"output": output}, "start_time": 0},
|
||||
]
|
||||
|
||||
agent_message = Message(
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[],
|
||||
session_id="s",
|
||||
)
|
||||
result = await process_agent_events(create_event_iterator(events), agent_message, send_message)
|
||||
|
||||
# No wrapping group; tools appear flat in fire order; final text last.
|
||||
types = [c.type for c in result.content_blocks]
|
||||
assert types == ["tool_use", "tool_use", "text"], types
|
||||
assert result.content_blocks[0].name == "tool_a"
|
||||
assert result.content_blocks[0].output == "a result"
|
||||
assert result.content_blocks[1].name == "tool_b"
|
||||
assert result.content_blocks[1].output == "b result"
|
||||
assert result.content_blocks[-1].text == "all done"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_end_updates_matching_pending_tool_only():
|
||||
"""Two identical tool invocations: tool_end must update the right one."""
|
||||
send_message = create_mock_send_message()
|
||||
|
||||
events = [
|
||||
{
|
||||
"event": "on_tool_start",
|
||||
"name": "search",
|
||||
"run_id": "run-1",
|
||||
"data": {"input": {"q": "shared"}},
|
||||
"start_time": 0,
|
||||
},
|
||||
{
|
||||
"event": "on_tool_end",
|
||||
"name": "search",
|
||||
"run_id": "run-1",
|
||||
"data": {"output": "first"},
|
||||
"start_time": 0,
|
||||
},
|
||||
{
|
||||
"event": "on_tool_start",
|
||||
"name": "search",
|
||||
"run_id": "run-2",
|
||||
"data": {"input": {"q": "shared"}},
|
||||
"start_time": 0,
|
||||
},
|
||||
{
|
||||
"event": "on_tool_end",
|
||||
"name": "search",
|
||||
"run_id": "run-2",
|
||||
"data": {"output": "second"},
|
||||
"start_time": 0,
|
||||
},
|
||||
]
|
||||
|
||||
agent_message = Message(
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[],
|
||||
session_id="s",
|
||||
)
|
||||
result = await process_agent_events(create_event_iterator(events), agent_message, send_message)
|
||||
|
||||
# Two distinct ToolContents, each with the right output -- the
|
||||
# `output is None` filter in handle_on_tool_end prevents the second
|
||||
# tool_end from clobbering the first tool's completed output.
|
||||
assert len(result.content_blocks) == 2
|
||||
assert [c.output for c in result.content_blocks] == ["first", "second"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_chain_start_is_a_noop_for_content_blocks():
|
||||
"""Synthetic 'Input' TextContent is gone -- chain_start touches nothing."""
|
||||
send_message = create_mock_send_message()
|
||||
|
||||
agent_message = Message(
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[],
|
||||
session_id="s",
|
||||
)
|
||||
events = [
|
||||
{
|
||||
"event": "on_chain_start",
|
||||
"data": {"input": {"input": "hello there", "chat_history": []}},
|
||||
"start_time": 0,
|
||||
},
|
||||
]
|
||||
result = await process_agent_events(create_event_iterator(events), agent_message, send_message)
|
||||
assert result.content_blocks == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_chain_end_does_not_append_text_block():
|
||||
"""chain_end stashes the final string in data["text"] but does NOT append a TextContent.
|
||||
|
||||
The on_chat_model_end handler is the source of truth for text
|
||||
blocks -- it walks each AIMessage.content list and appends text +
|
||||
tool_use in producer order so interleaved narration sits between
|
||||
the tool calls instead of being collapsed to one block at the end.
|
||||
chain_end appending its own TextContent would either duplicate the
|
||||
final round's text (already added by chat_model_end) or clobber
|
||||
the interleaved layout via the Message.text setter, which drops
|
||||
every existing TextContent and writes one at the end.
|
||||
|
||||
Message.text still returns "answer" because the getter falls back
|
||||
to data[text_key] when content_blocks holds no TextContent.
|
||||
"""
|
||||
send_message = create_mock_send_message()
|
||||
|
||||
output = AgentFinish(return_values={"output": "answer"}, log="")
|
||||
events = [{"event": "on_chain_end", "data": {"output": output}, "start_time": 0}]
|
||||
|
||||
agent_message = Message(
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[],
|
||||
session_id="s",
|
||||
)
|
||||
result = await process_agent_events(create_event_iterator(events), agent_message, send_message)
|
||||
|
||||
assert result.content_blocks == []
|
||||
assert result.text == "answer"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_chat_model_end_appends_interleaved_text_and_tool_blocks():
|
||||
"""on_chat_model_end walks AIMessage.content and appends each text and tool_use item in order.
|
||||
|
||||
Two model rounds:
|
||||
Round 1: [text "I'll fetch the date", tool_use get_date]
|
||||
-> on_chat_model_end appends both, in that order.
|
||||
-> on_tool_start finds the existing ToolContent (by name +
|
||||
output is None + not yet bound), overwrites its empty
|
||||
tool_input with the real input, no fallback append.
|
||||
-> on_tool_end fills .output.
|
||||
Round 2: [text "Got it -- summary"]
|
||||
-> on_chat_model_end appends one TextContent.
|
||||
-> on_chain_end stashes the same string in data["text"] but
|
||||
appends nothing.
|
||||
|
||||
Expected interleaved shape:
|
||||
[text "I'll fetch the date", tool_use get_date(output set),
|
||||
text "Got it -- summary"]
|
||||
"""
|
||||
send_message = create_mock_send_message()
|
||||
|
||||
round1 = AIMessage(
|
||||
content=[
|
||||
{"type": "text", "text": "I'll fetch the date"},
|
||||
# input_json_delta chunks haven't been merged yet at this
|
||||
# point in the real stream; the tool_use snapshot has {}.
|
||||
{"type": "tool_use", "name": "get_date", "input": {}, "id": "tu_1"},
|
||||
]
|
||||
)
|
||||
round2 = AIMessage(content=[{"type": "text", "text": "Got it -- summary"}])
|
||||
final = AgentFinish(return_values={"output": "Got it -- summary"}, log="")
|
||||
|
||||
events = [
|
||||
{"event": "on_chat_model_end", "data": {"output": round1}, "start_time": 0},
|
||||
# Real input arrives now via on_tool_start; the dedupe should
|
||||
# overwrite the empty {} on the existing ToolContent.
|
||||
{
|
||||
"event": "on_tool_start",
|
||||
"name": "get_date",
|
||||
"run_id": "run-1",
|
||||
"data": {"input": {"tz": "UTC"}},
|
||||
"start_time": 0,
|
||||
},
|
||||
{
|
||||
"event": "on_tool_end",
|
||||
"name": "get_date",
|
||||
"run_id": "run-1",
|
||||
"data": {"output": "2026-05-28 14:00 UTC"},
|
||||
"start_time": 0,
|
||||
},
|
||||
{"event": "on_chat_model_end", "data": {"output": round2}, "start_time": 0},
|
||||
{"event": "on_chain_end", "data": {"output": final}, "start_time": 0},
|
||||
]
|
||||
|
||||
agent_message = Message(
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[],
|
||||
session_id="s",
|
||||
)
|
||||
result = await process_agent_events(create_event_iterator(events), agent_message, send_message)
|
||||
|
||||
types = [c.type for c in result.content_blocks]
|
||||
assert types == ["text", "tool_use", "text"], types
|
||||
assert result.content_blocks[0].text == "I'll fetch the date"
|
||||
assert result.content_blocks[1].name == "get_date"
|
||||
# tool_input was overwritten by on_tool_start with the real value.
|
||||
assert result.content_blocks[1].tool_input == {"tz": "UTC"}
|
||||
assert result.content_blocks[1].output == "2026-05-28 14:00 UTC"
|
||||
assert result.content_blocks[2].text == "Got it -- summary"
|
||||
# Message.text concatenates every TextContent.
|
||||
assert result.text == "I'll fetch the dateGot it -- summary"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_chat_model_end_parallel_same_tool_keeps_order():
|
||||
"""Two parallel calls to the same tool: outputs land on the right block via order-based binding.
|
||||
|
||||
Each on_tool_start binds to the next unbound ToolContent in
|
||||
declaration order, so even when inputs are identical the
|
||||
name + output-is-None + not-yet-bound match keeps the bindings
|
||||
distinct.
|
||||
"""
|
||||
send_message = create_mock_send_message()
|
||||
|
||||
ai = AIMessage(
|
||||
content=[
|
||||
{"type": "tool_use", "name": "fetch", "input": {}, "id": "tu_a"},
|
||||
{"type": "tool_use", "name": "fetch", "input": {}, "id": "tu_b"},
|
||||
]
|
||||
)
|
||||
|
||||
events = [
|
||||
{"event": "on_chat_model_end", "data": {"output": ai}, "start_time": 0},
|
||||
{
|
||||
"event": "on_tool_start",
|
||||
"name": "fetch",
|
||||
"run_id": "run-a",
|
||||
"data": {"input": {"url": "a"}},
|
||||
"start_time": 0,
|
||||
},
|
||||
{
|
||||
"event": "on_tool_start",
|
||||
"name": "fetch",
|
||||
"run_id": "run-b",
|
||||
"data": {"input": {"url": "b"}},
|
||||
"start_time": 0,
|
||||
},
|
||||
{
|
||||
"event": "on_tool_end",
|
||||
"name": "fetch",
|
||||
"run_id": "run-a",
|
||||
"data": {"output": "result a"},
|
||||
"start_time": 0,
|
||||
},
|
||||
{
|
||||
"event": "on_tool_end",
|
||||
"name": "fetch",
|
||||
"run_id": "run-b",
|
||||
"data": {"output": "result b"},
|
||||
"start_time": 0,
|
||||
},
|
||||
]
|
||||
|
||||
agent_message = Message(
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name="Agent",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[],
|
||||
session_id="s",
|
||||
)
|
||||
result = await process_agent_events(create_event_iterator(events), agent_message, send_message)
|
||||
|
||||
assert len(result.content_blocks) == 2
|
||||
assert result.content_blocks[0].tool_input == {"url": "a"}
|
||||
assert result.content_blocks[0].output == "result a"
|
||||
assert result.content_blocks[1].tool_input == {"url": "b"}
|
||||
assert result.content_blocks[1].output == "result b"
|
||||
|
||||
@ -535,6 +535,46 @@ async def test_aupdate_message_with_dataframe_in_tool_output(created_message):
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class TestMessageBaseFromMessageAgentInit:
|
||||
"""Regression: in-flight agent Message must survive the no_content check.
|
||||
|
||||
The agent now initializes with flat ``content_blocks=[]`` (the wrapping
|
||||
``ContentBlock('Agent Steps', ...)`` is gone) and uses ``text=""`` as
|
||||
the "intentionally created, content will arrive" sentinel. If either
|
||||
side regresses, the build dies with "The message does not have the
|
||||
required fields (text, sender, sender_name)." before any agent event
|
||||
can populate the content_blocks list.
|
||||
"""
|
||||
|
||||
def test_from_message_accepts_in_flight_agent_message(self):
|
||||
from langflow.services.database.models.message.model import MessageTable
|
||||
|
||||
# Mirrors what AgentComponent / LCToolsAgentComponent / altk build.
|
||||
message = Message(
|
||||
text="",
|
||||
sender="Machine",
|
||||
sender_name="Agent",
|
||||
content_blocks=[],
|
||||
session_id="test-session",
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
)
|
||||
|
||||
result = MessageTable.from_message(message, flow_id=uuid4())
|
||||
|
||||
assert result.sender == "Machine"
|
||||
assert result.sender_name == "Agent"
|
||||
assert result.text == ""
|
||||
|
||||
def test_from_message_rejects_truly_empty_message(self):
|
||||
from langflow.services.database.models.message.model import MessageTable
|
||||
|
||||
# No text, no text_stream, no content_blocks -- not intentional.
|
||||
message = Message(sender="Machine", sender_name="Agent", content_blocks=[])
|
||||
|
||||
with pytest.raises(ValueError, match="required fields"):
|
||||
MessageTable.from_message(message, flow_id=uuid4())
|
||||
|
||||
|
||||
class TestMessageBaseFromMessageFilePaths:
|
||||
"""Tests for the file path handling in MessageBase.from_message."""
|
||||
|
||||
|
||||
@ -10,7 +10,8 @@ import ForwardedIconComponent from "../../common/genericIconComponent";
|
||||
import SimplifiedCodeTabComponent from "../codeTabsComponent";
|
||||
import DurationDisplay from "./DurationDisplay";
|
||||
import { SourcesStrip } from "./SourcesStrip";
|
||||
import { looksPreformatted, unwrapToolMessage } from "./toolOutput";
|
||||
import { ToolOutputDisplay } from "./ToolOutputDisplay";
|
||||
import { ToolSection } from "./ToolSection";
|
||||
|
||||
export default function ContentDisplay({
|
||||
content,
|
||||
@ -138,72 +139,12 @@ export default function ContentDisplay({
|
||||
break;
|
||||
|
||||
case "tool_use": {
|
||||
// Tool output rendering routes by the *unwrapped* payload shape:
|
||||
// - markdown-y string -> Markdown renderer (prose, no chrome)
|
||||
// - pre-formatted text -> code tab with language=text
|
||||
// (monospace, contained horizontal scroll,
|
||||
// built-in copy button)
|
||||
// - object / array -> code tab with language=json (same)
|
||||
// The LangChain ToolMessage envelope is unwrapped first so the
|
||||
// readable `content` field becomes the body and the plumbing
|
||||
// metadata (already shown by the accordion trigger) stays hidden.
|
||||
const formatToolOutput = (raw: JSONValue) => {
|
||||
const output = unwrapToolMessage(raw);
|
||||
if (output === null || output === undefined) return null;
|
||||
|
||||
if (typeof output === "string") {
|
||||
if (looksPreformatted(output)) {
|
||||
return <SimplifiedCodeTabComponent language="text" code={output} />;
|
||||
}
|
||||
return (
|
||||
<Markdown
|
||||
remarkPlugins={[remarkGfm]}
|
||||
rehypePlugins={[rehypeMathjax]}
|
||||
className="markdown prose max-w-full text-sm font-normal dark:prose-invert"
|
||||
components={{
|
||||
pre({ node, ...props }) {
|
||||
return <>{props.children}</>;
|
||||
},
|
||||
ol({ node, ...props }) {
|
||||
return <ol className="max-w-full">{props.children}</ol>;
|
||||
},
|
||||
ul({ node, ...props }) {
|
||||
return <ul className="max-w-full">{props.children}</ul>;
|
||||
},
|
||||
code: ({ node, className, children, ...props }) => {
|
||||
const content = String(children);
|
||||
if (isCodeBlock(className, props, content)) {
|
||||
return (
|
||||
<SimplifiedCodeTabComponent
|
||||
language={extractLanguage(className)}
|
||||
code={content.replace(/\n$/, "")}
|
||||
/>
|
||||
);
|
||||
}
|
||||
return (
|
||||
<code className={className} {...props}>
|
||||
{children}
|
||||
</code>
|
||||
);
|
||||
},
|
||||
}}
|
||||
>
|
||||
{output}
|
||||
</Markdown>
|
||||
);
|
||||
}
|
||||
|
||||
try {
|
||||
return (
|
||||
<SimplifiedCodeTabComponent
|
||||
language="json"
|
||||
code={JSON.stringify(output, null, 2)}
|
||||
/>
|
||||
);
|
||||
} catch {
|
||||
return String(output);
|
||||
}
|
||||
};
|
||||
// Tool output rendering lives in ToolOutputDisplay — it routes by
|
||||
// shape (markdown string / preformatted text / object) and, when
|
||||
// the producer hands us a LangChain ToolMessage envelope with
|
||||
// non-standard metadata (additional_kwargs, response_metadata,
|
||||
// artifact, ...), surfaces it under a 2-tab UI so the readable
|
||||
// content stays primary and the plumbing is one click away.
|
||||
|
||||
// Backend serializes ToolContent.tool_input under its alias `input`
|
||||
// when by_alias=True (AG-UI emission, certain dump paths). Prefer
|
||||
@ -219,28 +160,32 @@ export default function ContentDisplay({
|
||||
content.output !== null &&
|
||||
!(typeof content.output === "string" && content.output.trim() === "");
|
||||
const hasError = content.error != null;
|
||||
// Eyebrow labels (INPUT/OUTPUT/ERROR) used to bracket each section,
|
||||
// but the surrounding accordion card is already the "tool call"
|
||||
// context — extra labels just add chrome. Match the assistant-ui /
|
||||
// Claude pattern: args and result stack directly inside the card,
|
||||
// separated by a hairline rule. Empty sections render nothing.
|
||||
// Each section carries an eyebrow label (Arguments / Error) so the
|
||||
// parts of a tool call read clearly inside the accordion card. The
|
||||
// output section renders through ToolOutputDisplay, which supplies its
|
||||
// own Result/Metadata tabs, and a hairline rule separates the input
|
||||
// from the result. Empty sections render nothing.
|
||||
const showSeparator = hasInput && (hasOutput || hasError);
|
||||
contentData = (
|
||||
<div className="flex flex-col gap-3">
|
||||
{hasInput && <ToolInputDisplay input={toolInput} />}
|
||||
{hasInput && (
|
||||
<ToolSection eyebrow="Arguments">
|
||||
<ToolInputDisplay input={toolInput} />
|
||||
</ToolSection>
|
||||
)}
|
||||
{showSeparator && <div className="h-px bg-border" />}
|
||||
{hasOutput && (
|
||||
<div className="max-h-96 overflow-auto">
|
||||
{formatToolOutput(content.output as JSONValue)}
|
||||
</div>
|
||||
<ToolOutputDisplay output={content.output as JSONValue} />
|
||||
)}
|
||||
{hasError && (
|
||||
<div className="rounded-md bg-destructive/10 text-destructive">
|
||||
<SimplifiedCodeTabComponent
|
||||
language="json"
|
||||
code={JSON.stringify(content.error, null, 2)}
|
||||
/>
|
||||
</div>
|
||||
<ToolSection eyebrow="Error">
|
||||
<div className="rounded-md bg-destructive/10 text-destructive">
|
||||
<SimplifiedCodeTabComponent
|
||||
language="json"
|
||||
code={JSON.stringify(content.error, null, 2)}
|
||||
/>
|
||||
</div>
|
||||
</ToolSection>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
|
||||
@ -0,0 +1,216 @@
|
||||
import { AnimatePresence, motion } from "framer-motion";
|
||||
import { useState } from "react";
|
||||
import Markdown from "react-markdown";
|
||||
import rehypeMathjax from "rehype-mathjax/browser";
|
||||
import remarkGfm from "remark-gfm";
|
||||
import SimplifiedCodeTabComponent from "@/components/core/codeTabsComponent";
|
||||
import type { JSONValue } from "@/types/chat";
|
||||
import { extractLanguage, isCodeBlock } from "@/utils/codeBlockUtils";
|
||||
import { cn } from "@/utils/utils";
|
||||
import { ToolSection } from "./ToolSection";
|
||||
import {
|
||||
isToolMessageEnvelope,
|
||||
looksPreformatted,
|
||||
unwrapToolMessage,
|
||||
} from "./toolOutput";
|
||||
|
||||
/** Render a single tool output value as the best-fit primitive:
|
||||
* - markdown-y string -> Markdown renderer (prose)
|
||||
* - pre-formatted text -> SimplifiedCodeTabComponent (text)
|
||||
* - anything else -> SimplifiedCodeTabComponent (json)
|
||||
*
|
||||
* Extracted from ContentDisplay so the tabbed envelope visualizer can
|
||||
* reuse the same routing for whichever value lives behind a tab. */
|
||||
function FormattedOutput({ value }: { value: JSONValue }) {
|
||||
if (value === null || value === undefined) return null;
|
||||
|
||||
if (typeof value === "string") {
|
||||
if (looksPreformatted(value)) {
|
||||
return <SimplifiedCodeTabComponent language="text" code={value} />;
|
||||
}
|
||||
return (
|
||||
<Markdown
|
||||
remarkPlugins={[remarkGfm]}
|
||||
rehypePlugins={[rehypeMathjax]}
|
||||
className="markdown prose max-w-full text-sm font-normal dark:prose-invert"
|
||||
components={{
|
||||
pre({ node, ...props }) {
|
||||
return <>{props.children}</>;
|
||||
},
|
||||
ol({ node, ...props }) {
|
||||
return <ol className="max-w-full">{props.children}</ol>;
|
||||
},
|
||||
ul({ node, ...props }) {
|
||||
return <ul className="max-w-full">{props.children}</ul>;
|
||||
},
|
||||
code: ({ node, className, children, ...props }) => {
|
||||
const content = String(children);
|
||||
if (isCodeBlock(className, props, content)) {
|
||||
return (
|
||||
<SimplifiedCodeTabComponent
|
||||
language={extractLanguage(className)}
|
||||
code={content.replace(/\n$/, "")}
|
||||
/>
|
||||
);
|
||||
}
|
||||
return (
|
||||
<code className={className} {...props}>
|
||||
{children}
|
||||
</code>
|
||||
);
|
||||
},
|
||||
}}
|
||||
>
|
||||
{value}
|
||||
</Markdown>
|
||||
);
|
||||
}
|
||||
|
||||
try {
|
||||
return (
|
||||
<SimplifiedCodeTabComponent
|
||||
language="json"
|
||||
code={JSON.stringify(value, null, 2)}
|
||||
/>
|
||||
);
|
||||
} catch {
|
||||
return <span>{String(value)}</span>;
|
||||
}
|
||||
}
|
||||
|
||||
type Tab = "result" | "metadata";
|
||||
|
||||
/** Tab control sized for the inside of a tool-call card. Matches the
|
||||
* underline-on-active pattern used across assistant-ui / Claude /
|
||||
* ChatGPT — quiet by default, the active tab gets a 2px underline in
|
||||
* the primary color. */
|
||||
function TabButton({
|
||||
selected,
|
||||
onClick,
|
||||
children,
|
||||
}: {
|
||||
selected: boolean;
|
||||
onClick: () => void;
|
||||
children: React.ReactNode;
|
||||
}) {
|
||||
return (
|
||||
<button
|
||||
type="button"
|
||||
onClick={onClick}
|
||||
aria-selected={selected}
|
||||
role="tab"
|
||||
className={cn(
|
||||
"px-1 pb-1.5 -mb-px text-xs font-medium border-b-2 transition-colors",
|
||||
"focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring focus-visible:ring-offset-2 rounded-sm",
|
||||
selected
|
||||
? "text-primary border-primary"
|
||||
: "text-muted-foreground border-transparent hover:text-primary",
|
||||
)}
|
||||
>
|
||||
{children}
|
||||
</button>
|
||||
);
|
||||
}
|
||||
|
||||
/** Top-level renderer for a tool's output, wrapped in an "Output"
|
||||
* eyebrow so the section is unambiguous next to the "Arguments" block
|
||||
* above it. Routes by shape:
|
||||
* - LangChain ToolMessage envelope with non-standard metadata keys
|
||||
* (additional_kwargs, response_metadata, artifact, type, ...) gets
|
||||
* a 2-tab UI: "Result" shows the inner `.content` rendered through
|
||||
* FormattedOutput, "Metadata" shows the rest of the envelope as
|
||||
* pretty JSON. Plumbing keys (name, id, tool_call_id, status) are
|
||||
* suppressed because the accordion trigger already surfaces them.
|
||||
* - Anything else (simple string, plain dict, unwrappable envelope)
|
||||
* falls through to FormattedOutput directly under the eyebrow —
|
||||
* no tabs, just the body. */
|
||||
export function ToolOutputDisplay({ output }: { output: JSONValue }) {
|
||||
const [tab, setTab] = useState<Tab>("result");
|
||||
|
||||
if (!isToolMessageEnvelope(output)) {
|
||||
return (
|
||||
<ToolSection eyebrow="Output">
|
||||
<div className="max-h-96 overflow-auto">
|
||||
<FormattedOutput value={unwrapToolMessage(output)} />
|
||||
</div>
|
||||
</ToolSection>
|
||||
);
|
||||
}
|
||||
|
||||
const content = output.content;
|
||||
// Strip the standard ToolMessage plumbing keys from the metadata view —
|
||||
// the accordion trigger already shows tool name and status, and id /
|
||||
// tool_call_id aren't useful in the UI. What remains is the actually
|
||||
// interesting custom metadata (additional_kwargs, response_metadata,
|
||||
// artifact, type, custom fields).
|
||||
const metadata = Object.fromEntries(
|
||||
Object.entries(output).filter(
|
||||
([k]) => !TOOL_MESSAGE_KEYS_PLUS_CONTENT.has(k),
|
||||
),
|
||||
);
|
||||
const hasMetadata = Object.keys(metadata).length > 0;
|
||||
|
||||
// If after stripping plumbing there's no meaningful metadata left,
|
||||
// drop the tabs and render content directly.
|
||||
if (!hasMetadata) {
|
||||
return (
|
||||
<ToolSection eyebrow="Output">
|
||||
<FormattedOutput value={content} />
|
||||
</ToolSection>
|
||||
);
|
||||
}
|
||||
|
||||
return (
|
||||
<ToolSection eyebrow="Output">
|
||||
<div role="tablist" className="flex gap-4 border-b border-border">
|
||||
<TabButton selected={tab === "result"} onClick={() => setTab("result")}>
|
||||
Result
|
||||
</TabButton>
|
||||
<TabButton
|
||||
selected={tab === "metadata"}
|
||||
onClick={() => setTab("metadata")}
|
||||
>
|
||||
Metadata
|
||||
</TabButton>
|
||||
</div>
|
||||
{/* Stable height envelope: min-h holds the card chrome steady when
|
||||
* switching from a short Result tab to a tall Metadata tab (and
|
||||
* vice versa), and max-h caps growth so tall metadata scrolls
|
||||
* inside instead of pushing everything below the card down.
|
||||
* AnimatePresence mode="wait" lets the outgoing tab finish fading
|
||||
* before the incoming one paints, so we don't have to absolute-
|
||||
* position the layers (which would break the scroll). */}
|
||||
<div className="min-h-[180px] max-h-96 overflow-auto">
|
||||
<AnimatePresence initial={false} mode="wait">
|
||||
<motion.div
|
||||
key={tab}
|
||||
initial={{ opacity: 0 }}
|
||||
animate={{ opacity: 1 }}
|
||||
exit={{ opacity: 0 }}
|
||||
transition={{ duration: 0.08, ease: "easeOut" }}
|
||||
>
|
||||
{tab === "result" ? (
|
||||
<FormattedOutput value={content} />
|
||||
) : (
|
||||
<SimplifiedCodeTabComponent
|
||||
language="json"
|
||||
code={JSON.stringify(metadata, null, 2)}
|
||||
/>
|
||||
)}
|
||||
</motion.div>
|
||||
</AnimatePresence>
|
||||
</div>
|
||||
</ToolSection>
|
||||
);
|
||||
}
|
||||
|
||||
// `content` belongs in its own tab; the rest of the canonical plumbing
|
||||
// fields aren't worth exposing — keep this set local rather than
|
||||
// re-exporting yet another constant.
|
||||
const TOOL_MESSAGE_KEYS_PLUS_CONTENT = new Set([
|
||||
"content",
|
||||
"name",
|
||||
"id",
|
||||
"tool_call_id",
|
||||
"status",
|
||||
]);
|
||||
@ -0,0 +1,25 @@
|
||||
import type { ReactNode } from "react";
|
||||
|
||||
/** Eyebrow + body wrapper for a section inside a tool-call card.
|
||||
* Used to label Arguments, Output, and Error sections so the reader can
|
||||
* tell at a glance which part of the tool call they're looking at — the
|
||||
* surrounding accordion only signals "this is a tool call", not where
|
||||
* the boundary between input and result sits. The eyebrow style is
|
||||
* intentionally quiet (small-caps muted text) so it scaffolds the
|
||||
* sections without competing with the actual content. */
|
||||
export function ToolSection({
|
||||
eyebrow,
|
||||
children,
|
||||
}: {
|
||||
eyebrow: string;
|
||||
children: ReactNode;
|
||||
}) {
|
||||
return (
|
||||
<div className="flex flex-col gap-1.5">
|
||||
<div className="text-[10px] uppercase tracking-wider font-medium text-muted-foreground">
|
||||
{eyebrow}
|
||||
</div>
|
||||
{children}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
@ -282,9 +282,11 @@ describe("ContentDisplay", () => {
|
||||
expect(container.querySelector(".max-h-96")).toBeNull();
|
||||
});
|
||||
|
||||
it("keeps the JSON block when output has extra fields beyond ToolMessage", () => {
|
||||
// Don't unwrap if the producer added fields the user might need —
|
||||
// could be tool-specific data the renderer shouldn't drop silently.
|
||||
it("renders a tabbed view when output is an envelope with extra metadata", () => {
|
||||
// Tool-specific data (anything beyond {content, name, id,
|
||||
// tool_call_id, status}) gets surfaced under a Metadata tab so the
|
||||
// user can inspect it without burying the readable content. The
|
||||
// default-selected Content tab shows the inner `content` value.
|
||||
const tool = {
|
||||
type: "tool_use",
|
||||
name: "fetch_content",
|
||||
@ -294,9 +296,16 @@ describe("ContentDisplay", () => {
|
||||
custom_field: 42,
|
||||
},
|
||||
} as unknown as ContentBlockItem;
|
||||
render(<ContentDisplay content={tool} chatId="t-t-no-unwrap" />);
|
||||
// Falls through to the JSON code block (mocked as code-tabs).
|
||||
expect(screen.getByTestId("code-tabs")).toBeInTheDocument();
|
||||
render(<ContentDisplay content={tool} chatId="t-t-tabs" />);
|
||||
const tabs = screen.getAllByRole("tab");
|
||||
expect(tabs).toHaveLength(2);
|
||||
expect(tabs[0]).toHaveTextContent("Result");
|
||||
expect(tabs[1]).toHaveTextContent("Metadata");
|
||||
// Result tab is the default selection.
|
||||
expect(tabs[0]).toHaveAttribute("aria-selected", "true");
|
||||
expect(tabs[1]).toHaveAttribute("aria-selected", "false");
|
||||
// Body of the Result tab renders the inner string, not a JSON dump.
|
||||
expect(screen.getByText("the body")).toBeInTheDocument();
|
||||
});
|
||||
|
||||
it("renders the error body in a destructive-toned panel", () => {
|
||||
|
||||
@ -3,7 +3,13 @@ import DurationDisplay from "../DurationDisplay";
|
||||
|
||||
// Mock AnimatedNumber component
|
||||
jest.mock("../../../common/animatedNumbers", () => ({
|
||||
AnimatedNumber: ({ value, humanizedValue }: any) => (
|
||||
AnimatedNumber: ({
|
||||
value,
|
||||
humanizedValue,
|
||||
}: {
|
||||
value: number;
|
||||
humanizedValue: string;
|
||||
}) => (
|
||||
<span data-testid="animated-number" data-value={value}>
|
||||
{humanizedValue}
|
||||
</span>
|
||||
@ -13,7 +19,7 @@ jest.mock("../../../common/animatedNumbers", () => ({
|
||||
// Mock Loading component
|
||||
jest.mock("../../../ui/loading", () => ({
|
||||
__esModule: true,
|
||||
default: ({ className }: any) => (
|
||||
default: ({ className }: { className?: string }) => (
|
||||
<div data-testid="loading-spinner" className={className}>
|
||||
Loading...
|
||||
</div>
|
||||
|
||||
@ -1,5 +1,9 @@
|
||||
import type { JSONValue } from "@/types/chat";
|
||||
import { looksPreformatted, unwrapToolMessage } from "../toolOutput";
|
||||
import {
|
||||
isToolMessageEnvelope,
|
||||
looksPreformatted,
|
||||
unwrapToolMessage,
|
||||
} from "../toolOutput";
|
||||
|
||||
describe("unwrapToolMessage", () => {
|
||||
it("unwraps a canonical LangChain ToolMessage shape down to its .content", () => {
|
||||
@ -37,6 +41,53 @@ describe("unwrapToolMessage", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("isToolMessageEnvelope", () => {
|
||||
it("recognises an envelope carrying non-standard metadata", () => {
|
||||
// The shape ToolOutputDisplay surfaces under a Metadata tab: a
|
||||
// LangChain ToolMessage with the full plumbing (additional_kwargs,
|
||||
// response_metadata, type, artifact) that the user might want to
|
||||
// inspect.
|
||||
expect(
|
||||
isToolMessageEnvelope({
|
||||
content: "Current date: 2026-05-28",
|
||||
additional_kwargs: {},
|
||||
response_metadata: {},
|
||||
type: "tool",
|
||||
name: "get_current_date",
|
||||
id: "x",
|
||||
tool_call_id: "toolu_y",
|
||||
artifact: null,
|
||||
status: "success",
|
||||
}),
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it("does not recognise outputs whose keys are only standard plumbing", () => {
|
||||
// {content, name, id, tool_call_id, status} — no metadata worth
|
||||
// hiding behind a tab. unwrapToolMessage handles this directly.
|
||||
expect(
|
||||
isToolMessageEnvelope({
|
||||
content: "the body",
|
||||
name: "fetch",
|
||||
id: "abc",
|
||||
tool_call_id: "toolu_x",
|
||||
status: "success",
|
||||
}),
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
it("rejects objects without a content key", () => {
|
||||
expect(isToolMessageEnvelope({ result: "12" })).toBe(false);
|
||||
});
|
||||
|
||||
it("rejects arrays, primitives, and null", () => {
|
||||
expect(isToolMessageEnvelope([1, 2, 3] as JSONValue)).toBe(false);
|
||||
expect(isToolMessageEnvelope("hi")).toBe(false);
|
||||
expect(isToolMessageEnvelope(42)).toBe(false);
|
||||
expect(isToolMessageEnvelope(null)).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("looksPreformatted", () => {
|
||||
it("flags pandas-style column-padded output", () => {
|
||||
// The actual pandas df.to_string() output that broke the playground:
|
||||
|
||||
@ -4,7 +4,7 @@ import type { JSONValue } from "@/types/chat";
|
||||
* `.content`; the rest is plumbing already exposed by the surrounding
|
||||
* accordion trigger (tool name, id, status). Used to decide whether to
|
||||
* unwrap an output object down to its `.content` field. */
|
||||
const TOOL_MESSAGE_KEYS = new Set([
|
||||
export const TOOL_MESSAGE_KEYS = new Set([
|
||||
"content",
|
||||
"name",
|
||||
"id",
|
||||
@ -12,6 +12,23 @@ const TOOL_MESSAGE_KEYS = new Set([
|
||||
"status",
|
||||
]);
|
||||
|
||||
/** Detect an "envelope" tool output worth surfacing in a tabbed view:
|
||||
* a plain object with a `content` field AND at least one key OUTSIDE the
|
||||
* standard ToolMessage subset (additional_kwargs, response_metadata,
|
||||
* artifact, type, ...). Outputs that only carry standard plumbing keys
|
||||
* (name, id, tool_call_id, status) get unwrapped by `unwrapToolMessage`
|
||||
* — they don't earn a tab strip because there's nothing user-relevant
|
||||
* to hide behind it. */
|
||||
export function isToolMessageEnvelope(
|
||||
output: JSONValue,
|
||||
): output is Record<string, JSONValue> {
|
||||
if (output === null || typeof output !== "object" || Array.isArray(output)) {
|
||||
return false;
|
||||
}
|
||||
if (!("content" in output)) return false;
|
||||
return Object.keys(output).some((k) => !TOOL_MESSAGE_KEYS.has(k));
|
||||
}
|
||||
|
||||
/** Strip the LangChain ToolMessage envelope from a tool output, returning
|
||||
* the inner `content` value when the keys are a subset of the canonical
|
||||
* set. Any extra key (custom tool-specific data) prevents unwrap so the
|
||||
|
||||
File diff suppressed because one or more lines are too long
@ -19,7 +19,6 @@ from lfx.inputs.inputs import InputTypes
|
||||
from lfx.io import BoolInput, HandleInput, IntInput, MessageInput
|
||||
from lfx.log.logger import logger
|
||||
from lfx.memory import delete_message
|
||||
from lfx.schema.content_block import ContentBlock
|
||||
from lfx.schema.data import Data
|
||||
from lfx.schema.log import OnTokenFunctionType
|
||||
from lfx.schema.message import Message
|
||||
@ -243,7 +242,12 @@ class LCAgentComponent(Component):
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name=sender_name,
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
# `text=""` sentinel so MessageTable's no_content check accepts
|
||||
# an in-flight agent message whose content_blocks haven't been
|
||||
# populated yet. Mirrors ChatInput's convention.
|
||||
text="",
|
||||
# Flat chronological event log; see lfx.base.agents.events.
|
||||
content_blocks=[],
|
||||
session_id=session_id or uuid.uuid4(),
|
||||
)
|
||||
|
||||
|
||||
@ -26,7 +26,6 @@ from lfx.base.agents.utils import data_to_messages, get_chat_output_sender_name
|
||||
from lfx.components.models_and_agents import AgentComponent
|
||||
from lfx.log.logger import logger
|
||||
from lfx.memory import delete_message
|
||||
from lfx.schema.content_block import ContentBlock
|
||||
from lfx.schema.data import Data
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@ -379,7 +378,12 @@ class ALTKBaseAgentComponent(AgentComponent):
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name=sender_name,
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
# `text=""` sentinel so MessageTable's no_content check accepts
|
||||
# an in-flight agent message whose content_blocks haven't been
|
||||
# populated yet. Mirrors ChatInput's convention.
|
||||
text="",
|
||||
# Flat chronological event log; see lfx.base.agents.events.
|
||||
content_blocks=[],
|
||||
session_id=session_id or uuid.uuid4(),
|
||||
)
|
||||
try:
|
||||
|
||||
@ -5,10 +5,8 @@ from time import perf_counter
|
||||
from typing import Any, Protocol
|
||||
|
||||
from langchain_core.agents import AgentFinish
|
||||
from langchain_core.messages import AIMessageChunk, BaseMessage
|
||||
from typing_extensions import TypedDict
|
||||
from langchain_core.messages import AIMessageChunk
|
||||
|
||||
from lfx.schema.content_block import ContentBlock
|
||||
from lfx.schema.content_types import TextContent, ToolContent
|
||||
from lfx.schema.log import OnTokenFunctionType, SendMessageFunctionType
|
||||
from lfx.schema.message import Message
|
||||
@ -28,16 +26,6 @@ class ExceptionWithMessageError(Exception):
|
||||
)
|
||||
|
||||
|
||||
class InputDict(TypedDict):
|
||||
input: str
|
||||
chat_history: list[BaseMessage]
|
||||
|
||||
|
||||
def _build_agent_input_text_content(agent_input_dict: InputDict) -> str:
|
||||
final_input = agent_input_dict.get("input", "")
|
||||
return f"{final_input}"
|
||||
|
||||
|
||||
def _calculate_duration(start_time: float) -> int:
|
||||
"""Calculate duration in milliseconds from start time to now."""
|
||||
# Handle the calculation
|
||||
@ -54,42 +42,20 @@ def _calculate_duration(start_time: float) -> int:
|
||||
|
||||
|
||||
async def handle_on_chain_start(
|
||||
event: dict[str, Any],
|
||||
event: dict[str, Any], # noqa: ARG001
|
||||
agent_message: Message,
|
||||
send_message_callback: SendMessageFunctionType,
|
||||
send_message_callback: SendMessageFunctionType, # noqa: ARG001
|
||||
send_token_callback: OnTokenFunctionType | None, # noqa: ARG001
|
||||
start_time: float,
|
||||
*,
|
||||
had_streaming: bool = False, # noqa: ARG001
|
||||
message_id: str | None = None, # noqa: ARG001
|
||||
) -> tuple[Message, float]:
|
||||
# Create content blocks if they don't exist
|
||||
if not agent_message.content_blocks:
|
||||
agent_message.content_blocks = [ContentBlock(title="Agent Steps", contents=[])]
|
||||
|
||||
if event["data"].get("input"):
|
||||
input_data = event["data"].get("input")
|
||||
if isinstance(input_data, dict) and "input" in input_data:
|
||||
# Cast the input_data to InputDict
|
||||
input_message = input_data.get("input", "")
|
||||
if isinstance(input_message, BaseMessage):
|
||||
input_message = input_message.text()
|
||||
elif not isinstance(input_message, str):
|
||||
input_message = str(input_message)
|
||||
|
||||
input_dict: InputDict = {
|
||||
"input": input_message,
|
||||
"chat_history": input_data.get("chat_history", []),
|
||||
}
|
||||
text_content = TextContent(
|
||||
type="text",
|
||||
text=_build_agent_input_text_content(input_dict),
|
||||
duration=_calculate_duration(start_time),
|
||||
header={"title": "Input", "icon": "MessageSquare"},
|
||||
)
|
||||
agent_message.content_blocks[0].contents.append(text_content)
|
||||
agent_message = await send_message_callback(message=agent_message, skip_db_update=True)
|
||||
start_time = perf_counter()
|
||||
# No-op. The synthetic "Input" TextContent that used to live inside the
|
||||
# "Agent Steps" group is gone in the flat content_blocks design — the
|
||||
# user message is already rendered above the agent's reply, so echoing
|
||||
# it back as a content block was duplicative. content_blocks is now a
|
||||
# chronological event log, populated by the tool / chain handlers below.
|
||||
return agent_message, start_time
|
||||
|
||||
|
||||
@ -162,18 +128,17 @@ async def handle_on_chain_end(
|
||||
if data_output and isinstance(data_output, AgentFinish) and data_output.return_values.get("output"):
|
||||
output = data_output.return_values.get("output")
|
||||
|
||||
agent_message.text = _extract_output_text(output)
|
||||
# Don't reassign agent_message.text here. The setter drops every
|
||||
# existing TextContent and appends a single one at the end, which
|
||||
# would collapse the interleaved text + tool_use blocks the
|
||||
# on_chat_model_end handler appends in producer order. The text
|
||||
# for the final round is already in content_blocks, and Message.text
|
||||
# is a computed_field over those TextContent entries — getting
|
||||
# the answer back out is automatic. Stash the extracted string in
|
||||
# data["text"] so legacy consumers reading message.data["text"]
|
||||
# still see the final answer.
|
||||
agent_message.data[agent_message.text_key] = _extract_output_text(output) or ""
|
||||
agent_message.properties.state = "complete"
|
||||
# Add duration to the last content if it exists
|
||||
if agent_message.content_blocks:
|
||||
duration = _calculate_duration(start_time)
|
||||
text_content = TextContent(
|
||||
type="text",
|
||||
text=agent_message.text,
|
||||
duration=duration,
|
||||
header={"title": "Output", "icon": "MessageSquare"},
|
||||
)
|
||||
agent_message.content_blocks[0].contents.append(text_content)
|
||||
|
||||
# Only send final message if we didn't have streaming chunks
|
||||
# If we had streaming, frontend already accumulated the chunks
|
||||
@ -183,6 +148,94 @@ async def handle_on_chain_end(
|
||||
return agent_message, start_time
|
||||
|
||||
|
||||
def _coerce_ai_message_blocks(content: Any) -> list[dict[str, Any]]:
|
||||
"""Normalise an AIMessage.content into a list[dict] of typed blocks.
|
||||
|
||||
Anthropic emits ``content`` as ``list[dict]`` where each dict has a
|
||||
``type`` of ``"text"`` / ``"tool_use"`` / ``"input_json_delta"`` /
|
||||
etc. OpenAI-style providers emit ``content`` as a plain string for
|
||||
text-only turns and surface tool calls separately on ``.tool_calls``.
|
||||
|
||||
Return shape is always ``list[{"type": ..., ...}]`` so the caller
|
||||
walks one structure. Text-only strings turn into a single
|
||||
``{"type": "text", "text": str}``. Anything we don't recognise is
|
||||
skipped — the on_tool_start fallback in handle_on_tool_start will
|
||||
still pick up tool calls if a provider routes them outside .content.
|
||||
"""
|
||||
if isinstance(content, str):
|
||||
return [{"type": "text", "text": content}] if content else []
|
||||
if not isinstance(content, list):
|
||||
return []
|
||||
return [item for item in content if isinstance(item, dict) and item.get("type") in {"text", "tool_use"}]
|
||||
|
||||
|
||||
async def handle_on_chat_model_end(
|
||||
event: dict[str, Any],
|
||||
agent_message: Message,
|
||||
send_message_callback: SendMessageFunctionType,
|
||||
send_token_callback: OnTokenFunctionType | None, # noqa: ARG001
|
||||
start_time: float,
|
||||
*,
|
||||
had_streaming: bool = False, # noqa: ARG001
|
||||
message_id: str | None = None, # noqa: ARG001
|
||||
) -> tuple[Message, float]:
|
||||
"""Append the just-completed AIMessage's text + tool_use blocks in order.
|
||||
|
||||
Claude's tool-calling pattern is to emit a single AIMessage per agent
|
||||
turn with mixed content: ``[text "Let me check", tool_use A, text
|
||||
"Now compute", tool_use B]``. The model has already decided the
|
||||
interleaving order. Walk the content list and append each piece to
|
||||
``content_blocks`` chronologically so the renderer can show the
|
||||
narration before each tool call instead of one summary paragraph at
|
||||
the end.
|
||||
|
||||
Tool calls land here with ``output=None``; handle_on_tool_end fills
|
||||
them in once the tool actually returns. handle_on_tool_start sees the
|
||||
pre-populated ToolContent and skips its own append (dedup by name +
|
||||
tool_input + output is None), so providers that don't fire
|
||||
on_chat_model_end (or route tool calls outside .content) still get
|
||||
their ToolContent appended via the fallback.
|
||||
"""
|
||||
output = event["data"].get("output")
|
||||
if not output or not hasattr(output, "content"):
|
||||
return agent_message, start_time
|
||||
|
||||
blocks = _coerce_ai_message_blocks(output.content)
|
||||
if not blocks:
|
||||
return agent_message, start_time
|
||||
|
||||
if agent_message.content_blocks is None:
|
||||
agent_message.content_blocks = []
|
||||
|
||||
duration = _calculate_duration(start_time)
|
||||
appended = False
|
||||
for item in blocks:
|
||||
item_type = item.get("type")
|
||||
if item_type == "text":
|
||||
text = item.get("text") or ""
|
||||
if not text:
|
||||
continue
|
||||
agent_message.content_blocks.append(TextContent(type="text", text=text, duration=duration))
|
||||
appended = True
|
||||
elif item_type == "tool_use":
|
||||
agent_message.content_blocks.append(
|
||||
ToolContent(
|
||||
type="tool_use",
|
||||
name=item.get("name"),
|
||||
tool_input=item.get("input") or {},
|
||||
output=None,
|
||||
error=None,
|
||||
header={"title": f"Accessing **{item.get('name')}**", "icon": "Hammer"},
|
||||
duration=duration,
|
||||
)
|
||||
)
|
||||
appended = True
|
||||
|
||||
if appended:
|
||||
agent_message = await send_message_callback(message=agent_message, skip_db_update=True)
|
||||
return agent_message, perf_counter()
|
||||
|
||||
|
||||
async def handle_on_tool_start(
|
||||
event: dict[str, Any],
|
||||
agent_message: Message,
|
||||
@ -190,19 +243,64 @@ async def handle_on_tool_start(
|
||||
send_message_callback: SendMessageFunctionType,
|
||||
start_time: float,
|
||||
) -> tuple[Message, float]:
|
||||
"""Bind the run_id to the ToolContent the model already emitted.
|
||||
|
||||
handle_on_chat_model_end has typically already appended a ToolContent
|
||||
for this tool_use (with output=None) in the right interleaved
|
||||
position. We just need to find it and key it in tool_blocks_map so
|
||||
handle_on_tool_end can locate it via run_id.
|
||||
|
||||
Match by ``name`` + ``output is None`` + "not yet bound". The
|
||||
tool_use block emitted by on_chat_model_end has an empty
|
||||
``tool_input`` because the model streams the JSON args separately
|
||||
(as ``input_json_delta`` chunks accumulated by the runtime, not
|
||||
captured in our handler's snapshot); the real input only arrives
|
||||
here via on_tool_start. So match by name + unbound + waiting, then
|
||||
overwrite tool_input with what we receive now.
|
||||
|
||||
Fallback: if no matching ToolContent is found (provider routed tool
|
||||
calls outside .content, or on_chat_model_end never fired), append
|
||||
one ourselves so the tool still has a block — order will be best-
|
||||
effort but the call won't be dropped.
|
||||
"""
|
||||
tool_name = event["name"]
|
||||
tool_input = event["data"].get("input")
|
||||
tool_input = event["data"].get("input") or {}
|
||||
run_id = event.get("run_id", "")
|
||||
tool_key = f"{tool_name}_{run_id}"
|
||||
|
||||
# Create content blocks if they don't exist
|
||||
if not agent_message.content_blocks:
|
||||
agent_message.content_blocks = [ContentBlock(title="Agent Steps", contents=[])]
|
||||
if agent_message.content_blocks is None:
|
||||
agent_message.content_blocks = []
|
||||
|
||||
# Look for the ToolContent the model-end handler should have placed.
|
||||
# Skip any that handle_on_tool_start has already bound for an earlier
|
||||
# parallel call to the same tool, so each on_tool_start picks the
|
||||
# next unbound block in declaration order.
|
||||
bound_block_ids = {id(v) for v in tool_blocks_map.values()}
|
||||
existing = None
|
||||
for block in agent_message.content_blocks:
|
||||
if (
|
||||
isinstance(block, ToolContent)
|
||||
and block.name == tool_name
|
||||
and block.output is None
|
||||
and id(block) not in bound_block_ids
|
||||
):
|
||||
existing = block
|
||||
break
|
||||
|
||||
if existing is not None:
|
||||
# Overwrite tool_input with the real, accumulated args (the
|
||||
# model-end snapshot had {} because JSON-delta chunks land
|
||||
# later). But only when on_tool_start actually carries input —
|
||||
# providers that already populated the model-end block's
|
||||
# tool_input (non-streaming Anthropic) fire on_tool_start with an
|
||||
# empty payload, and clobbering with {} would lose the real args.
|
||||
existing.tool_input = tool_input or existing.tool_input
|
||||
tool_blocks_map[tool_key] = existing
|
||||
return agent_message, perf_counter()
|
||||
|
||||
# Fallback path — append the ToolContent ourselves.
|
||||
duration = _calculate_duration(start_time)
|
||||
new_start_time = perf_counter() # Get new start time for next operation
|
||||
|
||||
# Create new tool content with the input exactly as received
|
||||
new_start_time = perf_counter()
|
||||
tool_content = ToolContent(
|
||||
type="tool_use",
|
||||
name=tool_name,
|
||||
@ -210,16 +308,13 @@ async def handle_on_tool_start(
|
||||
output=None,
|
||||
error=None,
|
||||
header={"title": f"Accessing **{tool_name}**", "icon": "Hammer"},
|
||||
duration=duration, # Store the actual duration
|
||||
duration=duration,
|
||||
)
|
||||
|
||||
# Store in map and append to message
|
||||
tool_blocks_map[tool_key] = tool_content
|
||||
agent_message.content_blocks[0].contents.append(tool_content)
|
||||
|
||||
agent_message.content_blocks.append(tool_content)
|
||||
agent_message = await send_message_callback(message=agent_message, skip_db_update=True)
|
||||
if agent_message.content_blocks and agent_message.content_blocks[0].contents:
|
||||
tool_blocks_map[tool_key] = agent_message.content_blocks[0].contents[-1]
|
||||
if agent_message.content_blocks and isinstance(agent_message.content_blocks[-1], ToolContent):
|
||||
tool_blocks_map[tool_key] = agent_message.content_blocks[-1]
|
||||
return agent_message, new_start_time
|
||||
|
||||
|
||||
@ -240,21 +335,21 @@ async def handle_on_tool_end(
|
||||
agent_message = await send_message_callback(message=agent_message, skip_db_update=True)
|
||||
new_start_time = perf_counter()
|
||||
|
||||
# Now find and update the tool content in the current message
|
||||
# Now find and update the tool content in the current message. With
|
||||
# flat content_blocks we walk the list directly instead of indexing
|
||||
# into a single group's .contents.
|
||||
duration = _calculate_duration(start_time)
|
||||
tool_key = f"{tool_name}_{run_id}"
|
||||
|
||||
# Find the corresponding tool content in the updated message
|
||||
updated_tool_content = None
|
||||
if agent_message.content_blocks and agent_message.content_blocks[0].contents:
|
||||
for content in agent_message.content_blocks[0].contents:
|
||||
if (
|
||||
isinstance(content, ToolContent)
|
||||
and content.name == tool_name
|
||||
and content.tool_input == tool_content.tool_input
|
||||
):
|
||||
updated_tool_content = content
|
||||
break
|
||||
for content in agent_message.content_blocks or []:
|
||||
if (
|
||||
isinstance(content, ToolContent)
|
||||
and content.name == tool_name
|
||||
and content.tool_input == tool_content.tool_input
|
||||
and content.output is None
|
||||
):
|
||||
updated_tool_content = content
|
||||
break
|
||||
|
||||
# Update the tool content that's actually in the message
|
||||
if updated_tool_content:
|
||||
@ -304,7 +399,14 @@ async def handle_on_chain_stream(
|
||||
if isinstance(data_chunk, dict) and data_chunk.get("output"):
|
||||
output = data_chunk.get("output")
|
||||
if output and isinstance(output, str | list):
|
||||
agent_message.text = _extract_output_text(output)
|
||||
# Don't use the Message.text setter here. Like handle_on_chain_end,
|
||||
# the setter drops every existing TextContent and appends one at the
|
||||
# end, which collapses the interleaved text + tool_use blocks
|
||||
# on_chat_model_end appended in producer order. ALTK / legacy
|
||||
# AgentExecutor paths reach this branch via on_chain_stream. Stash
|
||||
# the extracted string in data[text_key] so legacy consumers still
|
||||
# read it while content_blocks stays the source of truth.
|
||||
agent_message.data[agent_message.text_key] = _extract_output_text(output) or ""
|
||||
agent_message.properties.state = "complete"
|
||||
# Don't call send_message_callback here - we must update in place
|
||||
# in order to keep the message id consistent throughout the stream.
|
||||
@ -335,7 +437,7 @@ class ToolEventHandler(Protocol):
|
||||
self,
|
||||
event: dict[str, Any],
|
||||
agent_message: Message,
|
||||
tool_blocks_map: dict[str, ContentBlock],
|
||||
tool_blocks_map: dict[str, ToolContent],
|
||||
send_message_callback: SendMessageFunctionType,
|
||||
start_time: float,
|
||||
) -> tuple[Message, float]: ...
|
||||
@ -363,6 +465,11 @@ CHAIN_EVENT_HANDLERS: dict[str, ChainEventHandler] = {
|
||||
"on_chain_end": handle_on_chain_end,
|
||||
"on_chain_stream": handle_on_chain_stream,
|
||||
"on_chat_model_stream": handle_on_chain_stream,
|
||||
# Per-round AIMessage. Fires after each on_chat_model_stream burst
|
||||
# and before the matching on_tool_start. Walks .content for the
|
||||
# interleaved text + tool_use the model emitted and appends them
|
||||
# to content_blocks in producer order.
|
||||
"on_chat_model_end": handle_on_chat_model_end,
|
||||
}
|
||||
|
||||
TOOL_EVENT_HANDLERS: dict[str, ToolEventHandler] = {
|
||||
|
||||
@ -50,7 +50,6 @@ from lfx.inputs.inputs import BoolInput, DropdownInput, ModelInput, StrInput
|
||||
from lfx.io import IntInput, MessageTextInput, MultilineInput, Output, SecretStrInput, TableInput
|
||||
from lfx.log.logger import logger
|
||||
from lfx.memory import delete_message
|
||||
from lfx.schema.content_block import ContentBlock
|
||||
from lfx.schema.data import Data
|
||||
from lfx.schema.dotdict import dotdict
|
||||
from lfx.schema.message import Message
|
||||
@ -684,7 +683,12 @@ class AgentComponent(ToolCallingAgentComponent):
|
||||
sender=MESSAGE_SENDER_AI,
|
||||
sender_name=sender_name,
|
||||
properties={"icon": "Bot", "state": "partial"},
|
||||
content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
|
||||
# `text=""` sentinel so MessageTable's no_content check accepts
|
||||
# an in-flight agent message whose content_blocks haven't been
|
||||
# populated yet. Mirrors ChatInput's convention.
|
||||
text="",
|
||||
# Flat chronological event log; see lfx.base.agents.events.
|
||||
content_blocks=[],
|
||||
session_id=session_id or uuid.uuid4(),
|
||||
)
|
||||
|
||||
|
||||
@ -15,6 +15,7 @@ separate "wrapper" shape; the wrapper *is* a ContentType.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Annotated, Any, Literal
|
||||
|
||||
from fastapi.encoders import jsonable_encoder
|
||||
@ -165,6 +166,24 @@ class ToolContent(BaseContent):
|
||||
error: Any | None = None
|
||||
duration: int | None = None
|
||||
|
||||
@field_validator("tool_input", mode="before")
|
||||
@classmethod
|
||||
def _coerce_tool_input(cls, v: Any) -> dict[str, Any]:
|
||||
# LangChain ``AgentAction.tool_input`` is ``str | dict`` during
|
||||
# streaming, and ``event["data"].get("input") or {}`` passes a
|
||||
# non-empty string straight through. The dict-typed field used to
|
||||
# raise ValidationError on it. Parse JSON objects; wrap any other
|
||||
# string under an ``input`` key so the field always validates.
|
||||
if isinstance(v, str):
|
||||
try:
|
||||
parsed = json.loads(v)
|
||||
except (ValueError, TypeError):
|
||||
return {"input": v}
|
||||
return parsed if isinstance(parsed, dict) else {"input": parsed}
|
||||
if v is None:
|
||||
return {}
|
||||
return v
|
||||
|
||||
|
||||
class _MediaContentMixin:
|
||||
"""Shared validation for media content types (image, audio, video)."""
|
||||
|
||||
@ -153,6 +153,12 @@ class Message(Data):
|
||||
text last.
|
||||
"""
|
||||
if isinstance(value, AsyncIterator | Iterator):
|
||||
# Drop any existing TextContent (and the data["text"] mirror) so the
|
||||
# getter doesn't return stale prior-round text while the stream sits
|
||||
# unconsumed. Non-text blocks keep their position.
|
||||
non_text = [b for b in self.content_blocks if not isinstance(b, TextContent)]
|
||||
object.__setattr__(self, "content_blocks", non_text)
|
||||
self.data[self.text_key] = ""
|
||||
object.__setattr__(self, "_text_stream", value)
|
||||
return
|
||||
# Clear any pending/exhausted stream
|
||||
@ -359,7 +365,7 @@ class Message(Data):
|
||||
sender = lc_message.type
|
||||
sender_name = lc_message.type
|
||||
|
||||
from lfx.schema.content_types import ImageContent
|
||||
from lfx.schema.content_types import ImageContent, ToolContent
|
||||
|
||||
blocks: list[Any] = []
|
||||
content = lc_message.content
|
||||
@ -390,6 +396,20 @@ class Message(Data):
|
||||
blocks.append(ImageContent(base64=b64, mime_type=mime))
|
||||
elif url:
|
||||
blocks.append(ImageContent(urls=[url]))
|
||||
elif item_type == "tool_use":
|
||||
# Anthropic raw content carries tool calls inline as
|
||||
# ``{"type":"tool_use","id","name","input"}``. LangChain
|
||||
# leaves ``.tool_calls`` empty for raw-content messages,
|
||||
# so the tool_calls fallback below won't fire — capture
|
||||
# them here so chat-history round-trips don't drop the
|
||||
# call.
|
||||
blocks.append(
|
||||
ToolContent(
|
||||
name=item.get("name", ""),
|
||||
tool_input=item.get("input", {}),
|
||||
id=item.get("id"),
|
||||
)
|
||||
)
|
||||
else:
|
||||
logger.debug(f"from_lc_message: skipping unsupported content type '{item_type}'")
|
||||
|
||||
@ -397,14 +417,18 @@ class Message(Data):
|
||||
# ``content``. Tool-calling agents typically emit ``content=""`` with
|
||||
# only ``tool_calls`` set, so this must run regardless of content shape.
|
||||
if hasattr(lc_message, "tool_calls") and lc_message.tool_calls:
|
||||
from lfx.schema.content_types import ToolContent
|
||||
|
||||
# The content walk above may have already captured tool_use blocks
|
||||
# (Anthropic raw content). Skip ids already present so a message
|
||||
# carrying both inline tool_use and a populated ``.tool_calls``
|
||||
# doesn't double the same logical call.
|
||||
seen_tool_ids = {b.id for b in blocks if isinstance(b, ToolContent) and b.id}
|
||||
# ``tc["id"]`` is LangChain's stable ``tool_call_id``: same value
|
||||
# at start, during args streaming, and on the result, so the same
|
||||
# logical tool call dedups to one ``ToolContent`` across re-fires.
|
||||
blocks.extend(
|
||||
ToolContent(name=tc.get("name", ""), tool_input=tc.get("args", {}), id=tc.get("id"))
|
||||
for tc in lc_message.tool_calls
|
||||
if tc.get("id") not in seen_tool_ids
|
||||
)
|
||||
|
||||
if hasattr(lc_message, "usage_metadata") and lc_message.usage_metadata:
|
||||
|
||||
82
src/lfx/tests/unit/base/agents/test_events_handlers.py
Normal file
82
src/lfx/tests/unit/base/agents/test_events_handlers.py
Normal file
@ -0,0 +1,82 @@
|
||||
"""Regression tests for the agent event-loop handlers in lfx.base.agents.events.
|
||||
|
||||
These pin two interleaving-preservation behaviors:
|
||||
- handle_on_chain_stream must not run the Message.text setter (which collapses
|
||||
interleaved text + tool_use blocks); it stashes the extracted answer in
|
||||
data["text"] instead.
|
||||
- handle_on_tool_start must not clobber a model-end tool_input snapshot with an
|
||||
empty on_tool_start payload.
|
||||
|
||||
The async ``send_message_callback`` here is a passthrough test harness for the
|
||||
handler's callback boundary, not a behavior mock; the code paths under test
|
||||
return before invoking it.
|
||||
"""
|
||||
|
||||
from time import perf_counter
|
||||
|
||||
from lfx.base.agents.events import handle_on_chain_stream, handle_on_tool_start
|
||||
from lfx.schema.content_types import TextContent, ToolContent
|
||||
from lfx.schema.message import Message
|
||||
|
||||
|
||||
async def _passthrough(*, message: Message, **_kwargs) -> Message:
|
||||
return message
|
||||
|
||||
|
||||
async def test_chain_stream_preserves_interleaved_blocks():
|
||||
"""A chunk.output event must not collapse interleaved content_blocks.
|
||||
|
||||
The Message.text setter drops every TextContent and appends one at the end,
|
||||
which would fuse ``[text, tool, text]`` into ``[tool, text]``. The handler
|
||||
must instead stash the extracted answer in data["text"] and leave
|
||||
content_blocks (the source of truth) untouched.
|
||||
"""
|
||||
msg = Message(
|
||||
content_blocks=[
|
||||
TextContent(text="Let me check"),
|
||||
ToolContent(name="search", tool_input={"q": "x"}),
|
||||
TextContent(text="Now compute"),
|
||||
],
|
||||
sender="Machine",
|
||||
sender_name="AI",
|
||||
)
|
||||
event = {"data": {"chunk": {"output": "Final answer"}}}
|
||||
|
||||
result, _ = await handle_on_chain_stream(event, msg, _passthrough, None, perf_counter())
|
||||
|
||||
block_types = [type(b).__name__ for b in result.content_blocks]
|
||||
assert block_types == ["TextContent", "ToolContent", "TextContent"]
|
||||
# The extracted answer is stashed for legacy consumers, not folded into a
|
||||
# single collapsing TextContent.
|
||||
assert result.data[result.text_key] == "Final answer"
|
||||
|
||||
|
||||
async def test_tool_start_does_not_clobber_existing_tool_input():
|
||||
"""An empty on_tool_start payload must not wipe a real model-end snapshot.
|
||||
|
||||
Providers that already populated the model-end ToolContent.tool_input
|
||||
(non-streaming Anthropic) fire on_tool_start with no input. Overwriting
|
||||
unconditionally would lose the real args.
|
||||
"""
|
||||
existing = ToolContent(name="search", tool_input={"q": "real query"}, output=None)
|
||||
msg = Message(content_blocks=[existing], sender="Machine", sender_name="AI")
|
||||
tool_blocks_map: dict = {}
|
||||
event = {"name": "search", "data": {"input": None}, "run_id": "r1"}
|
||||
|
||||
result, _ = await handle_on_tool_start(event, msg, tool_blocks_map, _passthrough, perf_counter())
|
||||
|
||||
bound = next(b for b in result.content_blocks if isinstance(b, ToolContent))
|
||||
assert bound.tool_input == {"q": "real query"}
|
||||
|
||||
|
||||
async def test_tool_start_overwrites_with_real_input_when_present():
|
||||
"""When on_tool_start carries the real args, they win over the empty model-end snapshot."""
|
||||
existing = ToolContent(name="search", tool_input={}, output=None)
|
||||
msg = Message(content_blocks=[existing], sender="Machine", sender_name="AI")
|
||||
tool_blocks_map: dict = {}
|
||||
event = {"name": "search", "data": {"input": {"q": "streamed"}}, "run_id": "r1"}
|
||||
|
||||
result, _ = await handle_on_tool_start(event, msg, tool_blocks_map, _passthrough, perf_counter())
|
||||
|
||||
bound = next(b for b in result.content_blocks if isinstance(b, ToolContent))
|
||||
assert bound.tool_input == {"q": "streamed"}
|
||||
@ -204,6 +204,21 @@ class TestToolContent:
|
||||
deserialized = ToolContent.model_validate(serialized)
|
||||
assert deserialized == tool
|
||||
|
||||
def test_tool_input_coerces_json_string(self):
|
||||
"""A JSON-object string tool_input parses to a dict instead of raising ValidationError."""
|
||||
tool = ToolContent(name="search", tool_input='{"q": "hi"}')
|
||||
assert tool.tool_input == {"q": "hi"}
|
||||
|
||||
def test_tool_input_wraps_plain_string(self):
|
||||
"""A non-JSON string tool_input is wrapped under an ``input`` key so the field still validates."""
|
||||
tool = ToolContent(name="search", tool_input="weather in SF")
|
||||
assert tool.tool_input == {"input": "weather in SF"}
|
||||
|
||||
def test_tool_input_via_input_alias_coerces(self):
|
||||
"""The same coercion applies when set through the ``input`` alias."""
|
||||
tool = ToolContent.model_validate({"type": "tool_use", "name": "s", "input": "raw"})
|
||||
assert tool.tool_input == {"input": "raw"}
|
||||
|
||||
|
||||
# --- New content type tests ---
|
||||
|
||||
|
||||
@ -269,6 +269,28 @@ class TestTextSetter:
|
||||
# Non-text block stays at its original position
|
||||
assert isinstance(msg.content_blocks[0], ToolContent)
|
||||
|
||||
def test_set_text_to_iterator_drops_stale_text_content(self):
|
||||
"""Setting text to a stream drops existing TextContent.
|
||||
|
||||
Otherwise the getter (which reads content_blocks first) returns the
|
||||
prior round's answer while the stream sits unconsumed. Non-text blocks
|
||||
are preserved and the data mirror is cleared.
|
||||
"""
|
||||
|
||||
def gen():
|
||||
yield "streamed"
|
||||
|
||||
tool_block = ToolContent(name="search", tool_input={"q": "x"})
|
||||
msg = Message(content_blocks=[TextContent(text="old"), tool_block])
|
||||
msg.text = gen()
|
||||
# The stale prior text must not leak through the getter.
|
||||
assert msg.text == ""
|
||||
# Non-text blocks stay.
|
||||
tool_blocks = [b for b in msg.content_blocks if isinstance(b, ToolContent)]
|
||||
assert len(tool_blocks) == 1
|
||||
# The stream is stashed for later consumption.
|
||||
assert msg.text_stream is not None
|
||||
|
||||
|
||||
class TestSerialization:
|
||||
"""Tests for model_dump / model_validate round-trip behavior."""
|
||||
@ -490,6 +512,45 @@ class TestFromLcMessageToolCallId:
|
||||
assert [b.text for b in text_blocks] == ["I'll search for that."]
|
||||
assert [b.id for b in tool_blocks] == ["call_abc"]
|
||||
|
||||
def test_raw_tool_use_block_in_content_is_captured(self):
|
||||
"""Capture an inline Anthropic ``tool_use`` content block.
|
||||
|
||||
LangChain leaves ``.tool_calls`` empty for raw-content messages, so the
|
||||
content walk must capture it or the call is dropped on round-trips.
|
||||
"""
|
||||
lc_msg = AIMessage(
|
||||
content=[
|
||||
{"type": "text", "text": "Let me check"},
|
||||
{"type": "tool_use", "id": "tu_1", "name": "search", "input": {"q": "x"}},
|
||||
],
|
||||
)
|
||||
# Raw-content AIMessages leave tool_calls empty — the fallback below
|
||||
# the content walk won't fire, so the content walk must capture it.
|
||||
assert lc_msg.tool_calls == []
|
||||
msg = Message.from_lc_message(lc_msg)
|
||||
tool_blocks = [b for b in msg.content_blocks if isinstance(b, ToolContent)]
|
||||
assert len(tool_blocks) == 1
|
||||
assert tool_blocks[0].name == "search"
|
||||
assert tool_blocks[0].id == "tu_1"
|
||||
assert tool_blocks[0].tool_input == {"q": "x"}
|
||||
|
||||
def test_tool_use_in_content_and_tool_calls_not_doubled(self):
|
||||
"""Inline tool_use plus a matching ``.tool_calls`` entry yield one block.
|
||||
|
||||
The same logical call (same id) must not be doubled across the content
|
||||
walk and the tool_calls fallback.
|
||||
"""
|
||||
lc_msg = AIMessage(
|
||||
content=[
|
||||
{"type": "tool_use", "id": "tu_1", "name": "search", "input": {"q": "x"}},
|
||||
],
|
||||
tool_calls=[{"name": "search", "args": {"q": "x"}, "id": "tu_1", "type": "tool_call"}],
|
||||
)
|
||||
msg = Message.from_lc_message(lc_msg)
|
||||
tool_blocks = [b for b in msg.content_blocks if isinstance(b, ToolContent)]
|
||||
assert len(tool_blocks) == 1
|
||||
assert tool_blocks[0].id == "tu_1"
|
||||
|
||||
|
||||
class TestMessageResponseFromMessage:
|
||||
"""Regression tests for ``MessageResponse.from_message`` timestamp parity.
|
||||
|
||||
Reference in New Issue
Block a user