refactor: redesign context config with orthogonal model - #9441
Open
Rail1bc wants to merge 11 commits into
Open
Conversation
…rigger/disposal model Replace the old implicit-value config (-1 = no limit, <=0 = disabled) with explicit bool switches and a clean separation between trigger conditions (WHEN) and disposal behaviors (WHAT). Core changes: - ContextConfig: 7 old fields → 12 orthogonal fields + summary_provider - ContextManager.process(): independent trigger checks → unified disposal (summary first, discard fallback) → retention constraint → double-check - Add _migra_context_config() + _validate_context_config() migration helpers for zero-downtime config upgrade Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
… agent pipeline - Add new context_management config fields to MainAgentBuildConfig - Update default config with new orthogonal trigger/disposal defaults - Thread context_config through ToolLoopAgentRunner and agent sub-stages - Update existing tests to match new config field names and defaults Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
…ocales Replace old truncate_and_compress keys with new context_management field translations covering all 13 new trigger/disposal fields. Updates zh-CN, en-US, and ru-RU locale files. Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
Add comprehensive test suites for: - ContextConfig validation and field defaults - Context migration helpers (_migra_context_config, _validate_context_config) - MainAgentBuildConfig new context_management fields Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
dosubot
Bot
added
size:XL
📄 Knowledge reviewDosu skipped reviewing this PR because your organization has used its
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 2 issues, and left some high level feedback:
- In ContextManager.init, _discard_compressor is only set in the non-custom_compressor branch, but several tests and potential callers assume it always exists; consider initializing it to None at the top like _summary_compressor/_unity_compressor to avoid attribute errors when a custom compressor is injected.
- In _migra_context_config, when max_context_length <= 0 you drop the old field and only set enable_turn_limit=False without setting a corresponding max_turns; if downstream logic expects max_turns to always be present, you may want to explicitly set it (e.g., to the default) for consistency.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In ContextManager.__init__, _discard_compressor is only set in the non-custom_compressor branch, but several tests and potential callers assume it always exists; consider initializing it to None at the top like _summary_compressor/_unity_compressor to avoid attribute errors when a custom compressor is injected.
- In _migra_context_config, when max_context_length <= 0 you drop the old field and only set enable_turn_limit=False without setting a corresponding max_turns; if downstream logic expects max_turns to always be present, you may want to explicitly set it (e.g., to the default) for consistency.
## Individual Comments
### Comment 1
<location path="tests/unit/test_context_migration.py" line_range="234-241" />
<code_context>
+ # Should not raise
+ _call_validate(ps)
+
+ def test_token_guard_threshold_below_min(self):
+ """token_guard_threshold < 0.5 should warn."""
+ ps_base = {
+ "enable_turn_limit": False,
+ "enable_token_guard": True,
+ "token_guard_threshold": 0.3, # below 0.5
+ "enable_summary": False,
+ "enable_discard": False,
+ "retention_method": "null",
+ }
+ with patch("astrbot.core.utils.migra_helper.logger") as mock_log:
+ _call_validate(ps_base)
+
+ # Should have at least one warning call
</code_context>
<issue_to_address>
**suggestion (testing):** Strengthen validation tests by asserting that logger.warning is actually called
These tests currently only verify that `_validate_context_config` doesn’t raise, but not that it emits warnings for misconfigurations. Since `astrbot.core.utils.migra_helper.logger` is already patched, please add assertions like `mock_log.warning.assert_called()` or `mock_log.warning.assert_any_call(...)` (optionally checking the message contents) so we verify that each invalid config actually produces the expected warning.
```suggestion
with patch("astrbot.core.utils.migra_helper.logger") as mock_log:
_call_validate(ps_base)
# Should emit at least one warning for a too-low threshold
mock_log.warning.assert_called()
```
</issue_to_address>
### Comment 2
<location path="astrbot/core/agent/context/manager.py" line_range="99" />
<code_context>
- result,
- keep_most_recent_turns=self.config.enforce_max_turns,
- drop_turns=self.config.truncate_turns,
+ def _token_guard_exceeded(self, tokens: int, max_context_tokens: int) -> bool:
+ """Return True if token guard is enabled and the ratio exceeds the threshold."""
+ if not self.config.enable_token_guard:
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the token guard logic so `_token_guard_exceeded` is a pure predicate and the enable/misconfiguration handling lives in `process`, keeping `_triggers_fired` focused on trigger evaluation only.
You can simplify the token guard path and reduce cross-calls by making `_token_guard_exceeded` a pure predicate and moving the logging/misconfig handling into `process`. This keeps your orthogonal trigger/disposal model, but removes stateful logging from a predicate and makes the trigger evaluation easier to follow.
### 1. Make `_token_guard_exceeded` pure
Current helper mixes:
- feature flag (`enable_token_guard`)
- configuration validation (`max_context_tokens` <= 0)
- stateful logging (`_token_guard_warning_emitted`)
- the actual predicate
Instead, let it just answer “given a valid guard window, are we over the threshold?”:
```python
def _token_guard_exceeded(self, tokens: int, max_context_tokens: int) -> bool:
"""Pure predicate: return True if tokens exceed the configured threshold."""
if tokens <= 0 or max_context_tokens <= 0:
return False
return (tokens / max_context_tokens) > self.config.token_guard_threshold
```
This removes logging and config checks from the predicate and makes `_triggers_fired`/`process` easier to reason about.
### 2. Centralize token-guard enable & misconfig handling in `process`
Handle `enable_token_guard` and the “max_context_tokens <= 0” misconfig case once at the top of `process`, then pass a normalized `guard_max_tokens` downstream:
```python
async def process(
self,
messages: list[Message],
trusted_token_usage: int = 0,
max_context_tokens: int = 0,
) -> list[Message]:
try:
result = messages
current_tokens = self.token_counter.count_tokens(
result,
trusted_token_usage,
)
total_turns = self._count_turns(result)
guard_max_tokens = 0
if self.config.enable_token_guard:
if max_context_tokens <= 0:
if not getattr(self, "_token_guard_warning_emitted", False):
logger.warning(
"Token guard is enabled but max_context_tokens is %s. "
"Token guarding is effectively disabled. "
"Set max_context_tokens in the provider config to enable it.",
max_context_tokens,
)
self._token_guard_warning_emitted = True
else:
guard_max_tokens = max_context_tokens
if not self._triggers_fired(
total_turns,
current_tokens,
guard_max_tokens,
):
return result
result = await self._select_disposal(result, total_turns)
tokens_after = self.token_counter.count_tokens(result, trusted_token_usage)
if guard_max_tokens and self._token_guard_exceeded(tokens_after, guard_max_tokens):
logger.info(
"Context still exceeds token guard threshold after disposal, "
"applying halving truncation (unconstrained by retention).",
)
result = self.truncator.truncate_by_halving(result)
return result
except Exception:
logger.error("Error during context processing.", exc_info=True)
return messages
```
### 3. Keep `_triggers_fired` focused
With the above, `_triggers_fired` no longer has to worry about config validation or logging; it can purely evaluate triggers, knowing `max_context_tokens` is either a valid guard window or `0` (disabled):
```python
def _triggers_fired(
self,
total_turns: int,
current_tokens: int,
max_context_tokens: int,
) -> bool:
"""Return True if any trigger condition is met."""
if self.config.enable_turn_limit and total_turns > self.config.max_turns:
return True
if self.config.enable_token_guard and self._token_guard_exceeded(
current_tokens, max_context_tokens
):
return True
return False
```
This refactor:
- keeps all existing features (turn limit, token guard, double-check, retention rules)
- removes stateful logging from a predicate helper
- consolidates token-guard config handling into one place, reducing branching and cross-calls without reverting the new orthogonal design.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
…_management The ru-RU locale file was previously committed with Python json.dump output, which converted it from upstream's CRLF + 4-space indent to LF + 2-space indent, causing a 3500-line diff. This restores upstream's exact formatting conventions and applies only the semantic changes: remove truncate_and_compress, add context_management block. Effective diff is now 101 lines. Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
…ameter The ContextManager.process() now passes max_context_tokens as a keyword argument. Two test mocks in test_tool_loop_agent_runner.py did not accept this parameter, causing TypeError in CI. Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
"lower bound on how many turns discard may remove" is ambiguous. Rephrased as "lower bound on how many turns remain after discard" to match the actual semantics (retain_turns = minimum turns kept). Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
… assertions 1. Remove unused _discard_compressor assignment in ContextManager.__init__ (discard is handled directly via self.truncator, not a compressor). 2. Set explicit max_turns=50 in migration else branch when old max_context_length was <= 0, for dict consistency. 3. Strengthen validation tests: remove `or True` no-op assertion, add mock_log.warning.assert_called() to all 8 _validate_context_config tests. Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
…rd handling Per review suggestion: _token_guard_exceeded no longer checks enable_token_guard or logs misconfiguration — it's now a pure ratio comparator. The enable check and misconfig logging move to process(), which resolves guard_max_tokens before passing it to _triggers_fired and the double-check. Also clean up test_context_manager.py: remove _discard_compressor assertion from test_init_with_minimal_config; skip 9 tests that reference the removed attribute (patterns covered by test_context_manager_new.py). Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
Two lines exceeded 88-char limit after refactoring. Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
Rail1bc
force-pushed
the
pr/1-core-integration
branch
from
July 29, 2026 10:20
f2754df to
c2c0b5c
Compare
Contributor
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - I've found 3 issues, and left some high level feedback:
- The new context management fields are now defined and wired in several places (ContextConfig, MainAgentBuildConfig, default provider_settings, pipeline internal stage, tool loop runner); consider centralizing a single source of truth or helper for constructing/validating these configs to avoid future drift between modules.
- In _migra_context_config, when max_context_length <= 0 you silently disable turn-limit and set max_turns=50; if the intent is to preserve the old "unlimited" semantics, it may be less surprising to leave max_turns untouched or explicitly document that migration changes this behavior.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The new context management fields are now defined and wired in several places (ContextConfig, MainAgentBuildConfig, default provider_settings, pipeline internal stage, tool loop runner); consider centralizing a single source of truth or helper for constructing/validating these configs to avoid future drift between modules.
- In _migra_context_config, when max_context_length <= 0 you silently disable turn-limit and set max_turns=50; if the intent is to preserve the old "unlimited" semantics, it may be less surprising to leave max_turns untouched or explicitly document that migration changes this behavior.
## Individual Comments
### Comment 1
<location path="astrbot/core/agent/context/manager.py" line_range="74-75" />
<code_context>
+ if len(result) >= len(messages):
+ return None # no effective reduction
+ return result
+ except Exception:
+ logger.warning(
+ "LLM summary compression failed, falling back.", exc_info=True
)
</code_context>
<issue_to_address>
**issue (bug_risk):** Avoid swallowing cancellation exceptions in summary compression.
Using a blanket `except Exception:` here will also catch `asyncio.CancelledError` (which may inherit from `Exception` depending on the environment), breaking cooperative cancellation when this function is part of a larger task chain. Cancellation should propagate rather than be treated as a recoverable error. For example:
```python
except asyncio.CancelledError:
raise
except Exception:
logger.warning(
"LLM summary compression failed, falling back.", exc_info=True
)
return None
```
The same pattern should be applied in the broader `process` method so that cancellation aborts the run instead of being handled as a normal error.
</issue_to_address>
### Comment 2
<location path="tests/unit/test_context_migration.py" line_range="224-233" />
<code_context>
+ # Should not raise
+ _call_validate(ps)
+
+ def test_token_guard_threshold_below_min(self):
+ """token_guard_threshold < 0.5 should warn."""
+ ps_base = {
+ "enable_turn_limit": False,
+ "enable_token_guard": True,
+ "token_guard_threshold": 0.3, # below 0.5
+ "enable_summary": False,
+ "enable_discard": False,
+ "retention_method": "null",
+ }
+ with patch("astrbot.core.utils.migra_helper.logger") as mock_log:
+ _call_validate(ps_base)
+
+ mock_log.warning.assert_called()
+
+ def test_token_guard_threshold_above_max(self):
</code_context>
<issue_to_address>
**suggestion (testing):** Add tests for non-numeric token_guard_threshold/retain_percentage cases covered by _validate_context_config
The validator also warns when `token_guard_threshold` or `retain_percentage` are non-numeric (e.g. strings), but those branches aren’t currently tested. Please add cases like:
- `token_guard_threshold="0.8"` and assert the non-numeric warning.
- `retention_method="percentage"` with `retain_percentage` as a string and assert the warning.
This will cover the type-check paths for misconfigured input.
Suggested implementation:
```python
def test_token_guard_threshold_below_min(self):
"""token_guard_threshold < 0.5 should warn."""
ps_base = {
"enable_turn_limit": False,
"enable_token_guard": True,
"token_guard_threshold": 0.3, # below 0.5
"enable_summary": False,
"enable_discard": False,
"retention_method": "null",
}
with patch("astrbot.core.utils.migra_helper.logger") as mock_log:
_call_validate(ps_base)
mock_log.warning.assert_called()
def test_token_guard_threshold_above_max(self):
"""token_guard_threshold > 0.99 should warn."""
ps = {
"enable_token_guard": True,
"token_guard_threshold": 1.5,
}
with patch("astrbot.core.utils.migra_helper.logger") as mock_log:
_call_validate(ps)
mock_log.warning.assert_called()
def test_token_guard_threshold_non_numeric_warns(self):
"""Non-numeric token_guard_threshold should warn."""
ps = {
"enable_turn_limit": False,
"enable_token_guard": True,
"token_guard_threshold": "0.8", # string instead of number
"enable_summary": False,
"enable_discard": False,
"retention_method": "null",
}
with patch("astrbot.core.utils.migra_helper.logger") as mock_log:
_call_validate(ps)
mock_log.warning.assert_called()
def test_retain_percentage_non_numeric_warns(self):
"""Non-numeric retain_percentage with percentage retention_method should warn."""
ps = {
"enable_turn_limit": False,
"enable_token_guard": False,
"enable_summary": False,
"enable_discard": True,
"discard_turns": 1,
"retention_method": "percentage",
"retain_percentage": "0.3", # string instead of number
}
with patch("astrbot.core.utils.migra_helper.logger") as mock_log:
_call_validate(ps)
mock_log.warning.assert_called()
```
- Ensure `patch` is imported from `unittest.mock` at the top of `tests/unit/test_context_migration.py` (it likely already is, since existing tests use it).
- If `_validate_context_config` emits more specific warning messages (e.g., mentioning "non-numeric"), you may tighten the assertions with `mock_log.warning.assert_called_once()` or `assert_any_call(...)` matching the expected message.
</issue_to_address>
### Comment 3
<location path="astrbot/core/agent/context/manager.py" line_range="45" />
<code_context>
+ drop_turns=requested,
+ )
- async def process(
- self, messages: list[Message], trusted_token_usage: int = 0
- ) -> list[Message]:
</code_context>
<issue_to_address>
**issue (complexity):** Consider centralizing trigger evaluation and disposal/halving into a single structured pipeline to reduce duplication and make the control flow easier to understand and extend.
You can reduce the new complexity without losing any behavior by centralizing trigger evaluation and disposal/halving into a single pipeline. Two focused refactors:
---
### 1. Centralize trigger evaluation
Right now:
- `process` resolves `guard_max_tokens` and calls `_triggers_fired`.
- `_triggers_fired` recomputes token guard using `_token_guard_exceeded`.
- Halving reuses `_token_guard_exceeded` with ad-hoc logic.
You can replace `_triggers_fired` + the scattered guard logic with a single structured evaluator that returns *why* it fired and the relevant thresholds, so you don’t recompute or scatter responsibilities.
```python
from dataclasses import dataclass
@dataclass
class TriggerEvaluation:
any_fired: bool
turn_limit: bool
token_guard: bool
guard_max_tokens: int
total_turns: int
tokens: int
def _evaluate_triggers(
self,
messages: list[Message],
trusted_token_usage: int,
max_context_tokens: int,
) -> TriggerEvaluation:
tokens = self.token_counter.count_tokens(messages, trusted_token_usage)
total_turns = self._count_turns(messages)
guard_max_tokens = 0
token_guard_fired = False
if self.config.enable_token_guard:
if max_context_tokens <= 0:
if not getattr(self, "_token_guard_warning_emitted", False):
logger.warning(
"Token guard is enabled but max_context_tokens is %s. "
"Token guarding is effectively disabled. "
"Set max_context_tokens in the provider config to enable it.",
max_context_tokens,
)
self._token_guard_warning_emitted = True
else:
guard_max_tokens = max_context_tokens
token_guard_fired = self._token_guard_exceeded(tokens, guard_max_tokens)
turn_limit_fired = (
self.config.enable_turn_limit
and total_turns > self.config.max_turns
)
any_fired = turn_limit_fired or token_guard_fired
return TriggerEvaluation(
any_fired=any_fired,
turn_limit=turn_limit_fired,
token_guard=token_guard_fired,
guard_max_tokens=guard_max_tokens,
total_turns=total_turns,
tokens=tokens,
)
```
Then `process` can be simplified and more self‑documenting:
```python
async def process(
self,
messages: list[Message],
trusted_token_usage: int = 0,
max_context_tokens: int = 0,
) -> list[Message]:
try:
result = messages
eval = self._evaluate_triggers(
result,
trusted_token_usage=trusted_token_usage,
max_context_tokens=max_context_tokens,
)
if not eval.any_fired:
return result
result = await self._apply_disposal_pipeline(result, eval)
return result
except Exception:
logger.error("Error during context processing.", exc_info=True)
return messages
```
This keeps turn/token logic in one place, makes tests simpler (you can assert on `TriggerEvaluation`), and removes duplicated guard computations.
---
### 2. Encapsulate disposal + halving in one pipeline
Currently:
- `_select_disposal` chooses custom/summary/discard.
- `process` is responsible for halving, using `_token_guard_exceeded` again.
- `_select_disposal` doesn’t know about the guard, and the guard doesn’t know about disposal.
You can fold halving into the disposal pipeline so that all disposal behavior is in one function, driven by the `TriggerEvaluation`:
```python
async def _apply_disposal_pipeline(
self,
messages: list[Message],
eval: TriggerEvaluation,
) -> list[Message]:
# 1. Custom compressor takes full control if present.
if self._unity_compressor is not None:
result = await self._unity_compressor(messages)
else:
# 2. Summary, then discard.
total_turns = eval.total_turns
result = messages
if self.config.enable_summary and self._summary_compressor is not None:
compressed = await self._try_summary(result)
if compressed is not None:
result = compressed
if result is messages and self.config.enable_discard:
discarded = self._try_discard(result, total_turns=total_turns)
if discarded is not None:
result = discarded
if result is messages:
logger.warning(
"Context disposal triggered but both summary and discard "
"are unavailable or disabled. No compression applied.",
)
# 3. Halving as a final guard-enforced step.
if (
eval.guard_max_tokens
and self._token_guard_exceeded(
self.token_counter.count_tokens(result, 0),
eval.guard_max_tokens,
)
):
logger.info(
"Context still exceeds token guard threshold after disposal, "
"applying halving truncation (unconstrained by retention).",
)
result = self.truncator.truncate_by_halving(result)
return result
```
With this:
- Halving is clearly part of the disposal pipeline.
- `process` no longer needs to know about halving at all.
- Guard behavior and disposal behavior are coherently coupled via `TriggerEvaluation`.
This keeps all existing features (orthogonal triggers, retention, fallback hierarchy, halving) but makes the control flow easier to follow and localizes future changes (e.g., adding new triggers or disposal modes) into `TriggerEvaluation` and `_apply_disposal_pipeline`.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+224
to
+233
| def test_token_guard_threshold_below_min(self): | ||
| """token_guard_threshold < 0.5 should warn.""" | ||
| ps_base = { | ||
| "enable_turn_limit": False, | ||
| "enable_token_guard": True, | ||
| "token_guard_threshold": 0.3, # below 0.5 | ||
| "enable_summary": False, | ||
| "enable_discard": False, | ||
| "retention_method": "null", | ||
| } |
Contributor
There was a problem hiding this comment.
suggestion (testing): Add tests for non-numeric token_guard_threshold/retain_percentage cases covered by _validate_context_config
The validator also warns when token_guard_threshold or retain_percentage are non-numeric (e.g. strings), but those branches aren’t currently tested. Please add cases like:
token_guard_threshold="0.8"and assert the non-numeric warning.retention_method="percentage"withretain_percentageas a string and assert the warning.
This will cover the type-check paths for misconfigured input.
Suggested implementation:
def test_token_guard_threshold_below_min(self):
"""token_guard_threshold < 0.5 should warn."""
ps_base = {
"enable_turn_limit": False,
"enable_token_guard": True,
"token_guard_threshold": 0.3, # below 0.5
"enable_summary": False,
"enable_discard": False,
"retention_method": "null",
}
with patch("astrbot.core.utils.migra_helper.logger") as mock_log:
_call_validate(ps_base)
mock_log.warning.assert_called()
def test_token_guard_threshold_above_max(self):
"""token_guard_threshold > 0.99 should warn."""
ps = {
"enable_token_guard": True,
"token_guard_threshold": 1.5,
}
with patch("astrbot.core.utils.migra_helper.logger") as mock_log:
_call_validate(ps)
mock_log.warning.assert_called()
def test_token_guard_threshold_non_numeric_warns(self):
"""Non-numeric token_guard_threshold should warn."""
ps = {
"enable_turn_limit": False,
"enable_token_guard": True,
"token_guard_threshold": "0.8", # string instead of number
"enable_summary": False,
"enable_discard": False,
"retention_method": "null",
}
with patch("astrbot.core.utils.migra_helper.logger") as mock_log:
_call_validate(ps)
mock_log.warning.assert_called()
def test_retain_percentage_non_numeric_warns(self):
"""Non-numeric retain_percentage with percentage retention_method should warn."""
ps = {
"enable_turn_limit": False,
"enable_token_guard": False,
"enable_summary": False,
"enable_discard": True,
"discard_turns": 1,
"retention_method": "percentage",
"retain_percentage": "0.3", # string instead of number
}
with patch("astrbot.core.utils.migra_helper.logger") as mock_log:
_call_validate(ps)
mock_log.warning.assert_called()- Ensure
patchis imported fromunittest.mockat the top oftests/unit/test_context_migration.py(it likely already is, since existing tests use it). - If
_validate_context_configemits more specific warning messages (e.g., mentioning "non-numeric"), you may tighten the assertions withmock_log.warning.assert_called_once()orassert_any_call(...)matching the expected message.
| def _count_turns(self, messages: list[Message]) -> int: | ||
| """Count the number of conversation turns (non-system rounds).""" | ||
| rounds = split_into_rounds(messages) | ||
| # Filter out rounds that are exclusively system messages |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider centralizing trigger evaluation and disposal/halving into a single structured pipeline to reduce duplication and make the control flow easier to understand and extend.
You can reduce the new complexity without losing any behavior by centralizing trigger evaluation and disposal/halving into a single pipeline. Two focused refactors:
1. Centralize trigger evaluation
Right now:
processresolvesguard_max_tokensand calls_triggers_fired._triggers_firedrecomputes token guard using_token_guard_exceeded.- Halving reuses
_token_guard_exceededwith ad-hoc logic.
You can replace _triggers_fired + the scattered guard logic with a single structured evaluator that returns why it fired and the relevant thresholds, so you don’t recompute or scatter responsibilities.
from dataclasses import dataclass
@dataclass
class TriggerEvaluation:
any_fired: bool
turn_limit: bool
token_guard: bool
guard_max_tokens: int
total_turns: int
tokens: int
def _evaluate_triggers(
self,
messages: list[Message],
trusted_token_usage: int,
max_context_tokens: int,
) -> TriggerEvaluation:
tokens = self.token_counter.count_tokens(messages, trusted_token_usage)
total_turns = self._count_turns(messages)
guard_max_tokens = 0
token_guard_fired = False
if self.config.enable_token_guard:
if max_context_tokens <= 0:
if not getattr(self, "_token_guard_warning_emitted", False):
logger.warning(
"Token guard is enabled but max_context_tokens is %s. "
"Token guarding is effectively disabled. "
"Set max_context_tokens in the provider config to enable it.",
max_context_tokens,
)
self._token_guard_warning_emitted = True
else:
guard_max_tokens = max_context_tokens
token_guard_fired = self._token_guard_exceeded(tokens, guard_max_tokens)
turn_limit_fired = (
self.config.enable_turn_limit
and total_turns > self.config.max_turns
)
any_fired = turn_limit_fired or token_guard_fired
return TriggerEvaluation(
any_fired=any_fired,
turn_limit=turn_limit_fired,
token_guard=token_guard_fired,
guard_max_tokens=guard_max_tokens,
total_turns=total_turns,
tokens=tokens,
)Then process can be simplified and more self‑documenting:
async def process(
self,
messages: list[Message],
trusted_token_usage: int = 0,
max_context_tokens: int = 0,
) -> list[Message]:
try:
result = messages
eval = self._evaluate_triggers(
result,
trusted_token_usage=trusted_token_usage,
max_context_tokens=max_context_tokens,
)
if not eval.any_fired:
return result
result = await self._apply_disposal_pipeline(result, eval)
return result
except Exception:
logger.error("Error during context processing.", exc_info=True)
return messagesThis keeps turn/token logic in one place, makes tests simpler (you can assert on TriggerEvaluation), and removes duplicated guard computations.
2. Encapsulate disposal + halving in one pipeline
Currently:
_select_disposalchooses custom/summary/discard.processis responsible for halving, using_token_guard_exceededagain._select_disposaldoesn’t know about the guard, and the guard doesn’t know about disposal.
You can fold halving into the disposal pipeline so that all disposal behavior is in one function, driven by the TriggerEvaluation:
async def _apply_disposal_pipeline(
self,
messages: list[Message],
eval: TriggerEvaluation,
) -> list[Message]:
# 1. Custom compressor takes full control if present.
if self._unity_compressor is not None:
result = await self._unity_compressor(messages)
else:
# 2. Summary, then discard.
total_turns = eval.total_turns
result = messages
if self.config.enable_summary and self._summary_compressor is not None:
compressed = await self._try_summary(result)
if compressed is not None:
result = compressed
if result is messages and self.config.enable_discard:
discarded = self._try_discard(result, total_turns=total_turns)
if discarded is not None:
result = discarded
if result is messages:
logger.warning(
"Context disposal triggered but both summary and discard "
"are unavailable or disabled. No compression applied.",
)
# 3. Halving as a final guard-enforced step.
if (
eval.guard_max_tokens
and self._token_guard_exceeded(
self.token_counter.count_tokens(result, 0),
eval.guard_max_tokens,
)
):
logger.info(
"Context still exceeds token guard threshold after disposal, "
"applying halving truncation (unconstrained by retention).",
)
result = self.truncator.truncate_by_halving(result)
return resultWith this:
- Halving is clearly part of the disposal pipeline.
processno longer needs to know about halving at all.- Guard behavior and disposal behavior are coherently coupled via
TriggerEvaluation.
This keeps all existing features (orthogonal triggers, retention, fallback hierarchy, halving) but makes the control flow easier to follow and localizes future changes (e.g., adding new triggers or disposal modes) into TriggerEvaluation and _apply_disposal_pipeline.
…t coverage 1. Add doc comment explaining why max_turns=50 is set even when enable_turn_limit=False (preserves old "unlimited" semantics). 2. Propagate asyncio.CancelledError in _try_summary and process() instead of swallowing it under bare except Exception. 3. Add two tests for non-numeric token_guard_threshold and retain_percentage validation warnings. Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation / 动机
旧版上下文管理配置使用魔法值（
-1= 无限制、0= 关闭）来表达启用/禁用，导致"未配置"与"显式禁用"无法区分。压缩管线每轮无条件执行截断，触发条件与处置行为混杂在同一代码路径中，行为难以预测。本次重构将上下文配置拆分（触发 / 处置 / 保留），用 7 个隐式值字段替换为 12 个显式字段。提供迁移辅助函数
_migra_context_config()将已有用户配置无缝转换。Modifications / 改动点
核心模型 —
astrbot/core/agent/context/config.py新的
ContextConfig数据类，三个独立维度：enable_turn_limit+max_turnsenable_token_guard+token_guard_thresholdenable_summary/summary_prompt/summary_providerenable_discard/discard_turnsretention_method/retain_turns/retain_percentage核心管线 —
astrbot/core/agent/context/manager.py重写
ContextManager.process()：_triggers_fired()— 独立检查两个触发条件，都不满足则零开销跳过_select_disposal()— 处置优先级：custom → summary → discard（逐级 fallthrough）_compute_discard_limit()— 保留约束计算最大可丢弃量enable_token_guard=True时触发，处置后仍超限则无条件折半接入层：
astrbot/core/astr_main_agent.py—MainAgentBuildConfig新增 12 个正交字段，从provider_settings读取astrbot/core/config/default.py— 默认provider_settings替换为新字段astrbot/core/agent/runners/tool_loop_agent_runner.py— Tool loop 使用新字段迁移辅助 —
astrbot/core/utils/migra_helper.py_migra_context_config(ps)将旧配置自动转换为新格式，旧→新映射表：max_context_lengthenable_turn_limit+max_turnsdequeue_context_lengthdiscard_turnscontext_limit_reached_strategyenable_summary+enable_discardllm_compress_instructionsummary_promptllm_compress_keep_recent_ratioretain_percentage+retention_method='percentage'llm_compress_provider_idsummary_provider_id_validate_context_config(ps)校验约束违规（如token_guard_threshold超出 0.5~0.99、两处置均关闭等）。国际化：
dashboard/src/i18n/locales/{en-US,zh-CN,ru-RU}/features/config-metadata.json—truncate_and_compress块替换为context_management块（63 行，覆盖 12 个新字段），保留上游原有格式（ru-RU 保持 CRLF + 4 空格缩进）测试（43 个新测试，全部通过）：
tests/agent/test_context_config_new.py(17) — 默认值、自定义压缩器注入、显式构造、向后兼容tests/unit/test_context_migration.py(22) — 6 种旧→新字段映射 + 8 种校验约束tests/unit/test_main_agent_build_config_new.py(4) — MainAgentBuildConfig 新字段默认值与构造函数传参This is NOT a breaking change. / 这不是一个破坏性变更。
Test Results / 测试结果
通过
-m "not tier_c and not tier_d"全量 blocking 测试共 1951 passed，36 failed（均为 Windows 路径分隔符等预先存在的环境问题，与本 PR 无关）。Checklist / 检查清单
Summary by Sourcery
Redesign context management into an orthogonal trigger/disposal/retention model and wire it through agents, configs, and runners with backward-compatible migration and validation.
Enhancements:
Tests: