diff --git a/astrbot/core/agent/message.py b/astrbot/core/agent/message.py
index ad3b57cb25..4292f4c04e 100644
--- a/astrbot/core/agent/message.py
+++ b/astrbot/core/agent/message.py
@@ -1,7 +1,7 @@
# Inspired by MoonshotAI/kosong, credits to MoonshotAI/kosong authors for the original implementation.
# License: Apache License 2.0
-from typing import Any, ClassVar, Literal, cast
+from typing import Any, ClassVar, Literal, TypeVar, cast
from pydantic import (
BaseModel,
@@ -13,6 +13,8 @@
)
from pydantic_core import core_schema
+ContentPartT = TypeVar("ContentPartT", bound="ContentPart")
+
class ContentPart(BaseModel):
"""A part of the content in a message."""
@@ -20,6 +22,7 @@ class ContentPart(BaseModel):
__content_part_registry: ClassVar[dict[str, type["ContentPart"]]] = {}
type: Literal["text", "think", "image_url", "audio_url"]
+ _no_save: bool = PrivateAttr(default=False)
def __init_subclass__(cls, **kwargs: Any) -> None:
super().__init_subclass__(**kwargs)
@@ -50,7 +53,10 @@ def validate_content_part(value: Any) -> Any:
if not isinstance(type_value, str):
raise ValueError(f"Cannot validate {value} as ContentPart")
target_class = cls.__content_part_registry[type_value]
- return target_class.model_validate(value)
+ part = target_class.model_validate(value)
+ if cast(dict[str, Any], value).get("_no_save"):
+ part._no_save = True
+ return part
raise ValueError(f"Cannot validate {value} as ContentPart")
@@ -59,6 +65,17 @@ def validate_content_part(value: Any) -> Any:
# for subclasses, use the default schema
return handler(source_type)
+ def mark_as_temp(self: ContentPartT) -> ContentPartT:
+ """Mark this content part as provider-facing only, not persisted."""
+ self._no_save = True
+ return self
+
+ def model_dump_for_context(self) -> dict[str, Any]:
+ data = self.model_dump()
+ if self._no_save:
+ data["_no_save"] = True
+ return data
+
class TextPart(ContentPart):
"""
@@ -329,7 +346,14 @@ def dump_messages_with_checkpoints(messages: list[Message]) -> list[dict]:
"""Dump runtime messages and reinsert bound checkpoint segments."""
dumped: list[dict] = []
for message in messages:
- dumped.append(message.model_dump())
+ message_data = message.model_dump()
+ if isinstance(message.content, list):
+ message_data["content"] = [
+ part.model_dump()
+ for part in message.content
+ if not getattr(part, "_no_save", False)
+ ]
+ dumped.append(message_data)
if message._checkpoint_after is not None:
dumped.append(
CheckpointMessageSegment(content=message._checkpoint_after).model_dump()
diff --git a/astrbot/core/provider/entities.py b/astrbot/core/provider/entities.py
index 9b64196e7a..8e12683ffb 100644
--- a/astrbot/core/provider/entities.py
+++ b/astrbot/core/provider/entities.py
@@ -206,7 +206,7 @@ async def assemble_context(self) -> dict:
# 2. 额外的内容块（系统提醒、指令等）
if self.extra_user_content_parts:
for part in self.extra_user_content_parts:
- content_blocks.append(part.model_dump())
+ content_blocks.append(part.model_dump_for_context())
# 3. 图片内容
if self.image_urls:
diff --git a/docs/en/dev/star/guides/listen-message-event.md b/docs/en/dev/star/guides/listen-message-event.md
index b63609818a..e798475add 100644
--- a/docs/en/dev/star/guides/listen-message-event.md
+++ b/docs/en/dev/star/guides/listen-message-event.md
@@ -295,6 +295,14 @@ async def my_custom_hook_1(self, event: AstrMessageEvent, req: ProviderRequest):
> )
> ```
>
+> If the appended content should only affect the current LLM request and should not be persisted into conversation history, call `.mark_as_temp()` to mark it as temporary:
+>
+> ```python
+> req.extra_user_content_parts.append(
+> TextPart(text="This hint only applies to the current request.").mark_as_temp()
+> )
+> ```
+>
> For long-term memory, knowledge bases, or external system queries that may be large or unnecessary for every round, do not put everything directly into the prompt. Prefer registering them as `llm_tool` functions so the model can call them when needed, or retrieve only a small relevant summary in your plugin and append that summary through `extra_user_content_parts`.
> You cannot use yield to send messages here. If you need to send, please use the `event.send()` method directly.
diff --git a/docs/zh/dev/star/guides/listen-message-event.md b/docs/zh/dev/star/guides/listen-message-event.md
index 9cf96f4366..b8187b00b9 100644
--- a/docs/zh/dev/star/guides/listen-message-event.md
+++ b/docs/zh/dev/star/guides/listen-message-event.md
@@ -314,6 +314,14 @@ async def my_custom_hook_1(self, event: AstrMessageEvent, req: ProviderRequest):
> )
> ```
>
+> 如果追加的内容只希望参与本轮 LLM 请求，不希望被持久化到会话历史中，可以调用 `.mark_as_temp()` 标记为临时内容（`>= v4.24.0`）：
+>
+> ```python
+> req.extra_user_content_parts.append(
+> TextPart(text="这段提示只在本轮请求中生效。").mark_as_temp()
+> )
+> ```
+>
> 对于长期记忆、知识库、外部系统查询等内容量较大或不一定每轮都需要的信息，不建议全部塞进提示词。可以优先注册为 `llm_tool`，让模型在需要时调用；也可以先在插件中检索出本轮真正相关的少量摘要，再放入 `extra_user_content_parts`。
#### LLM 请求完成时
diff --git a/docs/zh/dev/star/plugin.md b/docs/zh/dev/star/plugin.md
index 9e576f1c11..c7b2df5166 100644
--- a/docs/zh/dev/star/plugin.md
+++ b/docs/zh/dev/star/plugin.md
@@ -548,6 +548,14 @@ async def my_custom_hook_1(self, event: AstrMessageEvent, req: ProviderRequest):
> )
> ```
>
+> 如果追加的内容只希望参与本轮 LLM 请求，不希望被持久化到会话历史中，可以调用 `.mark_as_temp()` 标记为临时内容：
+>
+> ```python
+> req.extra_user_content_parts.append(
+> TextPart(text="这段提示只在本轮请求中生效。").mark_as_temp()
+> )
+> ```
+>
> 对于长期记忆、知识库、外部系统查询等内容量较大或不一定每轮都需要的信息，不建议全部塞进提示词。可以优先注册为 `llm_tool`，让模型在需要时调用；也可以先在插件中检索出本轮真正相关的少量摘要，再放入 `extra_user_content_parts`。
> 这里不能使用 yield 来发送消息。如需发送，请直接使用 `event.send()` 方法。
diff --git a/tests/test_conversation_checkpoint.py b/tests/test_conversation_checkpoint.py
index 96113b7d1d..40eae46ff9 100644
--- a/tests/test_conversation_checkpoint.py
+++ b/tests/test_conversation_checkpoint.py
@@ -4,11 +4,13 @@
CheckpointData,
CheckpointMessageSegment,
Message,
+ TextPart,
bind_checkpoint_messages,
dump_messages_with_checkpoints,
get_checkpoint_id,
strip_checkpoint_messages,
)
+from astrbot.core.provider.entities import ProviderRequest
from astrbot.core.provider.provider import Provider
from astrbot.dashboard.routes.chat import ChatRoute
@@ -81,6 +83,60 @@ def test_dump_checkpoint_messages_drops_checkpoint_when_message_is_dropped():
]
+def test_dump_messages_filters_temp_content_parts():
+ messages = [
+ Message(
+ role="user",
+ content=[
+ TextPart(text="persisted"),
+ TextPart(text="temporary").mark_as_temp(),
+ ],
+ ),
+ Message(role="assistant", content="ok"),
+ ]
+
+ assert dump_messages_with_checkpoints(messages) == [
+ {"role": "user", "content": [{"type": "text", "text": "persisted"}]},
+ {"role": "assistant", "content": "ok"},
+ ]
+
+
+def test_content_part_no_save_round_trip_from_dict():
+ message = Message.model_validate(
+ {
+ "role": "user",
+ "content": [
+ {"type": "text", "text": "persisted"},
+ {"type": "text", "text": "temporary", "_no_save": True},
+ ],
+ }
+ )
+
+ assert isinstance(message.content, list)
+ assert message.content[0]._no_save is False
+ assert message.content[1]._no_save is True
+ assert dump_messages_with_checkpoints([message]) == [
+ {"role": "user", "content": [{"type": "text", "text": "persisted"}]},
+ ]
+
+
+@pytest.mark.asyncio
+async def test_provider_request_assemble_context_preserves_temp_content_part_marker():
+ request = ProviderRequest(
+ prompt="hello",
+ extra_user_content_parts=[TextPart(text="temporary").mark_as_temp()],
+ )
+
+ message = Message.model_validate(await request.assemble_context())
+
+ assert isinstance(message.content, list)
+ assert message.content[1].text == "temporary"
+ assert message.content[1]._no_save is True
+ assert dump_messages_with_checkpoints([message]) == [
+ {"role": "user", "content": [{"type": "text", "text": "hello"}]},
+ ]
+
+
def test_provider_ensure_message_to_dicts_skips_checkpoints():
messages = [
Message(role="user", content="hello"),