mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(runtime): record textless human input only once per run (#5903)
Co-authored-by: kwsYegar <331533331+kwsYegar@users.noreply.github.com>
This commit is contained in:
parent
1dfd307c92
commit
b433e3c6a0
@ -1858,6 +1858,8 @@ request the binary capability retain the legacy JSON/base64 frame protocol.
|
||||
|
||||
**Visible Tool-Run Completion**: For interactive turns, DeerFlow retries an empty post-tool final response once, then surfaces a visible error instead of reporting a silent successful run.
|
||||
|
||||
**Run History**: Image-only user input is recorded once per run, even when the agent makes multiple model calls. Its media content remains in history without requiring a text summary.
|
||||
|
||||
### Reading a Referenced Conversation
|
||||
|
||||
Gateway API callers can opt into `read_conversation` and submit a
|
||||
|
||||
@ -26,6 +26,11 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c
|
||||
|
||||
**Message sequence placement:** Keep backend and frontend message identity rules aligned. Details: `backend/docs/runtime-guidance-details.md`.
|
||||
|
||||
**Human-input capture** (`runtime/journal.py`): track capture separately from
|
||||
the optional display summary. Image-only input has no text but must still stop
|
||||
the batch scan and later model calls from appending another human-input event.
|
||||
`tests/test_run_journal.py` covers callback and full/delta graph paths.
|
||||
|
||||
**LLM response callback coalescing** (`runtime/journal.py`): a provider may fire
|
||||
`on_llm_end` twice for one LangChain run id, first without usage (or with all token
|
||||
counts zero) and immediately again with usage populated. The first callback's generation
|
||||
|
||||
@ -299,6 +299,7 @@ class RunJournal(BaseCallbackHandler):
|
||||
# Convenience fields
|
||||
self._last_ai_msg: str | None = None
|
||||
self._first_human_msg: str | None = None
|
||||
self._first_human_message_captured = False
|
||||
self._msg_count = 0
|
||||
self._had_llm_error_fallback = False
|
||||
self._llm_error_fallback_message: str | None = None
|
||||
@ -437,7 +438,7 @@ class RunJournal(BaseCallbackHandler):
|
||||
if caller == "lead_agent":
|
||||
for batch in messages:
|
||||
self._reconcile_consumed_tool_messages(batch)
|
||||
if caller == "lead_agent" and not self._first_human_msg and messages:
|
||||
if caller == "lead_agent" and not self._first_human_message_captured and messages:
|
||||
for batch in reversed(messages):
|
||||
for m in reversed(batch):
|
||||
if _should_persist_human_input_message(m):
|
||||
@ -451,7 +452,7 @@ class RunJournal(BaseCallbackHandler):
|
||||
)
|
||||
self._record_message_summary(persisted_message, caller=caller)
|
||||
break
|
||||
if self._first_human_msg:
|
||||
if self._first_human_message_captured:
|
||||
break
|
||||
|
||||
def on_llm_start(self, serialized: dict, prompts: list[str], *, run_id: UUID, parent_run_id: UUID | None = None, tags: list[str] | None = None, metadata: dict[str, Any] | None = None, **kwargs: Any) -> None:
|
||||
@ -1065,6 +1066,8 @@ class RunJournal(BaseCallbackHandler):
|
||||
|
||||
def set_first_human_message(self, content: str) -> None:
|
||||
"""Record the first human message for convenience fields."""
|
||||
# Media-only input is still captured even when it has no display text.
|
||||
self._first_human_message_captured = True
|
||||
self._first_human_msg = content[:2000] if content else None
|
||||
|
||||
def record_middleware(self, tag: str, *, name: str, hook: str, action: str, changes: dict) -> None:
|
||||
|
||||
@ -2217,6 +2217,47 @@ class TestChatModelStartHumanMessage:
|
||||
human_events = [e for e in events if e["event_type"] == "llm.human.input"]
|
||||
assert len(human_events) == 1
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.parametrize(
|
||||
"content",
|
||||
[
|
||||
pytest.param("", id="empty-text"),
|
||||
pytest.param([{"type": "image_url", "image_url": {"url": "data:image/png;base64,aW1hZ2U="}}], id="image-only"),
|
||||
],
|
||||
)
|
||||
async def test_textless_human_input_is_captured_once(self, journal_setup, content):
|
||||
"""A missing display summary must not duplicate input on later model calls."""
|
||||
j, store = journal_setup
|
||||
message = HumanMessage(content=content, id="user-image")
|
||||
for _ in range(2):
|
||||
j.on_chat_model_start({}, [[message]], run_id=uuid4(), tags=["lead_agent"])
|
||||
await j.flush()
|
||||
|
||||
events = await store.list_events("t1", "r1")
|
||||
human_events = [event for event in events if event["event_type"] == "llm.human.input"]
|
||||
assert len(human_events) == 1
|
||||
assert human_events[0]["content"]["content"] == content
|
||||
assert j.get_completion_data()["message_count"] == 1
|
||||
assert j.get_completion_data()["first_human_message"] is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_textless_input_stops_capture_before_older_batches(self, journal_setup):
|
||||
j, store = journal_setup
|
||||
image = HumanMessage(content=[{"type": "image_url", "image_url": {"url": "data:image/png;base64,aW1hZ2U="}}], id="latest-input")
|
||||
j.on_chat_model_start(
|
||||
{},
|
||||
[[HumanMessage(content="Older question", id="old-input")], [image]],
|
||||
run_id=uuid4(),
|
||||
tags=["lead_agent"],
|
||||
)
|
||||
await j.flush()
|
||||
|
||||
events = await store.list_events("t1", "r1")
|
||||
human_events = [event for event in events if event["event_type"] == "llm.human.input"]
|
||||
assert [event["content"]["id"] for event in human_events] == ["latest-input"]
|
||||
assert j.get_completion_data()["first_human_message"] is None
|
||||
assert j.get_completion_data()["message_count"] == 1
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_empty_messages_no_crash(self, journal_setup):
|
||||
"""on_chat_model_start with empty messages does not crash."""
|
||||
@ -2226,6 +2267,46 @@ class TestChatModelStartHumanMessage:
|
||||
assert j._first_human_msg is None
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
async def test_image_input_is_journaled_once_across_graph_model_calls(mode):
|
||||
from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
from langgraph.graph import StateGraph
|
||||
|
||||
from deerflow.agents.thread_state import get_thread_state_schema
|
||||
from deerflow.runtime.checkpoint_state import CheckpointStateAccessor
|
||||
|
||||
store = MemoryRunEventStore()
|
||||
journal = RunJournal("image-run", "image-thread", store)
|
||||
model = FakeMessagesListChatModel(responses=[AIMessage(content="First response", id="a1"), AIMessage(content="Final response", id="a2")])
|
||||
|
||||
async def answer(state):
|
||||
return {"messages": [await model.ainvoke(state["messages"])]}
|
||||
|
||||
builder = StateGraph(get_thread_state_schema(mode))
|
||||
builder.add_node("first", answer)
|
||||
builder.add_node("second", answer)
|
||||
builder.set_entry_point("first")
|
||||
builder.add_edge("first", "second")
|
||||
builder.set_finish_point("second")
|
||||
saver = InMemorySaver()
|
||||
graph = builder.compile(checkpointer=saver)
|
||||
accessor = CheckpointStateAccessor.bind(graph, saver, mode=mode)
|
||||
image_content = [{"type": "image_url", "image_url": {"url": "data:image/png;base64,aW1hZ2U="}}]
|
||||
config = {"configurable": {"thread_id": "image-thread"}, "callbacks": [journal]}
|
||||
|
||||
await graph.ainvoke({"messages": [HumanMessage(content=image_content, id="h1")]}, config)
|
||||
await journal.flush()
|
||||
events = await store.list_messages("image-thread")
|
||||
assert [event["content"]["id"] for event in events] == ["h1", "a1", "a2"]
|
||||
assert events[0]["content"]["content"] == image_content
|
||||
assert journal.get_completion_data()["message_count"] == 3
|
||||
snapshot = await accessor.aget(config)
|
||||
assert [message.id for message in snapshot.values["messages"]] == ["h1", "a1", "a2"]
|
||||
await journal.close()
|
||||
|
||||
|
||||
class TestDeliveryTracking:
|
||||
"""Slice 1 (#4272): journal records artifact production for run.delivery."""
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user