feat: add temporary extra user content parts

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,13 +13,16 @@
 )
 from pydantic_core import core_schema
 
+ContentPartT = TypeVar("ContentPartT", bound="ContentPart")
+
 
 class ContentPart(BaseModel):
     """A part of the content in a message."""
 
     __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()

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:

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="<runtime_hint>This hint only applies to the current request.</runtime_hint>").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.

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="<runtime_hint>这段提示只在本轮请求中生效。</runtime_hint>").mark_as_temp()
+> )
+> ```
+>
 > 对于长期记忆、知识库、外部系统查询等内容量较大或不一定每轮都需要的信息，不建议全部塞进提示词。可以优先注册为 `llm_tool`，让模型在需要时调用；也可以先在插件中检索出本轮真正相关的少量摘要，再放入 `extra_user_content_parts`。
 
 #### LLM 请求完成时

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="<runtime_hint>这段提示只在本轮请求中生效。</runtime_hint>").mark_as_temp()
+> )
+> ```
+>
 > 对于长期记忆、知识库、外部系统查询等内容量较大或不一定每轮都需要的信息，不建议全部塞进提示词。可以优先注册为 `llm_tool`，让模型在需要时调用；也可以先在插件中检索出本轮真正相关的少量摘要，再放入 `extra_user_content_parts`。
 
 > 这里不能使用 yield 来发送消息。如需发送，请直接使用 `event.send()` 方法。

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"),