fix(runtime): omit leading reasoning from run summaries (#6419)

This commit is contained in:
Daoyuan Li 2026-10-07 02:20:11 -07:00 committed by GitHub
parent 787b44c74f
commit 734f9e0e92
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 182 additions and 11 deletions

View File

@ -2485,6 +2485,8 @@ Current MVP capabilities:
- When `channel_connections.enabled: true`, send scheduled task updates to the task owner's connected IM identities (outbox + delivery worker) on apps that support proactive push, which today is WeCom; Settings shows for each app whether updates are sent there, and other apps get none. Each occurrence sends at most one message: a finished, failed or goal-missed run, the automatic pause after three missed goals, a pause by the agent (its stop condition was met) or the task finishing (all `max_runs` done, `end_at` reached); when several apply, the pause or finish wins and still says how the last run went. A one-time task sends only its run's outcome. The message is queued in the same database transaction that records the outcome, so runs finalized after a crash or a lost lease notify exactly once. It reads on its own, in your web UI language (else `channel_connections.notification_locale`): the task title, what happened, a one-line result when the agent replied, and "Open DeerFlow → Scheduled tasks for details.", with no IDs and no links. Plain manual "run now" trials and interrupted runs stay silent, and so do occurrences that end without a finished run (launch error, queue timeout, interrupted by a restart). Channel/transport outages park deliveries without exhausting retries, for up to about a day; platform rejections retry for roughly 15 minutes before settling as `failed`. An identity you disconnect while a delivery is waiting is never pushed to: the row is dropped as `failed`.
- Browse execution history in pages of 50; older pages pause automatic refresh, with an explicit return to the latest runs. Counts appear only after a successful read; loading and failed reads are not reported as zero runs.
Derived run summaries omit leading inline `<think>` reasoning before limiting the text, so a long thought does not replace the actual answer in previews or scheduled notices. Original response history and literal tags in the answer are unchanged; previously stored run summaries are not rewritten.
**Filter execution history through the API**
To inspect failures without downloading every successful occurrence, authenticated clients with `threads:read` can request `GET /api/scheduled-tasks/{task_id}/runs?status=failed&limit=50&offset=0` for an owned task. The optional `status` accepts `queued`, `launching`, `running`, `success`, `failed`, `skipped`, `interrupted`, or `unmet`; these are occurrence statuses, so task statuses such as `completed` are invalid (422).

View File

@ -69,6 +69,7 @@ from deerflow.runtime import CancelOutcome, ConflictError, RunRecord, RunStatus,
from deerflow.runtime.runs.store.base import format_run_cursor_created_at, normalize_run_created_at_iso
from deerflow.runtime.secret_context import redact_config_secrets, redact_metadata_secrets
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.utils.llm_text import strip_leading_think_blocks
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, get_original_user_content_text, message_to_text
from deerflow.utils.thread_id import ThreadId
from deerflow.workspace_changes import get_workspace_changes_response
@ -592,10 +593,11 @@ def _run_last_ai_matches_message(record: RunRecord, message: Any) -> bool:
last_ai_message = (record.last_ai_message or "").strip()
if not last_ai_message:
return False
target_text = _message_text(message).strip()
target_text = _message_text(message)
if not target_text:
return False
return last_ai_message == target_text[: len(last_ai_message)]
# Match both historical raw summaries and newer visible-answer summaries.
return any(last_ai_message == text[: len(last_ai_message)] for text in (target_text.strip(), strip_leading_think_blocks(target_text)))
async def _find_target_run_id(

View File

@ -28,10 +28,10 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c
**Thread message cursors:** `list_messages` applies both exclusive bounds before `limit`, paging forward whenever `after_seq` is supplied.
**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.
**Journal capture**: Image-only input stops batch scans and later human-event
capture despite an empty summary. AI summaries strip leading think sections
before the 2000-character cap, preserving events, literal answer tags, and
prior useful summaries. Tests: `test_run_journal*.py` (callbacks, full/delta).
**Per-call LLM telemetry** (`runtime/journal.py`): `RunJournal` adds observation-only keys
to `llm.ai.response` metadata and to `llm.error` metadata (previously empty): `langchain_run_id`,

View File

@ -47,6 +47,7 @@ from deerflow.runtime.events.catalog import (
RUN_ERROR_EVENT,
RUN_START_EVENT,
)
from deerflow.utils.llm_text import strip_leading_think_blocks
from deerflow.utils.messages import message_to_text, restore_original_human_message
if TYPE_CHECKING:
@ -411,7 +412,7 @@ class RunJournal(BaseCallbackHandler):
is_ai_message = isinstance(message, AIMessage) or getattr(message, "type", None) == "ai"
if not is_ai_message or (caller is not None and caller != "lead_agent"):
return None
text = self._message_text(message).strip()
text = strip_leading_think_blocks(self._message_text(message))
return text[:2000] if text else None
def _record_message_summary(self, message: BaseMessage, *, caller: str | None = None) -> None:

View File

@ -21,6 +21,9 @@ suffix. Keep its case-insensitive tag handling, optional whitespace before
the closing `>`, and the `truncate_unclosed` behavior. Regression coverage
lives in `tests/test_utils_llm_text.py`.
Display summaries use `strip_leading_think_blocks` before limiting text;
`test_run_journal_visible_summary.py` preserves literal tags in the answer.
### Agent / Tool Assembly Off-Load
`start_run` also resolves the factory through `run_assembly` before admission:

View File

@ -57,6 +57,33 @@ def strip_think_blocks(text: str, *, truncate_unclosed: bool = True) -> str:
return "".join(parts).strip()
def strip_leading_think_blocks(text: str) -> str:
"""Remove leading reasoning for display summaries, leaving tags in the answer intact.
Unlike structured-response parsing, a displayed answer may explain the
tag in prose or code. Only a leading XML-style reasoning section is
removed; an unfinished section has no answer to summarize.
"""
start = 0
while True:
whitespace_start = start
while start < len(text) and text[start].isspace():
start += 1
newline = text.rfind("\n", whitespace_start, start)
indent = text[max(newline + 1, whitespace_start) : start]
if (whitespace_start == 0 or newline >= 0) and ("\t" in indent or indent.startswith(" ")):
break # Root-level indented Markdown code is a literal example.
opening = _THINK_OPEN_PREFIX_RE.match(text, start)
if opening is None or (opening.end() < len(text) and not (text[opening.end()].isspace() or text[opening.end()] == ">")):
break
end = text.find(">", opening.end())
closing = _find_think_close(text, end + 1) if end >= 0 else None
if closing is None:
return ""
start = closing[1]
return text[start:].strip()
def strip_markdown_code_fence(text: str) -> str:
"""Remove a single wrapping markdown code fence when present."""
stripped = text.strip()

View File

@ -0,0 +1,126 @@
"""Derived run summaries exclude leading reasoning without rewriting messages."""
from uuid import uuid4
import httpx
import pytest
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.outputs import ChatGeneration, LLMResult
from langchain_openai import ChatOpenAI
from app.scheduler.notification_text import render_notification_text
from deerflow.persistence.scheduled_task_runs.sql import run_summary
from deerflow.runtime.events.store.memory import MemoryRunEventStore
from deerflow.runtime.journal import RunJournal
ANSWER = "The release checklist is complete."
@pytest.mark.anyio
@pytest.mark.parametrize(
("content", "expected"),
[
(f"<think>Compare candidates.</think>{ANSWER}", ANSWER),
(f"\n<think>\nCompare candidates.\n</think>\n{ANSWER}", ANSWER),
(f'<THINK class="reasoning">Compare candidates.</THINK >\n{ANSWER}', ANSWER),
(f"<think>First.</think>\n<think>Second.</think>\n{ANSWER}", ANSWER),
(f"<think>First.</think> <think>Second.</think>{ANSWER}", ANSWER),
("<think>" + "r" * 2400 + f"</think>\n{ANSWER}", ANSWER),
("<think>Only reasoning.</think>", None),
("<think>Unfinished reasoning.", None),
("<think", None),
(ANSWER, ANSWER),
("```xml\n<think>literal example</think>\n```", "```xml\n<think>literal example</think>\n```"),
("`<think>literal example</think>`", "`<think>literal example</think>`"),
("Use <think>literal example</think> in this document.", "Use <think>literal example</think> in this document."),
(" <think>indented code</think>", "<think>indented code</think>"),
("<thinker>ordinary XML</thinker>", "<thinker>ordinary XML</thinker>"),
("<think>Reasoning.</think>\n`<think>literal example</think>`", "`<think>literal example</think>`"),
([{"type": "text", "text": "<think>Reasoning.</think>"}, {"type": "text", "text": ANSWER}], ANSWER),
(f"<think>Reasoning.</think>{'a' * 2500}", "a" * 2000),
],
ids=[
"short",
"multiline",
"attributes",
"two-blocks",
"same-line-blocks",
"long-reasoning",
"only-reasoning",
"unclosed",
"partial-tag",
"plain",
"fenced-literal",
"inline-literal",
"prose-literal",
"indented-literal",
"not-tag",
"literal-after-reasoning",
"content-blocks",
"long-answer",
],
)
async def test_visible_summary_preserves_original_response(content, expected):
store = MemoryRunEventStore()
journal = RunJournal("run-visible", "thread-visible", store)
message = AIMessage(content=content, usage_metadata={"input_tokens": 10, "output_tokens": 5, "total_tokens": 15})
original = message.model_dump()
journal.on_llm_end(LLMResult(generations=[[ChatGeneration(message=message)]]), run_id=uuid4(), tags=["lead_agent"])
await journal.flush()
completion = journal.get_completion_data()
assert completion["last_ai_message"] == expected
assert completion["message_count"] == 1
assert completion["total_tokens"] == 15
events = await store.list_events("thread-visible", "run-visible")
response = next(event for event in events if event["event_type"] == "llm.ai.response")
assert response["content"]["content"] == content
assert message.model_dump() == original
@pytest.mark.anyio
async def test_thought_only_response_keeps_previous_useful_summary():
journal = RunJournal("run-visible", "thread-visible", MemoryRunEventStore())
for content in (ANSWER, "<think>Unfinished reasoning."):
journal.on_llm_end(LLMResult(generations=[[ChatGeneration(message=AIMessage(content=content))]]), run_id=uuid4(), tags=["lead_agent"])
await journal.flush()
assert journal.get_completion_data()["last_ai_message"] == ANSWER
assert journal.get_completion_data()["message_count"] == 2
@pytest.mark.anyio
async def test_openai_compatible_reply_reaches_notice_without_inline_reasoning():
content = "<think>" + "r" * 2400 + f"</think>\n{ANSWER}"
requests = []
def respond(request):
assert request.method == "POST" and request.url.path == "/v1/chat/completions"
requests.append(request)
return httpx.Response(
200,
json={
"id": "offline-response",
"object": "chat.completion",
"created": 0,
"model": "reasoning-model",
"choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": content}}],
"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
},
)
store = MemoryRunEventStore()
journal = RunJournal("run-visible", "thread-visible", store)
async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client:
model = ChatOpenAI(model="reasoning-model", api_key="offline-test-key", base_url="https://model.invalid/v1", http_async_client=client)
message = await model.ainvoke([HumanMessage(content="Check the release checklist")], config={"callbacks": [journal], "tags": ["lead_agent"]})
await journal.flush()
summary = journal.get_completion_data()["last_ai_message"]
notice = render_notification_text({"event": "run_completed", "run_id": "run-visible", "payload": {"run_status": "success", "task_title": "Check release", "result_summary": run_summary(summary)}})
assert len(requests) == 1
assert message.content == content
assert run_summary(summary) == ANSWER
assert f"Result: {ANSWER}" in notice
assert "<think>" not in notice and "rrrr" not in notice
events = await store.list_events("thread-visible", "run-visible")
assert any(event["content"].get("content") == content for event in events if event["event_type"] == "llm.ai.response")

View File

@ -12,7 +12,7 @@ from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, Tool
from langgraph.checkpoint.base import empty_checkpoint, uuid6
from langgraph.checkpoint.memory import InMemorySaver
from deerflow.runtime import RunStatus
from deerflow.runtime import DisconnectMode, RunRecord, RunStatus
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
@ -1541,18 +1541,28 @@ def test_prepare_regenerate_payload_rejects_non_latest_assistant():
assert exc.value.detail == "Only the latest assistant message can be regenerated"
def test_prepare_regenerate_payload_falls_back_to_matching_run_when_events_are_missing():
@pytest.mark.parametrize(
("content", "summary"),
[
("answer", "answer"),
("<think>Reasoning.</think>answer", "<think>Reasoning.</think>answer"),
("<think>Reasoning.</think>answer", "answer"),
("<think>" + "r" * 2400 + "</think>" + "a" * 2500, "a" * 2000),
],
ids=["plain", "legacy-summary", "visible-summary", "long-reasoning"],
)
def test_prepare_regenerate_payload_falls_back_to_matching_run_when_events_are_missing(content, summary):
from app.gateway.routers.thread_runs import _prepare_regenerate_payload
human = HumanMessage(id="human-1", content="question")
ai = AIMessage(id="ai-1", content="answer")
ai = AIMessage(id="ai-1", content=content)
base = _checkpoint("ckpt-base", [])
after_human = _checkpoint("ckpt-human", [human])
latest = _checkpoint("ckpt-ai", [human, ai])
checkpointer = FakeCheckpointer([latest, after_human, base])
run_manager = FakeRunManager(
[
SimpleNamespace(run_id="run-latest", status=RunStatus.success, last_ai_message="answer"),
RunRecord(run_id="run-latest", thread_id="thread-1", assistant_id="lead_agent", status=RunStatus.success, on_disconnect=DisconnectMode.continue_, last_ai_message=summary),
SimpleNamespace(run_id="run-older", status=RunStatus.error, last_ai_message="answer"),
]
)