mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(goal): retain supplied evidence and retry one malformed evaluator verdict (#7345)
* fix(goal): use spare evidence budget for long user input * fix(goal): retry one malformed evaluator verdict --------- Co-authored-by: Totoro-qaq <279883115+Totoro-qaq@users.noreply.github.com>
This commit is contained in:
Totoro
GitHub
parent
4e20cbb4d6
commit
fc4f719664
@ -2268,6 +2268,9 @@ Supported commands:
|
||||
|
||||
After each Gateway-backed run, DeerFlow evaluates the visible conversation, including the assistant's tool calls and shortened tool results, against the active goal with a non-thinking evaluator model. A successful tool result alone does not satisfy a goal, and when the assistant had to guess missing or ambiguous information the evaluator reports `needs_user_input`. The evaluator must return a typed blocker (`missing_evidence`, `needs_user_input`, `run_failed`, `external_wait`, or `goal_not_met_yet`) plus visible evidence. DeerFlow only injects a hidden continuation when the latest assistant turn is durably checkpointed, the blocker is `goal_not_met_yet`, the thread did not change during evaluation, and the no-progress breaker has not fired. The safety cap defaults to 8 hidden continuations, and repeated identical non-progress evaluations stop after 2 attempts. `/goal clear` and any user-authored new input win over queued continuations. When the evaluator considers the goal satisfied, DeerFlow clears it after the run successfully finalizes and publishes the updated thread state. The same checkpoint records the completion as `goal_outcome`, which stays until the next goal change. If required artifact delivery or its receipt persistence fails, the run reports an error and the goal remains active for a later retry; this does not start another hidden continuation.
|
||||
|
||||
The evaluator keeps recent work and the latest user request within a fixed evidence budget, using spare room to retain more of long user input.
|
||||
If a normally completed evaluator response has an invalid verdict format, DeerFlow evaluates the same evidence once more unless you cancel or change the thread. Both calls are counted; a second failure stops automatic continuation without repeating the task or its tools.
|
||||
|
||||
For tool-using turns, successful operation results drive the no-progress count: changing only reply wording or a sandbox tool's UI explanation does not reset it, while newly observed work can keep it going. Text-only turns keep their reply-based check. An unmet goal still stops automatic continuation after a token or loop-detection hard stop and stays active for a later user turn.
|
||||
|
||||
Regenerating an earlier response keeps the goal currently set for the chat and its used continuation budget. It does not restore a goal that was cleared or achieved. The earlier answer's completion record is removed when its historical checkpoint is replayed.
|
||||
|
||||
@ -363,6 +363,16 @@ so recovery need not inspect mutable thread state. All non-interactive policies
|
||||
allow disclosed, low-risk reversible assumptions only with achievement evidence;
|
||||
their evaluator responses must explicitly provide the assumption boolean.
|
||||
|
||||
Goal evidence stays within 12,000 characters. Keep the selected recent work and
|
||||
omission labels intact; use remaining space to restore shortened user input
|
||||
rather than treating the 2,000-character request reservation as a hard cap.
|
||||
|
||||
Only a malformed verdict from an explicitly normal model completion may retry,
|
||||
once, with the original evidence. Recheck cancellation, Goal identity and
|
||||
checkpoint/messages first. Each complete evaluator invocation keeps its own usage
|
||||
ID and observer call; one final decision updates Goal counters. Valid verdicts,
|
||||
length/safety stops, and transport/configuration errors are not format retries.
|
||||
|
||||
Goal evaluator usage crosses a JSON-safe sink into
|
||||
`RunJournal.record_external_llm_usage_records` before response parsing or observer
|
||||
notification can fail. Critic tokens/cache reads and distinct calls belong to
|
||||
|
||||
@ -76,6 +76,10 @@ class GoalWriteConflict(RuntimeError):
|
||||
"""Raised when a goal write is based on a stale checkpoint."""
|
||||
|
||||
|
||||
class GoalEvaluationFormatError(ValueError):
|
||||
"""A normally completed evaluator response did not contain a valid verdict."""
|
||||
|
||||
|
||||
def goal_thread_lock(thread_id: str) -> AbstractAsyncContextManager[None]:
|
||||
"""Serialize goal read-modify-write sequences within the current event loop."""
|
||||
return _goal_locks.hold(thread_id)
|
||||
@ -306,7 +310,8 @@ def _cap_evidence(lines: list[str]) -> str:
|
||||
that work answers, and so is the latest Human Input Card answer left out. Each is shortened
|
||||
only when it and the lines after it do not fit whole. If the next assistant message back
|
||||
does not fit whole, its end fills the room left. A marker says how many lines were left
|
||||
out. No line starts midway without a label.
|
||||
out. Spare room then restores shortened user input without displacing the selected
|
||||
recent work. No line starts midway without a label.
|
||||
"""
|
||||
conversation = "\n\n".join(lines)
|
||||
if len(conversation) <= MAX_GOAL_CONVERSATION_CHARS:
|
||||
@ -345,6 +350,13 @@ def _cap_evidence(lines: list[str]) -> str:
|
||||
start = boundary
|
||||
omitted = start - len(head)
|
||||
marker = [f"[{omitted} earlier evidence lines omitted]"] if omitted else []
|
||||
spare = MAX_GOAL_CONVERSATION_CHARS - len("\n\n".join([*head, *marker, *kept]))
|
||||
for offset, index in enumerate(picks if head else []):
|
||||
if spare <= 0:
|
||||
break
|
||||
expanded = _truncate(lines[index], MAX_GOAL_REQUEST_CHARS + spare)
|
||||
spare += len(head[offset]) - len(expanded)
|
||||
head[offset] = expanded
|
||||
return "\n\n".join([*head, *marker, *kept])
|
||||
|
||||
|
||||
@ -547,7 +559,20 @@ async def evaluate_goal_completion(
|
||||
invoke=invoke_with_usage,
|
||||
task_store=task_store,
|
||||
)
|
||||
return parse_goal_evaluation_response(_extract_response_text(response.content), require_assumption_attribution=interaction_policy is not None and not interaction_policy.allows_clarification)
|
||||
try:
|
||||
return parse_goal_evaluation_response(_extract_response_text(response.content), require_assumption_attribution=interaction_policy is not None and not interaction_policy.allows_clarification)
|
||||
except ValueError as exc:
|
||||
metadata = getattr(response, "response_metadata", None)
|
||||
reasons = [metadata[key] for key in ("finish_reason", "stop_reason") if key in metadata] if isinstance(metadata, Mapping) else []
|
||||
blocks = response.content if isinstance(response.content, list) else []
|
||||
refusal = _additional_kwargs(response).get("refusal") or any(
|
||||
isinstance(block, Mapping) and (block.get("type") == "refusal" or (block.get("type") == "non_standard" and isinstance(block.get("value"), Mapping) and block["value"].get("refusal"))) for block in blocks
|
||||
)
|
||||
# Explicit refusal wins even when an adapter also reports normal stop.
|
||||
# Length/safety stops and unknown outcomes retain their failure behavior.
|
||||
if not refusal and reasons and all(isinstance(reason, str) and reason.casefold() in {"stop", "end_turn"} for reason in reasons):
|
||||
raise GoalEvaluationFormatError("The completed goal evaluator response was malformed.") from exc
|
||||
raise
|
||||
|
||||
|
||||
def _goal_evaluator_usage_record(response: Any, *, source_id: str, model_name: str | None, model: Any) -> dict[str, int | str | None]:
|
||||
|
||||
@ -71,6 +71,7 @@ from deerflow.runtime.events.message_identity import attach_message_seq, message
|
||||
from deerflow.runtime.goal import (
|
||||
DEFAULT_MAX_GOAL_CONTINUATIONS,
|
||||
DEFAULT_MAX_NO_PROGRESS_CONTINUATIONS,
|
||||
GoalEvaluationFormatError,
|
||||
GoalWriteConflict,
|
||||
_call_checkpointer_method,
|
||||
_is_visible_message,
|
||||
@ -2660,20 +2661,36 @@ async def _prepare_goal_continuation_input(
|
||||
|
||||
try:
|
||||
evaluator_model = evaluator_model_factory() if evaluator_model_factory is not None else None
|
||||
evaluation = await evaluate_goal_completion(
|
||||
goal,
|
||||
messages,
|
||||
model=evaluator_model,
|
||||
model_name=model_name,
|
||||
app_config=app_config,
|
||||
thread_id=thread_id,
|
||||
user_id=user_id,
|
||||
deerflow_trace_id=deerflow_trace_id,
|
||||
task_store=task_store,
|
||||
extensions=extensions,
|
||||
interaction_policy=interaction_policy,
|
||||
usage_callback=usage_callback,
|
||||
)
|
||||
for attempt in range(2):
|
||||
try:
|
||||
evaluation = await evaluate_goal_completion(
|
||||
goal,
|
||||
messages,
|
||||
model=evaluator_model,
|
||||
model_name=model_name,
|
||||
app_config=app_config,
|
||||
thread_id=thread_id,
|
||||
user_id=user_id,
|
||||
deerflow_trace_id=deerflow_trace_id,
|
||||
task_store=task_store,
|
||||
extensions=extensions,
|
||||
interaction_policy=interaction_policy,
|
||||
usage_callback=usage_callback,
|
||||
)
|
||||
break
|
||||
except GoalEvaluationFormatError:
|
||||
if attempt:
|
||||
raise
|
||||
if abort_event is not None and abort_event.is_set():
|
||||
return None
|
||||
current_goal, current_checkpoint = await _reread_goal_and_checkpoint(checkpointer, thread_id)
|
||||
if not _goal_instance_matches(goal, current_goal) or current_checkpoint is None:
|
||||
return None
|
||||
if _checkpoint_id(current_checkpoint) != checkpoint_id_before or visible_conversation_signature(await _materialized_checkpoint_messages(accessor, thread_id)) != conversation_signature_before:
|
||||
return None
|
||||
if abort_event is not None and abort_event.is_set():
|
||||
return None
|
||||
logger.info("Retrying one malformed goal evaluation for thread %s after run %s", thread_id, run_id)
|
||||
except Exception as exc:
|
||||
logger.warning("Goal evaluator failed for thread %s after run %s", thread_id, run_id, exc_info=True)
|
||||
if abort_event is not None and abort_event.is_set():
|
||||
|
||||
@ -30,13 +30,19 @@ class ScriptedModel(BaseChatModel):
|
||||
answer: str
|
||||
identifier: str
|
||||
reported_usage: dict | None = None
|
||||
answers: list[str] | None = None
|
||||
normal_completion: bool = False
|
||||
|
||||
@property
|
||||
def _llm_type(self):
|
||||
return "scripted-accounting"
|
||||
|
||||
def _generate(self, messages, stop=None, run_manager=None, **kwargs):
|
||||
response = AIMessage(content=self.answer, usage_metadata=self.reported_usage, response_metadata={"model_name": self.identifier})
|
||||
answer = self.answers.pop(0) if self.answers else self.answer
|
||||
metadata = {"model_name": self.identifier}
|
||||
if self.normal_completion:
|
||||
metadata["finish_reason"] = "stop"
|
||||
response = AIMessage(content=answer, usage_metadata=self.reported_usage, response_metadata=metadata)
|
||||
return ChatResult(generations=[ChatGeneration(message=response)])
|
||||
|
||||
|
||||
@ -55,7 +61,7 @@ async def run_store(request):
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("response", ["valid", "invalid_json", "missing_usage"])
|
||||
@pytest.mark.parametrize("response", ["valid", "invalid_json", "missing_usage", "recovered_json", "repeated_invalid_json", "unsatisfied_json"])
|
||||
async def test_worker_terminal_ledger_includes_evaluator_usage_without_visible_critic(run_store, monkeypatch, response):
|
||||
events = MemoryRunEventStore()
|
||||
checkpointer = InMemorySaver()
|
||||
@ -63,8 +69,12 @@ async def test_worker_terminal_ledger_includes_evaluator_usage_without_visible_c
|
||||
metadata = {"scheduled_task_id": "task", "scheduled_task_run_id": "occurrence", "scheduled_goal_objective": "Report"}
|
||||
record = await manager.create_or_reject("result", user_id="alice", metadata=metadata)
|
||||
lead = ScriptedModel(answer="The report is verified.", identifier="lead-model", reported_usage={"input_tokens": 30, "output_tokens": 10, "total_tokens": 40})
|
||||
verdict = json.dumps({"satisfied": response != "unsatisfied_json", "blocker": "missing_evidence", "reason": "PRIVATE_CRITIC_VERDICT", "relied_on_assumption": False})
|
||||
critic_calls = 2 if response in {"recovered_json", "repeated_invalid_json"} else 1
|
||||
critic = ScriptedModel(
|
||||
answer="PRIVATE_CRITIC_INVALID" if response == "invalid_json" else json.dumps({"satisfied": True, "reason": "PRIVATE_CRITIC_VERDICT", "relied_on_assumption": False}),
|
||||
answer="PRIVATE_CRITIC_INVALID" if response in {"invalid_json", "repeated_invalid_json"} else verdict,
|
||||
answers=["PRIVATE_CRITIC_INVALID", verdict] if response == "recovered_json" else None,
|
||||
normal_completion=response in {"recovered_json", "repeated_invalid_json", "unsatisfied_json"},
|
||||
identifier="critic-model",
|
||||
reported_usage=None if response == "missing_usage" else {"input_tokens": 80, "output_tokens": 20, "total_tokens": 100, "input_token_details": {"cache_read": 32}},
|
||||
)
|
||||
@ -90,7 +100,7 @@ async def test_worker_terminal_ledger_includes_evaluator_usage_without_visible_c
|
||||
config={"configurable": {"thread_id": "result"}, "context": {"non_interactive": True}},
|
||||
)
|
||||
row = await run_store.get(record.run_id, user_id="alice")
|
||||
assert row["llm_call_count"] == 2
|
||||
assert row["llm_call_count"] == 1 + critic_calls
|
||||
assert row["lead_agent_tokens"] == 40
|
||||
pricing = {"lead-model": _ModelPricing(2, 4, "USD"), "critic-model": _ModelPricing(2, 4, "USD", 0.5)}
|
||||
cost = _run_cost(pricing, model_name="lead-model", total_input_tokens=row["total_input_tokens"], total_output_tokens=row["total_output_tokens"], token_usage_by_model=row["token_usage_by_model"])
|
||||
@ -98,12 +108,15 @@ async def test_worker_terminal_ledger_includes_evaluator_usage_without_visible_c
|
||||
assert row["token_usage_by_model"]["critic-model"]["missing_usage_calls"] == 1
|
||||
assert cost is None
|
||||
else:
|
||||
assert row["middleware_tokens"] == 100
|
||||
assert row["total_input_tokens"] == 110
|
||||
assert row["total_output_tokens"] == 30
|
||||
assert row["total_tokens"] == 140
|
||||
assert row["token_usage_by_model"]["critic-model"]["cache_read_tokens"] == 32
|
||||
assert cost == pytest.approx(0.000292)
|
||||
assert row["middleware_tokens"] == 100 * critic_calls
|
||||
assert row["total_input_tokens"] == 30 + 80 * critic_calls
|
||||
assert row["total_output_tokens"] == 10 + 20 * critic_calls
|
||||
assert row["total_tokens"] == 40 + 100 * critic_calls
|
||||
assert row["token_usage_by_model"]["critic-model"]["cache_read_tokens"] == 32 * critic_calls
|
||||
assert cost == pytest.approx(0.0001 + 0.000192 * critic_calls)
|
||||
if response in {"recovered_json", "repeated_invalid_json", "unsatisfied_json"}:
|
||||
assert row["goal_verdict"]["satisfied"] is (response == "recovered_json")
|
||||
assert row["goal_verdict"]["continuations"] == 0
|
||||
history = await events.list_messages("result", user_id="alice")
|
||||
assert "PRIVATE_CRITIC" not in json.dumps(history)
|
||||
|
||||
|
||||
@ -366,12 +366,29 @@ def test_format_visible_conversation_keeps_a_long_request_whole_when_it_fits():
|
||||
assert lines[1] == f"User: {request}"
|
||||
|
||||
|
||||
def test_format_visible_conversation_uses_spare_budget_for_long_user_input_at_the_cap():
|
||||
request = "Review the supplied data. " + "x" * 5000 + " IMPORTANT SOURCE ROW: 314159 "
|
||||
request += "y" * (10000 - len(request))
|
||||
answer = "The result is " + "a" * (goal.MAX_GOAL_CONVERSATION_CHARS - len(f"User: {request}\n\nAssistant: The result is done.")) + " done."
|
||||
assert len(f"User: {request}\n\nAssistant: {answer}") == goal.MAX_GOAL_CONVERSATION_CHARS
|
||||
|
||||
for text in (answer, answer.replace("The result", "The result", 1)):
|
||||
formatted = goal.format_visible_conversation([HumanMessage(content=request), AIMessage(content=text)])
|
||||
lines = _evidence_lines(formatted)
|
||||
# Crossing the cap by one character must not discard most of the input
|
||||
# while leaving room unused. Preserve the selected recent answer too.
|
||||
assert "IMPORTANT SOURCE ROW: 314159" in lines[0]
|
||||
assert len(lines[0]) >= 9900
|
||||
assert lines[-1] == f"Assistant: {text}"
|
||||
assert len(formatted) >= goal.MAX_GOAL_CONVERSATION_CHARS - 64
|
||||
|
||||
|
||||
def test_format_visible_conversation_shortens_a_request_that_does_not_fit():
|
||||
request = "Build the table. " + "r" * 5000
|
||||
lines = _evidence_lines(goal.format_visible_conversation(_report_run(request)))
|
||||
|
||||
assert lines[0].startswith("User: Build the table. ") and lines[0].endswith(" more chars]")
|
||||
assert len(lines[0]) < goal.MAX_GOAL_REQUEST_CHARS + 40
|
||||
assert len(lines[0]) < len(f"User: {request}")
|
||||
|
||||
|
||||
def test_format_visible_conversation_marks_the_cut_when_no_request_is_in_the_window():
|
||||
@ -464,6 +481,28 @@ def test_make_goal_continuation_message_is_hidden_from_ui():
|
||||
assert "Tests have not run" in message.content
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
("metadata", "answer", "extra", "retryable"),
|
||||
[
|
||||
({"finish_reason": "stop"}, '{"satisfied":', {}, True),
|
||||
({"stop_reason": "end_turn"}, '{"satisfied": "yes"}', {}, True),
|
||||
({"status": "completed", "finish_reason": "length"}, '{"satisfied":', {}, False),
|
||||
({"finish_reason": "content_filter"}, "refused", {}, False),
|
||||
({"stop_reason": "refusal"}, "refused", {}, False),
|
||||
({}, '{"satisfied":', {}, False),
|
||||
({"finish_reason": "stop"}, "", {"refusal": "Not allowed"}, False),
|
||||
({"finish_reason": "stop"}, [{"type": "refusal", "refusal": "Not allowed"}], {}, False),
|
||||
],
|
||||
)
|
||||
async def test_evaluator_format_retry_requires_a_normal_completion(metadata, answer, extra, retryable):
|
||||
model = SimpleNamespace(ainvoke=AsyncMock(return_value=AIMessage(content=answer, response_metadata=metadata, additional_kwargs=extra)))
|
||||
with pytest.raises(ValueError) as error:
|
||||
await goal.evaluate_goal_completion(goal.build_goal_state("Report"), [HumanMessage(content="Report"), AIMessage(content="Done")], model=model)
|
||||
assert isinstance(error.value, goal.GoalEvaluationFormatError) is retryable
|
||||
model.ainvoke.assert_awaited_once()
|
||||
|
||||
|
||||
def test_evaluate_goal_completion_uses_non_thinking_model(monkeypatch):
|
||||
fake_model = MagicMock()
|
||||
fake_model.ainvoke = AsyncMock(return_value=SimpleNamespace(content='{"satisfied": true, "reason": "Done", "evidence_summary": "Done"}'))
|
||||
|
||||
@ -802,7 +802,8 @@ async def test_goal_worker_stands_down_without_durable_assistant_receipt():
|
||||
class _FailingEvaluatorModel:
|
||||
"""Evaluator chat model whose provider call raises, or whose answer is not valid JSON."""
|
||||
|
||||
def __init__(self, outcome: Exception | str, before=None) -> None:
|
||||
def __init__(self, outcome: Exception | str, before=None, *, normal_completion=False) -> None:
|
||||
self._normal_completion = normal_completion
|
||||
self._outcome = outcome
|
||||
self._before = before
|
||||
self.calls = 0
|
||||
@ -813,7 +814,7 @@ class _FailingEvaluatorModel:
|
||||
await self._before()
|
||||
if isinstance(self._outcome, Exception):
|
||||
raise self._outcome
|
||||
return AIMessage(content=self._outcome)
|
||||
return AIMessage(content=self._outcome, response_metadata={"finish_reason": "stop"} if self._normal_completion else {})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@ -864,6 +865,46 @@ async def test_goal_worker_records_a_failed_evaluator_call(outcome, error_type,
|
||||
assert "request-detail-7f3a" not in repr(bridge.events)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("user_change", ["abort", "clear", "replace", "message"])
|
||||
async def test_goal_format_retry_does_not_outlive_a_user_change(user_change):
|
||||
saver = InMemorySaver()
|
||||
thread_id = "format-retry-user-change"
|
||||
await _seed_goal_thread(saver, thread_id=thread_id, goal_text="Finish the report")
|
||||
abort_event = asyncio.Event()
|
||||
|
||||
async def change():
|
||||
if user_change == "abort":
|
||||
abort_event.set()
|
||||
elif user_change == "clear":
|
||||
await write_thread_goal(saver, thread_id, None)
|
||||
elif user_change == "replace":
|
||||
await write_thread_goal(saver, thread_id, build_goal_state("A different request"))
|
||||
else:
|
||||
await _write_messages(saver, thread_id=thread_id, messages=[HumanMessage(content="Actually, wait for me.")])
|
||||
|
||||
critic = _FailingEvaluatorModel("{PRIVATE_CRITIC", before=change, normal_completion=True)
|
||||
result = await worker._prepare_goal_continuation_input(
|
||||
accessor=_full_accessor(saver),
|
||||
bridge=_CollectingBridge(),
|
||||
checkpointer=saver,
|
||||
thread_id=thread_id,
|
||||
run_id="format-retry",
|
||||
model_name=None,
|
||||
app_config=None,
|
||||
evaluator_model_factory=lambda: critic,
|
||||
abort_event=abort_event,
|
||||
)
|
||||
assert result is None
|
||||
assert critic.calls == 1
|
||||
current_goal = await read_thread_goal(saver, thread_id)
|
||||
if user_change == "clear":
|
||||
assert current_goal is None
|
||||
else:
|
||||
assert current_goal["continuation_count"] == current_goal["no_progress_count"] == 0
|
||||
assert "last_evaluation" not in current_goal
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_goal_worker_does_not_record_a_failed_evaluator_call_after_abort():
|
||||
checkpointer = InMemorySaver()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user