From 94cf37a2ae9253ddd180696c72f262ccae0484b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= Date: Mon, 17 Aug 2026 10:12:43 +0800 Subject: [PATCH] fix: normalize system messages before model calls --- .../middlewares/rag_citation_middleware.py | 7 +- .../writing_setup_round_cap_middleware.py | 11 +-- .../deerflow/models/message_ordering.py | 82 ++++++++++++++++++ .../deerflow/models/mindie_provider.py | 12 +-- .../deerflow/models/patched_deepseek.py | 23 ++--- .../deerflow/models/patched_minimax.py | 19 +++-- .../harness/deerflow/models/patched_openai.py | 27 +++--- .../harness/deerflow/models/vllm_provider.py | 23 ++--- .../harness/tests/test_message_ordering.py | 84 +++++++++++++++++++ 9 files changed, 236 insertions(+), 52 deletions(-) create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/models/message_ordering.py create mode 100644 offline-backend-20260512/backend/packages/harness/tests/test_message_ordering.py diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/rag_citation_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/rag_citation_middleware.py index 07883766..53e47a4e 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/rag_citation_middleware.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/rag_citation_middleware.py @@ -12,7 +12,7 @@ from typing import Any, override from langchain.agents import AgentState from langchain.agents.middleware import AgentMiddleware from langchain.agents.middleware.types import ModelCallResult, ModelRequest, ModelResponse -from langchain_core.messages import SystemMessage +from langchain_core.messages import HumanMessage from deerflow.runtime.references import ( build_latest_reference_prompt, @@ -105,7 +105,10 @@ class RagCitationMiddleware(AgentMiddleware[AgentState]): ) patched_messages = [ *messages, - SystemMessage(content=prompt), + HumanMessage( + content=prompt, + additional_kwargs={"hide_from_ui": True, "lc_source": "rag_citation"}, + ), ] return request.override(messages=patched_messages) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/writing_setup_round_cap_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/writing_setup_round_cap_middleware.py index 5fadb642..cc27fd31 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/writing_setup_round_cap_middleware.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/writing_setup_round_cap_middleware.py @@ -14,13 +14,13 @@ from typing import override from langchain.agents import AgentState from langchain.agents.middleware import AgentMiddleware from langchain.agents.middleware.types import ModelCallResult, ModelRequest, ModelResponse -from langchain_core.messages import HumanMessage, SystemMessage +from langchain_core.messages import HumanMessage class WritingSetupRoundCapMiddleware(AgentMiddleware[AgentState]): """达到澄清轮数上限时，强制模型调用 ``setup_writing`` 提交。 - 软约束（注入 SystemMessage）+ 硬约束（``tool_choice`` 锁定工具）双保险：即使个别模型 + 软约束（注入隐藏 HumanMessage）+ 硬约束（``tool_choice`` 锁定工具）双保险：即使个别模型 无视提示，``tool_choice`` 也会逼它产出一次 ``setup_writing`` 调用。 """ @@ -33,8 +33,8 @@ class WritingSetupRoundCapMiddleware(AgentMiddleware[AgentState]): human_count = sum(1 for m in messages if isinstance(m, HumanMessage)) return human_count >= self._max_rounds - def _reminder(self) -> SystemMessage: - return SystemMessage( + def _reminder(self) -> HumanMessage: + return HumanMessage( content=( f"【系统强制】意图澄清已达 {self._max_rounds} 轮上限。" "现在必须立即调用 setup_writing 提交写作配置，禁止再向用户提任何问题。" @@ -42,7 +42,8 @@ class WritingSetupRoundCapMiddleware(AgentMiddleware[AgentState]): "word_count_target 默认 800、target_audience 默认「普通读者」、" "writing_mode 默认 loose、material_source 默认 knowledge_base、" "article_type 不确定就留空。" - ) + ), + additional_kwargs={"hide_from_ui": True, "lc_source": "writing_setup_round_cap"}, ) def _patch(self, request: ModelRequest) -> ModelRequest: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/models/message_ordering.py b/offline-backend-20260512/backend/packages/harness/deerflow/models/message_ordering.py new file mode 100644 index 00000000..d7811b49 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/models/message_ordering.py @@ -0,0 +1,82 @@ +"""Utilities for provider-specific chat message ordering constraints.""" + +from __future__ import annotations + +from typing import Any + +from langchain_core.messages import BaseMessage, SystemMessage + + +_SYSTEM_SEPARATOR = "\n\n---\n\n" + + +def normalize_chat_payload_messages(payload: dict[str, Any]) -> None: + """Ensure payload ``system`` messages are represented once at the beginning. + + Several OpenAI-compatible gateways reject requests when a transient + ``system`` message appears after user/assistant/tool history. LangChain and + DeerFlow middleware can legitimately create those transient messages, so the + provider boundary folds every system payload into a single leading message + while preserving all non-system messages in their original order. + """ + messages = payload.get("messages") + if not isinstance(messages, list): + return + + normalized = _normalize_payload_message_list(messages) + if normalized is not messages: + payload["messages"] = normalized + + +def normalize_langchain_messages(messages: list[BaseMessage]) -> list[BaseMessage]: + """Return LangChain messages with all system context merged at the front.""" + system_messages = [message for message in messages if isinstance(message, SystemMessage)] + if not system_messages: + return messages + + first_non_system_index = next((idx for idx, message in enumerate(messages) if not isinstance(message, SystemMessage)), len(messages)) + if len(system_messages) == first_non_system_index: + return messages + + merged = system_messages[0].model_copy( + update={ + "content": _merge_contents([message.content for message in system_messages]), + } + ) + return [merged, *(message for message in messages if not isinstance(message, SystemMessage))] + + +def _normalize_payload_message_list(messages: list[Any]) -> list[Any]: + system_messages = [message for message in messages if isinstance(message, dict) and message.get("role") == "system"] + if not system_messages: + return messages + + first_non_system_index = next( + ( + idx + for idx, message in enumerate(messages) + if not (isinstance(message, dict) and message.get("role") == "system") + ), + len(messages), + ) + if len(system_messages) == first_non_system_index: + return messages + + merged_system = dict(system_messages[0]) + merged_system["content"] = _merge_contents([message.get("content") for message in system_messages]) + return [merged_system, *(message for message in messages if not (isinstance(message, dict) and message.get("role") == "system"))] + + +def _merge_contents(contents: list[Any]) -> Any: + if all(isinstance(content, str) for content in contents): + return _SYSTEM_SEPARATOR.join(content for content in contents if content) + + blocks: list[Any] = [] + for idx, content in enumerate(contents): + if idx > 0: + blocks.append({"type": "text", "text": _SYSTEM_SEPARATOR}) + if isinstance(content, list): + blocks.extend(content) + elif content is not None: + blocks.append({"type": "text", "text": str(content)}) + return blocks diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/models/mindie_provider.py b/offline-backend-20260512/backend/packages/harness/deerflow/models/mindie_provider.py index bb32d1df..f745af3c 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/models/mindie_provider.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/models/mindie_provider.py @@ -6,10 +6,12 @@ import uuid from collections.abc import Iterator import httpx -from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, ToolMessage -from langchain_core.outputs import ChatGenerationChunk, ChatResult -from langchain_openai import ChatOpenAI - +from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, ToolMessage +from langchain_core.outputs import ChatGenerationChunk, ChatResult +from langchain_openai import ChatOpenAI + +from deerflow.models.message_ordering import normalize_langchain_messages + def _fix_messages(messages: list) -> list: """Sanitize incoming messages for MindIE compatibility. @@ -55,7 +57,7 @@ def _fix_messages(messages: list) -> list: fixed.append(msg.model_copy(update={"content": text})) - return fixed + return normalize_langchain_messages(fixed) def _parse_xml_tool_call_to_dict(content: str) -> tuple[str, list[dict]]: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/models/patched_deepseek.py b/offline-backend-20260512/backend/packages/harness/deerflow/models/patched_deepseek.py index 2baa3b05..deaac811 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/models/patched_deepseek.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/models/patched_deepseek.py @@ -9,10 +9,12 @@ on all assistant messages when thinking mode is enabled. from typing import Any -from langchain_core.language_models import LanguageModelInput -from langchain_core.messages import AIMessage -from langchain_deepseek import ChatDeepSeek - +from langchain_core.language_models import LanguageModelInput +from langchain_core.messages import AIMessage +from langchain_deepseek import ChatDeepSeek + +from deerflow.models.message_ordering import normalize_chat_payload_messages + class PatchedChatDeepSeek(ChatDeepSeek): """ChatDeepSeek with proper reasoning_content preservation. @@ -65,9 +67,10 @@ class PatchedChatDeepSeek(ChatDeepSeek): ai_messages = [m for m in original_messages if isinstance(m, AIMessage)] assistant_payloads = [(i, m) for i, m in enumerate(payload_messages) if m.get("role") == "assistant"] - for (idx, payload_msg), ai_msg in zip(assistant_payloads, ai_messages): - reasoning_content = ai_msg.additional_kwargs.get("reasoning_content") - if reasoning_content is not None: - payload_messages[idx]["reasoning_content"] = reasoning_content - - return payload + for (idx, payload_msg), ai_msg in zip(assistant_payloads, ai_messages): + reasoning_content = ai_msg.additional_kwargs.get("reasoning_content") + if reasoning_content is not None: + payload_messages[idx]["reasoning_content"] = reasoning_content + + normalize_chat_payload_messages(payload) + return payload diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/models/patched_minimax.py b/offline-backend-20260512/backend/packages/harness/deerflow/models/patched_minimax.py index 946e4fe2..7d57f2d5 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/models/patched_minimax.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/models/patched_minimax.py @@ -20,11 +20,13 @@ from langchain_core.language_models import LanguageModelInput from langchain_core.messages import AIMessage, AIMessageChunk from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult from langchain_openai import ChatOpenAI -from langchain_openai.chat_models.base import ( - _convert_delta_to_message_chunk, - _create_usage_metadata, -) - +from langchain_openai.chat_models.base import ( + _convert_delta_to_message_chunk, + _create_usage_metadata, +) + +from deerflow.models.message_ordering import normalize_chat_payload_messages + _THINK_TAG_RE = re.compile(r"\s*(.*?)\s*", re.DOTALL) @@ -112,9 +114,10 @@ class PatchedChatMiniMax(ChatOpenAI): **extra_body, "reasoning_split": True, } - else: - payload["extra_body"] = {"reasoning_split": True} - return payload + else: + payload["extra_body"] = {"reasoning_split": True} + normalize_chat_payload_messages(payload) + return payload def _convert_chunk_to_generation_chunk( self, diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/models/patched_openai.py b/offline-backend-20260512/backend/packages/harness/deerflow/models/patched_openai.py index f8bf1918..6dec1f5b 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/models/patched_openai.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/models/patched_openai.py @@ -29,9 +29,11 @@ from typing import Any from langchain_core.language_models import LanguageModelInput from langchain_core.messages import AIMessage, AIMessageChunk -from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult -from langchain_openai import ChatOpenAI - +from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult +from langchain_openai import ChatOpenAI + +from deerflow.models.message_ordering import normalize_chat_payload_messages + class PatchedChatOpenAI(ChatOpenAI): """ChatOpenAI with ``thought_signature`` preservation for Gemini thinking via OpenAI gateway. @@ -87,15 +89,16 @@ class PatchedChatOpenAI(ChatOpenAI): if payload_msg.get("role") == "assistant" and isinstance(orig_msg, AIMessage): _restore_tool_call_signatures(payload_msg, orig_msg) _restore_reasoning_content(payload_msg, orig_msg) - else: - # Fallback: match assistant-role entries positionally against AIMessages. - ai_messages = [m for m in original_messages if isinstance(m, AIMessage)] - assistant_payloads = [(i, m) for i, m in enumerate(payload_messages) if m.get("role") == "assistant"] - for (_, payload_msg), ai_msg in zip(assistant_payloads, ai_messages): - _restore_tool_call_signatures(payload_msg, ai_msg) - _restore_reasoning_content(payload_msg, ai_msg) - - return payload + else: + # Fallback: match assistant-role entries positionally against AIMessages. + ai_messages = [m for m in original_messages if isinstance(m, AIMessage)] + assistant_payloads = [(i, m) for i, m in enumerate(payload_messages) if m.get("role") == "assistant"] + for (_, payload_msg), ai_msg in zip(assistant_payloads, ai_messages): + _restore_tool_call_signatures(payload_msg, ai_msg) + _restore_reasoning_content(payload_msg, ai_msg) + + normalize_chat_payload_messages(payload) + return payload def _convert_chunk_to_generation_chunk( self, diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/models/vllm_provider.py b/offline-backend-20260512/backend/packages/harness/deerflow/models/vllm_provider.py index 78f83a73..d72bfb8c 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/models/vllm_provider.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/models/vllm_provider.py @@ -32,9 +32,11 @@ from langchain_core.messages import ( ) from langchain_core.messages.tool import tool_call_chunk from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult -from langchain_openai import ChatOpenAI -from langchain_openai.chat_models.base import _create_usage_metadata - +from langchain_openai import ChatOpenAI +from langchain_openai.chat_models.base import _create_usage_metadata + +from deerflow.models.message_ordering import normalize_chat_payload_messages + def _normalize_vllm_chat_template_kwargs(payload: dict[str, Any]) -> None: """Map DeerFlow's legacy ``thinking`` toggle to vLLM/Qwen's ``enable_thinking``. @@ -182,13 +184,14 @@ class VllmChatModel(ChatOpenAI): for payload_msg, orig_msg in zip(payload_messages, original_messages): if payload_msg.get("role") == "assistant" and isinstance(orig_msg, AIMessage): _restore_reasoning_field(payload_msg, orig_msg) - else: - ai_messages = [message for message in original_messages if isinstance(message, AIMessage)] - assistant_payloads = [message for message in payload_messages if message.get("role") == "assistant"] - for payload_msg, ai_msg in zip(assistant_payloads, ai_messages): - _restore_reasoning_field(payload_msg, ai_msg) - - return payload + else: + ai_messages = [message for message in original_messages if isinstance(message, AIMessage)] + assistant_payloads = [message for message in payload_messages if message.get("role") == "assistant"] + for payload_msg, ai_msg in zip(assistant_payloads, ai_messages): + _restore_reasoning_field(payload_msg, ai_msg) + + normalize_chat_payload_messages(payload) + return payload def _create_chat_result(self, response: dict | openai.BaseModel, generation_info: dict | None = None) -> ChatResult: """Preserve vLLM reasoning on non-streaming responses.""" diff --git a/offline-backend-20260512/backend/packages/harness/tests/test_message_ordering.py b/offline-backend-20260512/backend/packages/harness/tests/test_message_ordering.py new file mode 100644 index 00000000..d289f9b4 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/tests/test_message_ordering.py @@ -0,0 +1,84 @@ +import pytest +from langchain_core.messages import AIMessage, HumanMessage, SystemMessage + +from deerflow.agents.middlewares.writing_setup_round_cap_middleware import WritingSetupRoundCapMiddleware +from deerflow.models.message_ordering import normalize_chat_payload_messages, normalize_langchain_messages +from deerflow.models.patched_openai import PatchedChatOpenAI + + +def _multi_turn_payload_with_late_systems(turns: int): + messages = [{"role": "system", "content": "base system"}] + for idx in range(turns): + messages.append({"role": "user", "content": f"user {idx}"}) + messages.append({"role": "assistant", "content": f"assistant {idx}"}) + if idx >= 9 and (idx + 1) % 10 == 0: + messages.append({"role": "system", "content": f"late summary after {idx + 1} turns"}) + messages.append({"role": "system", "content": "tail forced instruction"}) + return {"messages": messages} + + +@pytest.mark.parametrize("turns", [12, 25, 50]) +def test_normalize_chat_payload_messages_merges_late_systems_after_many_turns(turns): + payload = _multi_turn_payload_with_late_systems(turns) + + normalize_chat_payload_messages(payload) + + roles = [message["role"] for message in payload["messages"]] + assert roles[0] == "system" + assert roles.count("system") == 1 + assert roles[1:5] == ["user", "assistant", "user", "assistant"] + assert "base system" in payload["messages"][0]["content"] + assert "late summary after 10 turns" in payload["messages"][0]["content"] + assert "tail forced instruction" in payload["messages"][0]["content"] + + +def test_normalize_langchain_messages_moves_late_systems_before_history(): + messages = [SystemMessage(content="base system")] + for idx in range(12): + messages.append(HumanMessage(content=f"user {idx}")) + messages.append(AIMessage(content=f"assistant {idx}")) + if idx == 10: + messages.append(SystemMessage(content="late compaction summary")) + + normalized = normalize_langchain_messages(messages) + + assert isinstance(normalized[0], SystemMessage) + assert sum(isinstance(message, SystemMessage) for message in normalized) == 1 + assert normalized[0].content == "base system\n\n---\n\nlate compaction summary" + assert isinstance(normalized[1], HumanMessage) + assert normalized[1].content == "user 0" + + +@pytest.mark.parametrize("turns", [12, 25, 50]) +def test_patched_openai_payload_has_only_leading_system_after_many_turns(turns): + model = PatchedChatOpenAI( + model="test-model", + api_key="test-key", + base_url="http://127.0.0.1:9/v1", + ) + messages = [SystemMessage(content="base system")] + for idx in range(turns): + messages.append(HumanMessage(content=f"user {idx}")) + messages.append(AIMessage(content=f"assistant {idx}")) + if idx >= 9 and (idx + 1) % 10 == 0: + messages.append(SystemMessage(content=f"late system after {idx + 1} turns")) + messages.append(SystemMessage(content="tail system")) + + payload = model._get_request_payload(messages) + + roles = [message["role"] for message in payload["messages"]] + assert roles[0] == "system" + assert roles.count("system") == 1 + assert "base system" in payload["messages"][0]["content"] + assert "late system after 10 turns" in payload["messages"][0]["content"] + assert "tail system" in payload["messages"][0]["content"] + + +def test_writing_round_cap_reminder_is_not_late_system_message(): + middleware = WritingSetupRoundCapMiddleware(max_rounds=3) + + reminder = middleware._reminder() + + assert isinstance(reminder, HumanMessage) + assert not isinstance(reminder, SystemMessage) + assert reminder.additional_kwargs["hide_from_ui"] is True