mirror of
https://github.com/AstrBotDevs/AstrBot.git
synced 2026-10-16 15:06:04 +08:00
fix(cron): honor session context compression settings (#9992)
* fix(cron): honor session context compression settings * fix(agent): share session compression config across wakeups
This commit is contained in:
@@ -20,6 +20,7 @@ from astrbot.core.astr_agent_context import AstrAgentContext
|
||||
from astrbot.core.astr_main_agent_resources import (
|
||||
BACKGROUND_TASK_RESULT_WOKE_SYSTEM_PROMPT,
|
||||
)
|
||||
from astrbot.core.config.agent_runner import resolve_context_compression_config
|
||||
from astrbot.core.cron.events import CronMessageEvent
|
||||
from astrbot.core.message.components import Image
|
||||
from astrbot.core.message.message_event_result import (
|
||||
@@ -570,6 +571,9 @@ class FunctionToolExecutor(BaseFunctionToolExecutor[AstrAgentContext]):
|
||||
)
|
||||
config = MainAgentBuildConfig(
|
||||
tool_call_timeout=run_context.tool_call_timeout,
|
||||
**resolve_context_compression_config(
|
||||
cfg.get("agent_runner", {}).get("config", {}).get("compression", {})
|
||||
),
|
||||
streaming_response=provider_settings.get("stream", False),
|
||||
llm_safety_mode=persona_config.get("safety_mode", True),
|
||||
safety_mode_strategy=persona_config.get(
|
||||
|
||||
@@ -99,6 +99,41 @@ def get_agent_runner_config_default(runner_type: str) -> dict[str, Any]:
|
||||
return copy.deepcopy(AGENT_RUNNER_CONFIG_DEFAULTS[runner_type])
|
||||
|
||||
|
||||
def resolve_context_compression_config(
|
||||
compression_config: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""Map session compression settings to main agent build arguments.
|
||||
|
||||
Args:
|
||||
compression_config: The ``agent_runner.config.compression`` section.
|
||||
An empty mapping preserves the ordinary chat parser's defaults.
|
||||
|
||||
Returns:
|
||||
Build arguments shared by chat, Cron and background-result wakeups.
|
||||
The input mapping is not modified.
|
||||
"""
|
||||
max_turns = compression_config.get("max_turns", -1)
|
||||
trim_turns = compression_config.get("trim_turns", 1)
|
||||
dequeue_turns = min(
|
||||
max(1, trim_turns), max_turns - 1 if max_turns > 0 else trim_turns
|
||||
)
|
||||
return {
|
||||
"context_limit_reached_strategy": compression_config.get(
|
||||
"overflow_strategy", "truncate_by_turns"
|
||||
),
|
||||
"llm_compress_instruction": compression_config.get("instruction", ""),
|
||||
"llm_compress_keep_recent_ratio": compression_config.get(
|
||||
"keep_recent_ratio", 0.15
|
||||
),
|
||||
"llm_compress_provider_id": compression_config.get("provider_id", ""),
|
||||
"max_context_length": max_turns,
|
||||
"dequeue_context_length": max(1, dequeue_turns),
|
||||
"fallback_max_context_tokens": compression_config.get(
|
||||
"fallback_max_tokens", 128000
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def _normalize_value(value: Any, default: Any) -> Any:
|
||||
if isinstance(default, dict):
|
||||
if not isinstance(value, dict):
|
||||
|
||||
@@ -12,6 +12,7 @@ from apscheduler.triggers.date import DateTrigger
|
||||
|
||||
from astrbot import logger
|
||||
from astrbot.core.agent.tool import ToolSet
|
||||
from astrbot.core.config.agent_runner import resolve_context_compression_config
|
||||
from astrbot.core.cron.events import CronMessageEvent
|
||||
from astrbot.core.db import BaseDatabase
|
||||
from astrbot.core.db.po import CronJob
|
||||
@@ -462,6 +463,9 @@ class CronJobManager:
|
||||
)
|
||||
config = MainAgentBuildConfig(
|
||||
tool_call_timeout=tool_call_timeout,
|
||||
**resolve_context_compression_config(
|
||||
cfg.get("agent_runner", {}).get("config", {}).get("compression", {})
|
||||
),
|
||||
llm_safety_mode=persona_config.get("safety_mode", True),
|
||||
safety_mode_strategy=persona_config.get(
|
||||
"safety_mode_strategy", "system_prompt"
|
||||
|
||||
@@ -19,6 +19,7 @@ from astrbot.core.astr_main_agent import (
|
||||
MainAgentBuildResult,
|
||||
build_main_agent,
|
||||
)
|
||||
from astrbot.core.config.agent_runner import resolve_context_compression_config
|
||||
from astrbot.core.message.components import File, Image, Record, Reply, Video
|
||||
from astrbot.core.message.message_event_result import (
|
||||
MessageChain,
|
||||
@@ -95,28 +96,6 @@ class InternalAgentSubStage(Stage):
|
||||
"moonshotai_api_key", ""
|
||||
)
|
||||
|
||||
# 上下文管理相关
|
||||
self.context_limit_reached_strategy: str = compression_config.get(
|
||||
"overflow_strategy", "truncate_by_turns"
|
||||
)
|
||||
self.llm_compress_instruction: str = compression_config.get("instruction", "")
|
||||
self.llm_compress_keep_recent_ratio: float = compression_config.get(
|
||||
"keep_recent_ratio", 0.15
|
||||
)
|
||||
self.llm_compress_provider_id: str = compression_config.get("provider_id", "")
|
||||
self.max_context_length = compression_config.get("max_turns", -1)
|
||||
self.dequeue_context_length: int = min(
|
||||
max(1, compression_config.get("trim_turns", 1)),
|
||||
self.max_context_length - 1
|
||||
if self.max_context_length > 0
|
||||
else compression_config.get("trim_turns", 1),
|
||||
)
|
||||
if self.dequeue_context_length <= 0:
|
||||
self.dequeue_context_length = 1
|
||||
self.fallback_max_context_tokens: int = compression_config.get(
|
||||
"fallback_max_tokens", 128000
|
||||
)
|
||||
|
||||
self.llm_safety_mode = persona_config.get("safety_mode", True)
|
||||
self.safety_mode_strategy = persona_config.get(
|
||||
"safety_mode_strategy", "system_prompt"
|
||||
@@ -139,13 +118,7 @@ class InternalAgentSubStage(Stage):
|
||||
file_extract_enabled=self.file_extract_enabled,
|
||||
file_extract_prov=self.file_extract_prov,
|
||||
file_extract_msh_api_key=self.file_extract_msh_api_key,
|
||||
context_limit_reached_strategy=self.context_limit_reached_strategy,
|
||||
llm_compress_instruction=self.llm_compress_instruction,
|
||||
llm_compress_keep_recent_ratio=self.llm_compress_keep_recent_ratio,
|
||||
llm_compress_provider_id=self.llm_compress_provider_id,
|
||||
max_context_length=self.max_context_length,
|
||||
dequeue_context_length=self.dequeue_context_length,
|
||||
fallback_max_context_tokens=self.fallback_max_context_tokens,
|
||||
**resolve_context_compression_config(compression_config),
|
||||
llm_safety_mode=self.llm_safety_mode,
|
||||
safety_mode_strategy=self.safety_mode_strategy,
|
||||
computer_use_runtime=self.computer_use_runtime,
|
||||
|
||||
@@ -13,6 +13,7 @@ from astrbot.core.agent.message import Message, dump_messages_with_checkpoints
|
||||
from astrbot.core.agent.run_context import ContextWrapper
|
||||
from astrbot.core.agent.tool import FunctionTool, ToolSet
|
||||
from astrbot.core.astr_agent_tool_exec import FunctionToolExecutor
|
||||
from astrbot.core.config.agent_runner import resolve_context_compression_config
|
||||
from astrbot.core.conversation_mgr import Conversation
|
||||
from astrbot.core.cron.manager import CronJobManager
|
||||
from astrbot.core.message.components import File, Image, Plain, Reply, Video
|
||||
@@ -1629,6 +1630,54 @@ class TestBuildMainAgent:
|
||||
mock_runner.reset.assert_awaited_once()
|
||||
assert mock_runner.reset.await_args.kwargs["enforce_max_turns"] == 7
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_main_agent_passes_session_compression_to_runner(
|
||||
self, mock_event, mock_context, mock_provider
|
||||
):
|
||||
"""Pass the summary provider, token budget and turn limits to the runner."""
|
||||
compressor = MagicMock(spec=Provider)
|
||||
mock_context.get_provider_by_id.side_effect = lambda provider_id: (
|
||||
compressor if provider_id == "summary-model" else None
|
||||
)
|
||||
mock_context.get_using_provider.return_value = mock_provider
|
||||
mock_provider.get_model.return_value = "unknown-model-for-compression-test"
|
||||
_setup_conversation_for_build(mock_context.conversation_manager)
|
||||
|
||||
with (
|
||||
patch("astrbot.core.astr_main_agent.AgentRunner") as runner_cls,
|
||||
patch("astrbot.core.astr_main_agent.AstrAgentContext"),
|
||||
):
|
||||
runner = runner_cls.return_value
|
||||
runner.reset = AsyncMock()
|
||||
result = await ama.build_main_agent(
|
||||
event=mock_event,
|
||||
plugin_context=mock_context,
|
||||
config=ama.MainAgentBuildConfig(
|
||||
tool_call_timeout=60,
|
||||
**resolve_context_compression_config(
|
||||
{
|
||||
"overflow_strategy": "llm_compress",
|
||||
"provider_id": "summary-model",
|
||||
"instruction": "Keep unfinished tasks.",
|
||||
"keep_recent_ratio": 0.3,
|
||||
"max_turns": 12,
|
||||
"trim_turns": 3,
|
||||
"fallback_max_tokens": 16384,
|
||||
}
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
runner.reset.assert_awaited_once()
|
||||
kwargs = runner.reset.await_args.kwargs
|
||||
assert kwargs["llm_compress_provider"] is compressor
|
||||
assert kwargs["llm_compress_instruction"] == "Keep unfinished tasks."
|
||||
assert kwargs["llm_compress_keep_recent_ratio"] == 0.3
|
||||
assert kwargs["enforce_max_turns"] == 12
|
||||
assert kwargs["truncate_turns"] == 3
|
||||
assert kwargs["provider"].provider_config["max_context_tokens"] == 16384
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_main_agent_no_provider(self, mock_event, mock_context):
|
||||
"""Test building main agent when no provider is available."""
|
||||
|
||||
@@ -0,0 +1,204 @@
|
||||
"""Regression tests for session compression in chat, Cron and background wakeups."""
|
||||
|
||||
import copy
|
||||
import json
|
||||
from dataclasses import replace
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from astrbot.core.agent.tool import FunctionTool
|
||||
from astrbot.core.astr_agent_tool_exec import FunctionToolExecutor
|
||||
from astrbot.core.astr_main_agent import MainAgentBuildConfig
|
||||
from astrbot.core.config.agent_runner import resolve_context_compression_config
|
||||
from astrbot.core.config.default import DEFAULT_CONFIG
|
||||
from astrbot.core.cron.manager import CronJobManager
|
||||
from astrbot.core.pipeline.process_stage.method.agent_sub_stages.internal import (
|
||||
InternalAgentSubStage,
|
||||
)
|
||||
from astrbot.core.provider.entities import LLMResponse
|
||||
|
||||
|
||||
def test_explicit_build_settings_and_per_request_overrides_are_preserved():
|
||||
"""Keep direct callers and dataclasses.replace independent of raw settings."""
|
||||
defaults = MainAgentBuildConfig(tool_call_timeout=60)
|
||||
assert defaults.max_context_length == 50
|
||||
assert defaults.dequeue_context_length == 10
|
||||
direct = MainAgentBuildConfig(
|
||||
tool_call_timeout=60, max_context_length=7, dequeue_context_length=2
|
||||
)
|
||||
assert direct.max_context_length == 7
|
||||
assert direct.dequeue_context_length == 2
|
||||
|
||||
source = {"max_turns": 12, "trim_turns": 3, "provider_id": "summary-model"}
|
||||
configured = MainAgentBuildConfig(
|
||||
tool_call_timeout=60, **resolve_context_compression_config(source)
|
||||
)
|
||||
source["max_turns"] = 99
|
||||
per_request = replace(configured, streaming_response=False, max_context_length=8)
|
||||
assert configured.max_context_length == 12
|
||||
assert per_request.max_context_length == 8
|
||||
assert per_request.dequeue_context_length == 3
|
||||
assert per_request.llm_compress_provider_id == "summary-model"
|
||||
assert per_request.streaming_response is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("entrypoint", ["cron", "background"])
|
||||
@pytest.mark.parametrize(
|
||||
("compression", "expected"),
|
||||
[
|
||||
pytest.param(
|
||||
{
|
||||
"overflow_strategy": "llm_compress",
|
||||
"instruction": "Keep decisions and unfinished tasks.",
|
||||
"keep_recent_ratio": 0.3,
|
||||
"provider_id": "summary-model",
|
||||
"max_turns": 12,
|
||||
"trim_turns": 3,
|
||||
"fallback_max_tokens": 16384,
|
||||
},
|
||||
(
|
||||
"llm_compress",
|
||||
"Keep decisions and unfinished tasks.",
|
||||
0.3,
|
||||
"summary-model",
|
||||
12,
|
||||
3,
|
||||
16384,
|
||||
),
|
||||
id="configured-summary-model",
|
||||
),
|
||||
pytest.param(
|
||||
{},
|
||||
("truncate_by_turns", "", 0.15, "", -1, 1, 128000),
|
||||
id="empty-config",
|
||||
),
|
||||
pytest.param(
|
||||
None,
|
||||
("truncate_by_turns", "", 0.15, "", -1, 1, 128000),
|
||||
id="missing-cron-compression-section",
|
||||
),
|
||||
pytest.param(
|
||||
{"max_turns": 4, "trim_turns": 20},
|
||||
("truncate_by_turns", "", 0.15, "", 4, 3, 128000),
|
||||
id="trim-clamped-to-preserve-recent-turn",
|
||||
),
|
||||
pytest.param(
|
||||
{"max_turns": 1, "trim_turns": 0},
|
||||
("truncate_by_turns", "", 0.15, "", 1, 1, 128000),
|
||||
id="single-turn-minimum-trim",
|
||||
),
|
||||
pytest.param(
|
||||
{"max_turns": -1, "trim_turns": 5},
|
||||
("truncate_by_turns", "", 0.15, "", -1, 5, 128000),
|
||||
id="unlimited-turns",
|
||||
),
|
||||
pytest.param(
|
||||
{"max_turns": -1, "trim_turns": -3},
|
||||
("truncate_by_turns", "", 0.15, "", -1, 1, 128000),
|
||||
id="unlimited-turns-minimum-trim",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_wakeup_uses_same_compression_settings_as_chat(
|
||||
compression, expected, entrypoint
|
||||
):
|
||||
"""Resolve compression from the bound session without changing its history."""
|
||||
conf = copy.deepcopy(DEFAULT_CONFIG)
|
||||
conf["agent_runner"]["config"]["compression"] = compression or {}
|
||||
ctx = MagicMock()
|
||||
ctx.get_config.return_value = conf
|
||||
stage = InternalAgentSubStage()
|
||||
await stage.initialize(
|
||||
SimpleNamespace(
|
||||
astrbot_config=conf,
|
||||
plugin_manager=SimpleNamespace(context=ctx),
|
||||
)
|
||||
)
|
||||
if compression is None:
|
||||
del conf["agent_runner"]["config"]["compression"]
|
||||
original_conf = copy.deepcopy(conf)
|
||||
|
||||
async def steps(max_step):
|
||||
if False:
|
||||
yield None
|
||||
|
||||
runner = MagicMock()
|
||||
runner.step_until_done.side_effect = steps
|
||||
runner.get_final_llm_resp.return_value = LLMResponse(
|
||||
role="assistant", completion_text="Task completed."
|
||||
)
|
||||
history = [
|
||||
{"role": "user", "content": "An earlier request"},
|
||||
{"role": "assistant", "content": "An earlier answer"},
|
||||
]
|
||||
conv = SimpleNamespace(history=json.dumps(history))
|
||||
manager = CronJobManager(MagicMock())
|
||||
manager.ctx = ctx
|
||||
ctx.get_config.reset_mock()
|
||||
with (
|
||||
patch(
|
||||
"astrbot.core.astr_main_agent._get_session_conv",
|
||||
AsyncMock(return_value=conv),
|
||||
),
|
||||
patch(
|
||||
"astrbot.core.astr_main_agent.build_main_agent",
|
||||
AsyncMock(return_value=SimpleNamespace(agent_runner=runner)),
|
||||
) as build,
|
||||
patch("astrbot.core.cron.manager.persist_agent_history", AsyncMock()),
|
||||
patch("astrbot.core.astr_agent_tool_exec.persist_agent_history", AsyncMock()),
|
||||
):
|
||||
if entrypoint == "cron":
|
||||
await manager._woke_main_agent(
|
||||
message="Run the scheduled task",
|
||||
session_str="test:FriendMessage:user123",
|
||||
extras={"cron_job": {"id": "job-1"}, "cron_payload": {}},
|
||||
)
|
||||
else:
|
||||
ctx.get_llm_tool_manager.return_value.get_builtin_tool.return_value = (
|
||||
FunctionTool(
|
||||
name="send_message_to_user",
|
||||
description="Send the background result.",
|
||||
parameters={"type": "object", "properties": {}},
|
||||
)
|
||||
)
|
||||
await FunctionToolExecutor._wake_main_agent_for_background_result(
|
||||
SimpleNamespace(
|
||||
context=SimpleNamespace(
|
||||
event=SimpleNamespace(
|
||||
unified_msg_origin="test:FriendMessage:user123",
|
||||
role="member",
|
||||
),
|
||||
context=ctx,
|
||||
),
|
||||
tool_call_timeout=120,
|
||||
),
|
||||
task_id="task-1",
|
||||
tool_name="background-tool",
|
||||
result_text="Task completed.",
|
||||
tool_args={},
|
||||
note="Background task finished",
|
||||
summary_name="BackgroundTask",
|
||||
)
|
||||
|
||||
ctx.get_config.assert_called_once_with(umo="test:FriendMessage:user123")
|
||||
wakeup_config = build.await_args.kwargs["config"]
|
||||
fields = (
|
||||
"context_limit_reached_strategy",
|
||||
"llm_compress_instruction",
|
||||
"llm_compress_keep_recent_ratio",
|
||||
"llm_compress_provider_id",
|
||||
"max_context_length",
|
||||
"dequeue_context_length",
|
||||
"fallback_max_context_tokens",
|
||||
)
|
||||
assert tuple(getattr(wakeup_config, name) for name in fields) == expected
|
||||
assert tuple(getattr(stage.main_agent_cfg, name) for name in fields) == expected
|
||||
assert wakeup_config.streaming_response is False
|
||||
request = build.await_args.kwargs["req"]
|
||||
assert request.contexts == history
|
||||
assert "An earlier request" not in request.system_prompt
|
||||
assert conf == original_conf
|
||||
assert json.loads(conv.history) == history
|
||||
Reference in New Issue
Block a user