refactor: redesign context config with orthogonal model and add /compact command#9340
Closed
Rail1bc wants to merge 9 commits into
Closed
refactor: redesign context config with orthogonal model and add /compact command#9340Rail1bc wants to merge 9 commits into
Rail1bc wants to merge 9 commits into
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 - MainAgentBuildConfig: new fields added, old fields marked deprecated - provider_settings defaults replaced with new orthogonal defaults - _migra_context_config() + _validate_context_config() wired into migra() Dashboard i18n: all 3 locale files (zh-CN/en-US/ru-RU) updated from truncate_and_compress to context_management with 13 new field translations. Tests: 67 new tests (ContextConfig, ContextManager, migration, build config) plus 9 old-API tests skipped. 143 pass, 0 fail. Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
Add a /compact command that manually triggers ContextManager.process() on the current conversation, applying the orthogonal trigger/disposal model (summary first, discard fallback, retention bound, double-check) without fully clearing the conversation history. Permission follows the same RstScene-based system as /reset: admin-required in group non-unique sessions by default. Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
dosubot
Bot
added
size:XL
![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 5 issues, and left some high level feedback:
- In the new
/compactcommand, conversation history is reconstructed using onlyroleandcontent, which will drop other metadata (e.g. tool calls, names, additional fields); consider preserving the full message schema to avoid losing context-specific information on compression. - The
/compactcommand builds aContextManagerwithout passing amax_context_tokensvalue intoprocess, which effectively disables the token-guard trigger there; if the intent is to respect the model’s context window during manual compaction, you may want to fetch the provider’smax_context_tokensand pass it through.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In the new `/compact` command, conversation history is reconstructed using only `role` and `content`, which will drop other metadata (e.g. tool calls, names, additional fields); consider preserving the full message schema to avoid losing context-specific information on compression.
- The `/compact` command builds a `ContextManager` without passing a `max_context_tokens` value into `process`, which effectively disables the token-guard trigger there; if the intent is to respect the model’s context window during manual compaction, you may want to fetch the provider’s `max_context_tokens` and pass it through.
## Individual Comments
### Comment 1
<location path="tests/unit/test_context_migration.py" line_range="190-108" />
<code_context>
+ ):
+ assert old_key not in ps, f"{old_key} should have been removed"
+
+ def test_retain_percentage_zero_uses_turns(self):
+ """llm_compress_keep_recent_ratio=0 → retain_percentage=0, retention_method stays 'turns'."""
+ ps = make_old_settings(llm_compress_keep_recent_ratio=0)
+ result = _call_migra(ps)
+
+ assert result is True
+ assert ps.get("retain_percentage") == 0
+ # When ratio is 0, it's ambiguous; the migration may or may not
+ # change retention_method. Both outcomes are reasonable.
+ assert "llm_compress_keep_recent_ratio" not in ps
+
+
</code_context>
<issue_to_address>
**issue (testing):** Align test expectations with actual migration behavior for retain_percentage=0
The test docstring claims `retention_method` "stays 'turns'", but `_migra_context_config` currently sets `retention_method = 'percentage'` whenever `llm_compress_keep_recent_ratio` is present, even when it is 0. Since the test never asserts on `retention_method`, its description and name are misleading. Please either (a) add an assertion that `retention_method` becomes `'percentage'` for ratio=0, or (b) change the migration logic so that ratio=0 keeps `'turns'` and assert that behavior.
</issue_to_address>
### Comment 2
<location path="astrbot/core/agent/context/manager.py" line_range="143" />
<code_context>
+ if not self._triggers_fired(result, current_tokens, max_context_tokens):
+ return result
+
+ # 2. 处置入口：custom > summary > discard
+ if hasattr(self, "_unity_compressor"):
+ result = await self._unity_compressor(result)
</code_context>
<issue_to_address>
**issue (complexity):** Consider explicitly initializing compressor fields and consolidating trigger checks into a single helper to simplify the disposal flow and reduce indirection.
You can reduce the indirection without changing behavior by (1) making compressor selection explicit (no `hasattr` checks) and (2) collapsing trigger checks into a single `_should_dispose` helper that lives next to `process`.
### 1. Make compressor selection explicit
Instead of relying on `hasattr(self, "_unity_compressor")`, initialize explicit fields and use `is not None` checks:
```python
class ContextManager:
def __init__(self, config: ContextConfig) -> None:
self.config = config
self.token_counter = config.custom_token_counter or EstimateTokenCounter()
self.truncator = ContextTruncator()
self._unity_compressor = None
self._summary_compressor = None
self._discard_compressor = None
if config.custom_compressor:
self._unity_compressor = config.custom_compressor
else:
if config.summary_provider:
self._summary_compressor = LLMSummaryCompressor(
provider=config.summary_provider,
keep_recent_ratio=config.retain_percentage,
instruction_text=config.summary_prompt,
compression_threshold=config.token_guard_threshold,
token_counter=self.token_counter,
)
self._discard_compressor = TruncateByTurnsCompressor(
truncate_turns=config.discard_turns,
compression_threshold=config.token_guard_threshold,
)
```
Then in `process`:
```python
# 2. 处置入口：custom > summary > discard
if self._unity_compressor is not None:
result = await self._unity_compressor(result)
else:
compressed = await self._try_summary(result)
if compressed is not None:
result = compressed
else:
discarded = self._try_discard(result)
...
```
This keeps the same priority chain but removes implicit existence checks and makes the object graph easier to reason about.
### 2. Collapse trigger logic into a single `_should_dispose` helper
You can inline `_triggers_fired` into a clearer `_should_dispose` that sits next to `process`, so the flow from tokens/turns → disposal is readable in one place:
```python
def _should_dispose(
self,
messages: list[Message],
current_tokens: int,
max_context_tokens: int,
) -> bool:
# turn-limit trigger
if self.config.enable_turn_limit:
if self._count_turns(messages) > self.config.max_turns:
return True
# token-guard trigger
if self.config.enable_token_guard and max_context_tokens > 0 and current_tokens > 0:
ratio = current_tokens / max_context_tokens
return ratio > self.config.token_guard_threshold
return False
```
And in `process`:
```python
current_tokens = self.token_counter.count_tokens(result, trusted_token_usage)
if not self._should_dispose(result, current_tokens, max_context_tokens):
return result
```
This keeps the orthogonal trigger/disposal model but reduces “mental jumping” between helpers: `process` now clearly shows `count_tokens → should_dispose? → disposal path → optional halving`, and the policy is encapsulated in one obvious method.
</issue_to_address>
### Comment 3
<location path="astrbot/core/utils/migra_helper.py" line_range="196" />
<code_context>
+ logger.error(traceback.format_exc())
+
+
+def _migra_context_config(ps: dict) -> bool:
+ """Migrate old context config fields to new orthogonal fields.
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the migration and validation helpers into small, declarative sub-functions so each rule is isolated, testable, and easier to extend.
You can reduce the cognitive load of both functions by making the migration rules and validation rules more declarative and localized, while preserving all behavior.
### 1. Make `_migra_context_config` rule-based
Right now each legacy field is handled via an independent `if` that mutates `ps`. You can encapsulate each mapping as a small function and drive it from a list, which makes the overall mapping clearer and easier to extend:
```python
def _migra_context_config(ps: dict) -> bool:
"""Migrate old context config fields to new orthogonal fields."""
migrated = False
def _migrate_max_context_length(ps: dict) -> bool:
if "max_context_length" not in ps:
return False
old_val = ps.pop("max_context_length")
if old_val > 0:
ps["enable_turn_limit"] = True
ps["max_turns"] = old_val
else:
ps["enable_turn_limit"] = False
return True
def _migrate_dequeue_context_length(ps: dict) -> bool:
if "dequeue_context_length" not in ps:
return False
ps["discard_turns"] = ps.pop("dequeue_context_length")
return True
def _migrate_context_limit_strategy(ps: dict) -> bool:
if "context_limit_reached_strategy" not in ps:
return False
strategy = ps.pop("context_limit_reached_strategy")
ps["enable_discard"] = True
ps["enable_summary"] = (strategy == "llm_compress")
return True
# ...small mappers for each legacy field...
for mapper in (
_migrate_max_context_length,
_migrate_dequeue_context_length,
_migrate_context_limit_strategy,
# _migrate_llm_compress_instruction,
# _migrate_llm_compress_keep_recent_ratio,
# _migrate_llm_compress_provider_id,
):
if mapper(ps):
migrated = True
if migrated:
logger.info("Migrated old context config to orthogonal trigger/disposal model.")
return migrated
```
This keeps all side effects but makes each legacy field’s transformation isolated and testable, and the “overall mapping” is visible in the mapper list.
### 2. Split `_validate_context_config` by concern
You can keep the same warnings/rules but group them into small helpers. This reduces nesting and repeated `_has` calls, and makes future additions easier to place:
```python
def _validate_context_config(ps: dict) -> None:
"""Validate new context config fields and log warnings for violations."""
def _has(k: str) -> bool:
return k in ps
def _validate_turn_limit():
if _has("enable_turn_limit") and _has("max_turns"):
if ps.get("enable_turn_limit") and ps.get("max_turns", 50) < 2:
logger.warning("max_turns should be >= 2 when enable_turn_limit is True.")
def _validate_token_guard():
if _has("token_guard_threshold"):
val = ps["token_guard_threshold"]
if not (0.5 <= val <= 0.99):
logger.warning(
"token_guard_threshold %.2f is outside recommended range [0.5, 0.99].",
val,
)
def _validate_discard():
if _has("discard_turns") and ps.get("discard_turns", 1) < 1:
logger.warning("discard_turns should be >= 1.")
if _has("enable_summary") and _has("enable_discard"):
if not ps.get("enable_summary") and not ps.get("enable_discard"):
logger.warning(
"Both enable_summary and enable_discard are False. "
"Context will not be compressed when triggers fire.",
)
def _validate_retention():
if not _has("retention_method"):
return
method = ps["retention_method"]
if method not in ("turns", "percentage", "null"):
logger.warning("Unknown retention_method '%s'. Use 'turns', 'percentage', or 'null'.", method)
return
if method == "turns" and _has("retain_turns") and ps.get("retain_turns", 20) < 1:
logger.warning("retain_turns should be >= 1.")
if method == "percentage" and _has("retain_percentage"):
val = ps["retain_percentage"]
if not (0.1 <= val <= 0.9):
logger.warning(
"retain_percentage %.2f is outside recommended range [0.1, 0.9].",
val,
)
if (
ps.get("enable_turn_limit")
and method == "turns"
and _has("retain_turns")
and _has("max_turns")
and ps["retain_turns"] > ps["max_turns"]
):
logger.warning(
"retain_turns (%d) > max_turns (%d). Retention lower bound will take priority.",
ps["retain_turns"],
ps["max_turns"],
)
_validate_turn_limit()
_validate_token_guard()
_validate_discard()
_validate_retention()
```
This preserves all current behavior and logging but makes each rule set (turn limit, token guard, disposal, retention) easier to reason about and extend.
</issue_to_address>
### Comment 4
<location path="astrbot/builtin_stars/builtin_commands/commands/conversation.py" line_range="196" />
<code_context>
message.set_result(MessageEventResult().message(ret))
+ async def compact(self, message: AstrMessageEvent) -> None:
+ """手动触发上下文的处置与压缩"""
+ umo = message.unified_msg_origin
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting shared permission checks, context configuration building, and history/message conversion into reusable helpers so the new `compact` method focuses on high-level flow and avoids duplicated logic.
The new `compact` method is doing multiple distinct things and duplicates logic from `reset`, which increases complexity and maintenance cost. You can keep all functionality but reduce complexity via small extractions:
### 1. Shared permission + scene resolution helper
The permission check is almost identical to `reset`. Extract it and reuse:
```python
# helpers.py (or near RstScene definition)
async def check_compact_permission(message: AstrMessageEvent, is_group: bool, is_unique_session: bool) -> Optional[str]:
scene = RstScene.get_scene(is_group, is_unique_session)
alter_cmd_cfg = await sp.get_async("global", "global", "alter_cmd", {})
plugin_config = alter_cmd_cfg.get("astrbot", {})
compact_cfg = plugin_config.get("compact", {})
required_perm = compact_cfg.get(
scene.key,
"admin" if is_group and not is_unique_session else "member",
)
if required_perm == "admin" and message.role != "admin":
return (
f"Compact command requires admin permission in {scene.name} scenario, "
f"you (ID {message.get_sender_id()}) are not admin, cannot perform this action."
)
return None
```
Usage in `compact`:
```python
is_group = bool(message.get_group_id())
is_unique_session = cfg["platform_settings"]["unique_session"]
err = await check_compact_permission(message, is_group, is_unique_session)
if err:
message.set_result(MessageEventResult().message(err))
return
```
This keeps the permission logic centralized and makes both `reset` and `compact` easier to adjust.
### 2. ContextConfig factory
Config construction is inlined and likely duplicates other places where `ContextConfig` is built. Extract a factory:
```python
# context_config_factory.py
from astrbot.core.agent.context.config import ContextConfig
def build_context_config_from_settings(settings: dict, summary_provider) -> ContextConfig:
return ContextConfig(
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider=summary_provider,
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
)
```
Usage in `compact`:
```python
summary_provider_id = settings.get("summary_provider_id", "")
summary_provider = (
self.context.get_provider_by_id(summary_provider_id)
if summary_provider_id
else self.context.get_using_provider(umo=umo)
)
config = build_context_config_from_settings(settings, summary_provider)
cm = ContextManager(config)
```
This removes config noise from `compact` and aligns behavior with other agent/context initialization paths.
### 3. History ↔ Message conversion utilities
The manual loops for converting between `dict` and `Message` are verbose and type-specific. Extract tiny helpers:
```python
# history_utils.py
import json
from typing import List
from astrbot.core.agent.message import Message # adjust import as needed
def conversation_history_to_messages(history_raw) -> List[Message]:
if isinstance(history_raw, str):
history: list[dict] = json.loads(history_raw or "[]")
else:
history = history_raw or []
return [
Message(role=item.get("role", "user"), content=item.get("content", ""))
for item in history
]
def messages_to_history_dict(messages: List[Message]) -> list[dict]:
result: list[dict] = []
for msg in messages:
if isinstance(msg.content, str):
content = msg.content
else:
content = str(msg.content) if msg.content else ""
result.append({"role": msg.role, "content": content})
return result
```
Usage in `compact`:
```python
conv = await self.context.conversation_manager.get_conversation(umo, cid)
messages = conversation_history_to_messages(conv.history)
if not messages:
message.set_result(
MessageEventResult().message("ℹ️ Conversation is empty, nothing to compact."),
)
return
original_len = len(messages)
compressed = await cm.process(messages)
result = messages_to_history_dict(compressed)
await self.context.conversation_manager.update_conversation(umo, cid, result)
```
These extractions keep `compact` focused on the high-level flow (permission → context → process → save) and make each piece easier to test independently without changing the behavior.
</issue_to_address>
### Comment 5
<location path="astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py" line_range="94" />
<code_context>
)
- # 上下文管理相关
+ # 上下文管理相关（正交触发/处置模型）
self.context_limit_reached_strategy: str = settings.get(
"context_limit_reached_strategy",
</code_context>
<issue_to_address>
**issue (complexity):** Consider encapsulating all context-related settings into a dedicated config builder so `initialize` only wires grouped data into `MainAgentBuildConfig` instead of managing many individual fields and derived values directly.
The added orthogonal fields and inline logic for `max_context_length` / `dequeue_context_length` do increase complexity in `initialize`, mainly via the expanded parameter list to `MainAgentBuildConfig` and mixed concerns in this method.
You can reduce complexity while preserving functionality by:
1. Encapsulating all context-related settings in a helper/factory.
2. Letting that helper handle legacy vs new fields and derived values.
3. Passing a single `context_config` (or a small set of grouped configs) into `MainAgentBuildConfig`.
Example 1: Extract context config building logic
```python
@dataclass
class ContextConfig:
context_limit_reached_strategy: str
llm_compress_instruction: str
llm_compress_keep_recent_ratio: float
llm_compress_provider_id: str
max_context_length: int
dequeue_context_length: int
enable_turn_limit: bool
max_turns: int
enable_token_guard: bool
token_guard_threshold: float
enable_summary: bool
enable_discard: bool
discard_turns: int
summary_prompt: str
summary_provider_id: str
retention_method: str
retain_turns: int
retain_percentage: float
def build_context_config(settings: dict) -> ContextConfig:
max_context_length = settings.get("max_context_length", -1)
dequeue_context_length = min(
max(1, settings.get("dequeue_context_length", 1)),
max(1, max_context_length - 1),
)
if dequeue_context_length <= 0:
dequeue_context_length = 1
return ContextConfig(
context_limit_reached_strategy=settings.get(
"context_limit_reached_strategy", "truncate_by_turns"
),
llm_compress_instruction=settings.get("llm_compress_instruction", ""),
llm_compress_keep_recent_ratio=settings.get(
"llm_compress_keep_recent_ratio", 0.15
),
llm_compress_provider_id=settings.get("llm_compress_provider_id", ""),
max_context_length=max_context_length,
dequeue_context_length=dequeue_context_length,
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider_id=settings.get("summary_provider_id", ""),
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
)
```
Then `initialize` only wires the grouped config:
```python
def initialize(self, settings: dict, conf: dict) -> None:
# ...
context_config = build_context_config(settings)
# ...
main_agent_build_config = MainAgentBuildConfig(
# ...
context_config=context_config,
fallback_max_context_tokens=self.fallback_max_context_tokens,
# ...
)
```
Example 2: If `MainAgentBuildConfig` cannot be changed yet, at least isolate the wiring
```python
def _build_main_agent_context_kwargs(self, settings: dict) -> dict:
cfg = build_context_config(settings)
return {
"context_limit_reached_strategy": cfg.context_limit_reached_strategy,
"llm_compress_instruction": cfg.llm_compress_instruction,
"llm_compress_keep_recent_ratio": cfg.llm_compress_keep_recent_ratio,
"llm_compress_provider_id": cfg.llm_compress_provider_id,
"max_context_length": cfg.max_context_length,
"dequeue_context_length": cfg.dequeue_context_length,
"enable_turn_limit": cfg.enable_turn_limit,
"max_turns": cfg.max_turns,
"enable_token_guard": cfg.enable_token_guard,
"token_guard_threshold": cfg.token_guard_threshold,
"enable_summary": cfg.enable_summary,
"enable_discard": cfg.enable_discard,
"discard_turns": cfg.discard_turns,
"summary_prompt": cfg.summary_prompt,
"summary_provider_id": cfg.summary_provider_id,
"retention_method": cfg.retention_method,
"retain_turns": cfg.retain_turns,
"retain_percentage": cfg.retain_percentage,
}
```
Usage:
```python
context_kwargs = self._build_main_agent_context_kwargs(settings)
main_agent_build_config = MainAgentBuildConfig(
# ...
**context_kwargs,
fallback_max_context_tokens=self.fallback_max_context_tokens,
# ...
)
```
This keeps all behavior intact but moves the legacy/new field interplay and derived value logic out of `initialize`, making the stage easier to read and maintain while you still support both models.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| assert result is True | ||
| assert ps.get("retain_percentage") == 0.25 | ||
| assert ps.get("retention_method") == "percentage" | ||
| assert "llm_compress_keep_recent_ratio" not in ps |
Contributor
There was a problem hiding this comment.
issue (testing): Align test expectations with actual migration behavior for retain_percentage=0
The test docstring claims retention_method "stays 'turns'", but _migra_context_config currently sets retention_method = 'percentage' whenever llm_compress_keep_recent_ratio is present, even when it is 0. Since the test never asserts on retention_method, its description and name are misleading. Please either (a) add an assertion that retention_method becomes 'percentage' for ratio=0, or (b) change the migration logic so that ratio=0 keeps 'turns' and assert that behavior.
| if not self._triggers_fired(result, current_tokens, max_context_tokens): | ||
| return result | ||
|
|
||
| # 2. 处置入口：custom > summary > discard |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider explicitly initializing compressor fields and consolidating trigger checks into a single helper to simplify the disposal flow and reduce indirection.
You can reduce the indirection without changing behavior by (1) making compressor selection explicit (no hasattr checks) and (2) collapsing trigger checks into a single _should_dispose helper that lives next to process.
1. Make compressor selection explicit
Instead of relying on hasattr(self, "_unity_compressor"), initialize explicit fields and use is not None checks:
class ContextManager:
def __init__(self, config: ContextConfig) -> None:
self.config = config
self.token_counter = config.custom_token_counter or EstimateTokenCounter()
self.truncator = ContextTruncator()
self._unity_compressor = None
self._summary_compressor = None
self._discard_compressor = None
if config.custom_compressor:
self._unity_compressor = config.custom_compressor
else:
if config.summary_provider:
self._summary_compressor = LLMSummaryCompressor(
provider=config.summary_provider,
keep_recent_ratio=config.retain_percentage,
instruction_text=config.summary_prompt,
compression_threshold=config.token_guard_threshold,
token_counter=self.token_counter,
)
self._discard_compressor = TruncateByTurnsCompressor(
truncate_turns=config.discard_turns,
compression_threshold=config.token_guard_threshold,
)Then in process:
# 2. 处置入口：custom > summary > discard
if self._unity_compressor is not None:
result = await self._unity_compressor(result)
else:
compressed = await self._try_summary(result)
if compressed is not None:
result = compressed
else:
discarded = self._try_discard(result)
...This keeps the same priority chain but removes implicit existence checks and makes the object graph easier to reason about.
2. Collapse trigger logic into a single _should_dispose helper
You can inline _triggers_fired into a clearer _should_dispose that sits next to process, so the flow from tokens/turns → disposal is readable in one place:
def _should_dispose(
self,
messages: list[Message],
current_tokens: int,
max_context_tokens: int,
) -> bool:
# turn-limit trigger
if self.config.enable_turn_limit:
if self._count_turns(messages) > self.config.max_turns:
return True
# token-guard trigger
if self.config.enable_token_guard and max_context_tokens > 0 and current_tokens > 0:
ratio = current_tokens / max_context_tokens
return ratio > self.config.token_guard_threshold
return FalseAnd in process:
current_tokens = self.token_counter.count_tokens(result, trusted_token_usage)
if not self._should_dispose(result, current_tokens, max_context_tokens):
return resultThis keeps the orthogonal trigger/disposal model but reduces “mental jumping” between helpers: process now clearly shows count_tokens → should_dispose? → disposal path → optional halving, and the policy is encapsulated in one obvious method.
| logger.error(traceback.format_exc()) | ||
|
|
||
|
|
||
| def _migra_context_config(ps: dict) -> bool: |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider refactoring the migration and validation helpers into small, declarative sub-functions so each rule is isolated, testable, and easier to extend.
You can reduce the cognitive load of both functions by making the migration rules and validation rules more declarative and localized, while preserving all behavior.
1. Make _migra_context_config rule-based
Right now each legacy field is handled via an independent if that mutates ps. You can encapsulate each mapping as a small function and drive it from a list, which makes the overall mapping clearer and easier to extend:
def _migra_context_config(ps: dict) -> bool:
"""Migrate old context config fields to new orthogonal fields."""
migrated = False
def _migrate_max_context_length(ps: dict) -> bool:
if "max_context_length" not in ps:
return False
old_val = ps.pop("max_context_length")
if old_val > 0:
ps["enable_turn_limit"] = True
ps["max_turns"] = old_val
else:
ps["enable_turn_limit"] = False
return True
def _migrate_dequeue_context_length(ps: dict) -> bool:
if "dequeue_context_length" not in ps:
return False
ps["discard_turns"] = ps.pop("dequeue_context_length")
return True
def _migrate_context_limit_strategy(ps: dict) -> bool:
if "context_limit_reached_strategy" not in ps:
return False
strategy = ps.pop("context_limit_reached_strategy")
ps["enable_discard"] = True
ps["enable_summary"] = (strategy == "llm_compress")
return True
# ...small mappers for each legacy field...
for mapper in (
_migrate_max_context_length,
_migrate_dequeue_context_length,
_migrate_context_limit_strategy,
# _migrate_llm_compress_instruction,
# _migrate_llm_compress_keep_recent_ratio,
# _migrate_llm_compress_provider_id,
):
if mapper(ps):
migrated = True
if migrated:
logger.info("Migrated old context config to orthogonal trigger/disposal model.")
return migratedThis keeps all side effects but makes each legacy field’s transformation isolated and testable, and the “overall mapping” is visible in the mapper list.
2. Split _validate_context_config by concern
You can keep the same warnings/rules but group them into small helpers. This reduces nesting and repeated _has calls, and makes future additions easier to place:
def _validate_context_config(ps: dict) -> None:
"""Validate new context config fields and log warnings for violations."""
def _has(k: str) -> bool:
return k in ps
def _validate_turn_limit():
if _has("enable_turn_limit") and _has("max_turns"):
if ps.get("enable_turn_limit") and ps.get("max_turns", 50) < 2:
logger.warning("max_turns should be >= 2 when enable_turn_limit is True.")
def _validate_token_guard():
if _has("token_guard_threshold"):
val = ps["token_guard_threshold"]
if not (0.5 <= val <= 0.99):
logger.warning(
"token_guard_threshold %.2f is outside recommended range [0.5, 0.99].",
val,
)
def _validate_discard():
if _has("discard_turns") and ps.get("discard_turns", 1) < 1:
logger.warning("discard_turns should be >= 1.")
if _has("enable_summary") and _has("enable_discard"):
if not ps.get("enable_summary") and not ps.get("enable_discard"):
logger.warning(
"Both enable_summary and enable_discard are False. "
"Context will not be compressed when triggers fire.",
)
def _validate_retention():
if not _has("retention_method"):
return
method = ps["retention_method"]
if method not in ("turns", "percentage", "null"):
logger.warning("Unknown retention_method '%s'. Use 'turns', 'percentage', or 'null'.", method)
return
if method == "turns" and _has("retain_turns") and ps.get("retain_turns", 20) < 1:
logger.warning("retain_turns should be >= 1.")
if method == "percentage" and _has("retain_percentage"):
val = ps["retain_percentage"]
if not (0.1 <= val <= 0.9):
logger.warning(
"retain_percentage %.2f is outside recommended range [0.1, 0.9].",
val,
)
if (
ps.get("enable_turn_limit")
and method == "turns"
and _has("retain_turns")
and _has("max_turns")
and ps["retain_turns"] > ps["max_turns"]
):
logger.warning(
"retain_turns (%d) > max_turns (%d). Retention lower bound will take priority.",
ps["retain_turns"],
ps["max_turns"],
)
_validate_turn_limit()
_validate_token_guard()
_validate_discard()
_validate_retention()This preserves all current behavior and logging but makes each rule set (turn limit, token guard, disposal, retention) easier to reason about and extend.
|
|
||
| message.set_result(MessageEventResult().message(ret)) | ||
|
|
||
| async def compact(self, message: AstrMessageEvent) -> None: |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider extracting shared permission checks, context configuration building, and history/message conversion into reusable helpers so the new compact method focuses on high-level flow and avoids duplicated logic.
The new compact method is doing multiple distinct things and duplicates logic from reset, which increases complexity and maintenance cost. You can keep all functionality but reduce complexity via small extractions:
1. Shared permission + scene resolution helper
The permission check is almost identical to reset. Extract it and reuse:
# helpers.py (or near RstScene definition)
async def check_compact_permission(message: AstrMessageEvent, is_group: bool, is_unique_session: bool) -> Optional[str]:
scene = RstScene.get_scene(is_group, is_unique_session)
alter_cmd_cfg = await sp.get_async("global", "global", "alter_cmd", {})
plugin_config = alter_cmd_cfg.get("astrbot", {})
compact_cfg = plugin_config.get("compact", {})
required_perm = compact_cfg.get(
scene.key,
"admin" if is_group and not is_unique_session else "member",
)
if required_perm == "admin" and message.role != "admin":
return (
f"Compact command requires admin permission in {scene.name} scenario, "
f"you (ID {message.get_sender_id()}) are not admin, cannot perform this action."
)
return NoneUsage in compact:
is_group = bool(message.get_group_id())
is_unique_session = cfg["platform_settings"]["unique_session"]
err = await check_compact_permission(message, is_group, is_unique_session)
if err:
message.set_result(MessageEventResult().message(err))
returnThis keeps the permission logic centralized and makes both reset and compact easier to adjust.
2. ContextConfig factory
Config construction is inlined and likely duplicates other places where ContextConfig is built. Extract a factory:
# context_config_factory.py
from astrbot.core.agent.context.config import ContextConfig
def build_context_config_from_settings(settings: dict, summary_provider) -> ContextConfig:
return ContextConfig(
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider=summary_provider,
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
)Usage in compact:
summary_provider_id = settings.get("summary_provider_id", "")
summary_provider = (
self.context.get_provider_by_id(summary_provider_id)
if summary_provider_id
else self.context.get_using_provider(umo=umo)
)
config = build_context_config_from_settings(settings, summary_provider)
cm = ContextManager(config)This removes config noise from compact and aligns behavior with other agent/context initialization paths.
3. History ↔ Message conversion utilities
The manual loops for converting between dict and Message are verbose and type-specific. Extract tiny helpers:
# history_utils.py
import json
from typing import List
from astrbot.core.agent.message import Message # adjust import as needed
def conversation_history_to_messages(history_raw) -> List[Message]:
if isinstance(history_raw, str):
history: list[dict] = json.loads(history_raw or "[]")
else:
history = history_raw or []
return [
Message(role=item.get("role", "user"), content=item.get("content", ""))
for item in history
]
def messages_to_history_dict(messages: List[Message]) -> list[dict]:
result: list[dict] = []
for msg in messages:
if isinstance(msg.content, str):
content = msg.content
else:
content = str(msg.content) if msg.content else ""
result.append({"role": msg.role, "content": content})
return resultUsage in compact:
conv = await self.context.conversation_manager.get_conversation(umo, cid)
messages = conversation_history_to_messages(conv.history)
if not messages:
message.set_result(
MessageEventResult().message("ℹ️ Conversation is empty, nothing to compact."),
)
return
original_len = len(messages)
compressed = await cm.process(messages)
result = messages_to_history_dict(compressed)
await self.context.conversation_manager.update_conversation(umo, cid, result)These extractions keep compact focused on the high-level flow (permission → context → process → save) and make each piece easier to test independently without changing the behavior.
| ) | ||
|
|
||
| # 上下文管理相关 | ||
| # 上下文管理相关（正交触发/处置模型） |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider encapsulating all context-related settings into a dedicated config builder so initialize only wires grouped data into MainAgentBuildConfig instead of managing many individual fields and derived values directly.
The added orthogonal fields and inline logic for max_context_length / dequeue_context_length do increase complexity in initialize, mainly via the expanded parameter list to MainAgentBuildConfig and mixed concerns in this method.
You can reduce complexity while preserving functionality by:
- Encapsulating all context-related settings in a helper/factory.
- Letting that helper handle legacy vs new fields and derived values.
- Passing a single
context_config(or a small set of grouped configs) intoMainAgentBuildConfig.
Example 1: Extract context config building logic
@dataclass
class ContextConfig:
context_limit_reached_strategy: str
llm_compress_instruction: str
llm_compress_keep_recent_ratio: float
llm_compress_provider_id: str
max_context_length: int
dequeue_context_length: int
enable_turn_limit: bool
max_turns: int
enable_token_guard: bool
token_guard_threshold: float
enable_summary: bool
enable_discard: bool
discard_turns: int
summary_prompt: str
summary_provider_id: str
retention_method: str
retain_turns: int
retain_percentage: float
def build_context_config(settings: dict) -> ContextConfig:
max_context_length = settings.get("max_context_length", -1)
dequeue_context_length = min(
max(1, settings.get("dequeue_context_length", 1)),
max(1, max_context_length - 1),
)
if dequeue_context_length <= 0:
dequeue_context_length = 1
return ContextConfig(
context_limit_reached_strategy=settings.get(
"context_limit_reached_strategy", "truncate_by_turns"
),
llm_compress_instruction=settings.get("llm_compress_instruction", ""),
llm_compress_keep_recent_ratio=settings.get(
"llm_compress_keep_recent_ratio", 0.15
),
llm_compress_provider_id=settings.get("llm_compress_provider_id", ""),
max_context_length=max_context_length,
dequeue_context_length=dequeue_context_length,
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider_id=settings.get("summary_provider_id", ""),
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
)Then initialize only wires the grouped config:
def initialize(self, settings: dict, conf: dict) -> None:
# ...
context_config = build_context_config(settings)
# ...
main_agent_build_config = MainAgentBuildConfig(
# ...
context_config=context_config,
fallback_max_context_tokens=self.fallback_max_context_tokens,
# ...
)Example 2: If MainAgentBuildConfig cannot be changed yet, at least isolate the wiring
def _build_main_agent_context_kwargs(self, settings: dict) -> dict:
cfg = build_context_config(settings)
return {
"context_limit_reached_strategy": cfg.context_limit_reached_strategy,
"llm_compress_instruction": cfg.llm_compress_instruction,
"llm_compress_keep_recent_ratio": cfg.llm_compress_keep_recent_ratio,
"llm_compress_provider_id": cfg.llm_compress_provider_id,
"max_context_length": cfg.max_context_length,
"dequeue_context_length": cfg.dequeue_context_length,
"enable_turn_limit": cfg.enable_turn_limit,
"max_turns": cfg.max_turns,
"enable_token_guard": cfg.enable_token_guard,
"token_guard_threshold": cfg.token_guard_threshold,
"enable_summary": cfg.enable_summary,
"enable_discard": cfg.enable_discard,
"discard_turns": cfg.discard_turns,
"summary_prompt": cfg.summary_prompt,
"summary_provider_id": cfg.summary_provider_id,
"retention_method": cfg.retention_method,
"retain_turns": cfg.retain_turns,
"retain_percentage": cfg.retain_percentage,
}Usage:
context_kwargs = self._build_main_agent_context_kwargs(settings)
main_agent_build_config = MainAgentBuildConfig(
# ...
**context_kwargs,
fallback_max_context_tokens=self.fallback_max_context_tokens,
# ...
)This keeps all behavior intact but moves the legacy/new field interplay and derived value logic out of initialize, making the stage easier to read and maintain while you still support both models.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request introduces a redesigned context management system based on an orthogonal trigger/disposal model, replacing old implicit-value strategies with independent triggers (turn limits and token guards) and disposal actions (LLM summary and turn discarding). It includes a new manual /compact command, updated configuration schemas, migration and validation helpers, and comprehensive test coverage. The review feedback identifies three important issues: the manual /compact command fails to pass max_context_tokens to the processor (preventing token guards from firing) and corrupts multimodal message histories by stringifying list content during serialization. Additionally, the migration helper unconditionally overwrites the retention method to 'percentage' when migrating llm_compress_keep_recent_ratio, which incorrectly alters settings for users transitioning from turn-based truncation.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
Comment on lines
+298
to
+299
| try: | ||
| compressed = await cm.process(messages) |
Contributor
There was a problem hiding this comment.
In the manual /compact command, cm.process(messages) is called without passing max_context_tokens. Under the new orthogonal model, if max_context_tokens is 0 (the default), the token guard trigger will never fire. Furthermore, since enable_turn_limit defaults to False, the manual compact command will do nothing under default settings. Passing the resolved provider's max_context_tokens ensures the token guard trigger can function correctly.
Suggested change
| try: | |
| compressed = await cm.process(messages) | |
| max_context_tokens = summary_provider.provider_config.get("max_context_tokens", 0) if summary_provider else 0 | |
| try: | |
| compressed = await cm.process(messages, max_context_tokens=max_context_tokens) |
Comment on lines
+310
to
+313
| if isinstance(msg.content, str): | ||
| result.append({"role": msg.role, "content": msg.content}) | ||
| else: | ||
| result.append({"role": msg.role, "content": str(msg.content) if msg.content else ""}) |
Contributor
There was a problem hiding this comment.
When converting Message objects back to dicts, if msg.content is a list (which is the case for multimodal messages containing parts like images or audio), calling str(msg.content) will serialize the Python list representation (e.g., "[TextPart(...)]"), corrupting the multimodal history. Preserving lists directly avoids this data corruption.
Suggested change
| if isinstance(msg.content, str): | |
| result.append({"role": msg.role, "content": msg.content}) | |
| else: | |
| result.append({"role": msg.role, "content": str(msg.content) if msg.content else ""}) | |
| if isinstance(msg.content, (str, list)): | |
| result.append({"role": msg.role, "content": msg.content}) | |
| else: | |
| result.append({"role": msg.role, "content": str(msg.content) if msg.content else ""}) |
Comment on lines
+218
to
+238
| # 3. context_limit_reached_strategy → enable_summary + enable_discard | ||
| if "context_limit_reached_strategy" in ps: | ||
| strategy = ps.pop("context_limit_reached_strategy") | ||
| migrated = True | ||
| if strategy == "llm_compress": | ||
| ps["enable_summary"] = True | ||
| ps["enable_discard"] = True | ||
| else: # truncate_by_turns or other | ||
| ps["enable_summary"] = False | ||
| ps["enable_discard"] = True | ||
|
|
||
| # 4. llm_compress_instruction → summary_prompt | ||
| if "llm_compress_instruction" in ps: | ||
| ps["summary_prompt"] = ps.pop("llm_compress_instruction") | ||
| migrated = True | ||
|
|
||
| # 5. llm_compress_keep_recent_ratio → retain_percentage + retention_method | ||
| if "llm_compress_keep_recent_ratio" in ps: | ||
| ps["retain_percentage"] = ps.pop("llm_compress_keep_recent_ratio") | ||
| ps["retention_method"] = "percentage" | ||
| migrated = True |
Contributor
There was a problem hiding this comment.
In the migration logic, llm_compress_keep_recent_ratio is unconditionally migrated to retention_method = "percentage". However, since llm_compress_keep_recent_ratio is present in the default settings, this will overwrite the retention method to "percentage" even for configurations whose strategy was "truncate_by_turns". We should set the retention method based on the active strategy to preserve the correct behavior.
# 3. context_limit_reached_strategy → enable_summary + enable_discard
strategy = ps.pop("context_limit_reached_strategy", None)
if strategy is not None:
migrated = True
if strategy == "llm_compress":
ps["enable_summary"] = True
ps["enable_discard"] = True
ps["retention_method"] = "percentage"
else: # truncate_by_turns or other
ps["enable_summary"] = False
ps["enable_discard"] = True
ps["retention_method"] = "turns"
# 4. llm_compress_instruction → summary_prompt
if "llm_compress_instruction" in ps:
ps["summary_prompt"] = ps.pop("llm_compress_instruction")
migrated = True
# 5. llm_compress_keep_recent_ratio → retain_percentage + retention_method
if "llm_compress_keep_recent_ratio" in ps:
ps["retain_percentage"] = ps.pop("llm_compress_keep_recent_ratio")
if "retention_method" not in ps:
ps["retention_method"] = "percentage"
migrated = True
Rail1bc
marked this pull request as draft
July 21, 2026 10:00
- Pass max_context_tokens from provider to process() so token guard trigger and double-check halving work during manual compression - Preserve tool_calls/tool_call_id in dict↔Message conversion to prevent data loss when round-tripping conversation history - Correct test name/doc for retain_percentage_zero → 'percentage' test: add 23 tests covering previously untested scenarios - Custom compressor (_unity_compressor) end-to-end path through process() - _try_summary edge cases: identity check, no-reduction, longer result, compressor exception with/without discard fallback - _try_discard edge cases: retention prevents discard, total_turns=0, retain_percentage boundaries (0 and 1.0) - trusted_token_usage non-zero passthrough verification - /compact command: 9 tests covering permission deny/allow, third-party runner skip, no-conversation, empty history, successful compression, tool_calls metadata preservation, and error handling Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
Contributor
Author
|
@sourcery-ai review |
Contributor
Author
|
/gemini review |
Contributor
There was a problem hiding this comment.
Hey - I've found 1 issue, and left some high level feedback:
- The construction of ContextConfig from provider_settings (in /compact, MainAgentBuildConfig/build_main_agent, tool_loop_agent_runner, and pipeline internal.initialize) is now duplicated; consider centralizing this mapping into a single factory/helper to avoid drift between call sites.
- Ranges and defaults for context management (e.g. token_guard_threshold 0.5–0.99, retain_percentage 0.1–0.9, min max_turns/retain_turns/discard_turns) are currently encoded in multiple places (ContextConfig defaults, _validate_context_config, tests); it would be more maintainable to define these constraints once and reuse them.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The construction of ContextConfig from provider_settings (in /compact, MainAgentBuildConfig/build_main_agent, tool_loop_agent_runner, and pipeline internal.initialize) is now duplicated; consider centralizing this mapping into a single factory/helper to avoid drift between call sites.
- Ranges and defaults for context management (e.g. token_guard_threshold 0.5–0.99, retain_percentage 0.1–0.9, min max_turns/retain_turns/discard_turns) are currently encoded in multiple places (ContextConfig defaults, _validate_context_config, tests); it would be more maintainable to define these constraints once and reuse them.
## Individual Comments
### Comment 1
<location path="astrbot/core/agent/context/manager.py" line_range="46" />
<code_context>
+ compression_threshold=config.token_guard_threshold,
)
- async def process(
- self,
- messages: list[Message],
</code_context>
<issue_to_address>
**issue (complexity):** Consider consolidating trigger and token-guard handling, making compressor attributes explicit, and centralizing discard/retention math to streamline the context disposal pipeline.
You can keep the orthogonal model and features, but reduce cognitive overhead by tightening the pipeline and simplifying wiring.
### 1. Inline `_triggers_fired` and centralize token guard
The current token guard logic is split between `_triggers_fired` and the “double-check” at the end of `process`. You can keep the pre-/post-disposal semantics but make them explicit in one place.
For example, move the trigger logic into `process` and introduce a single helper for token guard that handles both phases:
```python
def _token_guard_exceeded(
self,
tokens: int,
max_context_tokens: int,
) -> bool:
if not self.config.enable_token_guard or max_context_tokens <= 0 or tokens <= 0:
return False
ratio = tokens / max_context_tokens
return ratio > self.config.token_guard_threshold
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,
)
# Trigger evaluation: turn limit + token guard in one place
trigger_fired = False
if self.config.enable_turn_limit:
total_turns = self._count_turns(result)
if total_turns > self.config.max_turns:
trigger_fired = True
if not trigger_fired and self._token_guard_exceeded(current_tokens, max_context_tokens):
trigger_fired = True
if not trigger_fired:
return result
# Disposal ordering unchanged
if self._unity_compressor is not None:
result = await self._unity_compressor(result)
else:
compressed = await self._try_summary(result)
if compressed is not None:
result = compressed
else:
discarded = self._try_discard(result)
if discarded is not None:
result = discarded
else:
logger.warning(
"Context disposal triggered but both summary and discard "
"are unavailable or disabled. No compression applied.",
)
# Post-disposal token guard halving, using the same helper
tokens_after = self.token_counter.count_tokens(result)
if self._token_guard_exceeded(tokens_after, max_context_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
```
This keeps the same behavior but removes duplicated guard conditions and indirection through `_triggers_fired`.
### 2. Replace `hasattr` with explicit compressor attributes
Using `hasattr(self, "_unity_compressor")` introduces dynamic checks and implicit state. You already initialize compressors conditionally, so you can make them explicit and always present as attributes:
```python
def __init__(self, config: ContextConfig) -> None:
self.config = config
self.token_counter = config.custom_token_counter or EstimateTokenCounter()
self.truncator = ContextTruncator()
self._summary_compressor: LLMSummaryCompressor | None = None
self._discard_compressor: TruncateByTurnsCompressor | None = None
self._unity_compressor = config.custom_compressor
if config.summary_provider:
self._summary_compressor = LLMSummaryCompressor(
provider=config.summary_provider,
keep_recent_ratio=config.retain_percentage,
instruction_text=config.summary_prompt,
compression_threshold=config.token_guard_threshold,
token_counter=self.token_counter,
)
self._discard_compressor = TruncateByTurnsCompressor(
truncate_turns=config.discard_turns,
compression_threshold=config.token_guard_threshold,
)
```
And in `process`:
```python
if self._unity_compressor is not None:
result = await self._unity_compressor(result)
else:
compressed = await self._try_summary(result)
...
```
This removes dynamic attribute checks and makes compressor presence/ordering clearer.
### 3. Simplify retention/discard calculation
The discard flow currently touches `_compute_discard_limit`, `_count_turns`, `discard_turns`, and `enable_discard` across multiple places. You can encapsulate “how many turns may be thrown away” in a single helper that returns `0` when retention or config prevents discard:
```python
def _compute_discard_turns(self, messages: list[Message]) -> int:
if not self.config.enable_discard:
return 0
total_turns = self._count_turns(messages)
if total_turns <= 0:
return 0
method = self.config.retention_method
if method == "turns":
min_keep = self.config.retain_turns
elif method == "percentage":
min_keep = int(total_turns * self.config.retain_percentage)
else: # "null"
min_keep = 0
min_keep = max(0, min_keep)
max_discardable = max(0, total_turns - min_keep)
return min(self.config.discard_turns, max_discardable)
```
Then `_try_discard` becomes a thin wrapper:
```python
def _try_discard(self, messages: list[Message]) -> list[Message] | None:
requested = self._compute_discard_turns(messages)
if requested <= 0:
return None
return self.truncator.truncate_by_dropping_oldest_turns(
messages, drop_turns=requested,
)
```
This keeps retention behavior intact but centralizes the discard/retention logic into one helper, making the disposal pipeline easier to follow.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| 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 consolidating trigger and token-guard handling, making compressor attributes explicit, and centralizing discard/retention math to streamline the context disposal pipeline.
You can keep the orthogonal model and features, but reduce cognitive overhead by tightening the pipeline and simplifying wiring.
1. Inline _triggers_fired and centralize token guard
The current token guard logic is split between _triggers_fired and the “double-check” at the end of process. You can keep the pre-/post-disposal semantics but make them explicit in one place.
For example, move the trigger logic into process and introduce a single helper for token guard that handles both phases:
def _token_guard_exceeded(
self,
tokens: int,
max_context_tokens: int,
) -> bool:
if not self.config.enable_token_guard or max_context_tokens <= 0 or tokens <= 0:
return False
ratio = tokens / max_context_tokens
return ratio > self.config.token_guard_threshold
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,
)
# Trigger evaluation: turn limit + token guard in one place
trigger_fired = False
if self.config.enable_turn_limit:
total_turns = self._count_turns(result)
if total_turns > self.config.max_turns:
trigger_fired = True
if not trigger_fired and self._token_guard_exceeded(current_tokens, max_context_tokens):
trigger_fired = True
if not trigger_fired:
return result
# Disposal ordering unchanged
if self._unity_compressor is not None:
result = await self._unity_compressor(result)
else:
compressed = await self._try_summary(result)
if compressed is not None:
result = compressed
else:
discarded = self._try_discard(result)
if discarded is not None:
result = discarded
else:
logger.warning(
"Context disposal triggered but both summary and discard "
"are unavailable or disabled. No compression applied.",
)
# Post-disposal token guard halving, using the same helper
tokens_after = self.token_counter.count_tokens(result)
if self._token_guard_exceeded(tokens_after, max_context_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 messagesThis keeps the same behavior but removes duplicated guard conditions and indirection through _triggers_fired.
2. Replace hasattr with explicit compressor attributes
Using hasattr(self, "_unity_compressor") introduces dynamic checks and implicit state. You already initialize compressors conditionally, so you can make them explicit and always present as attributes:
def __init__(self, config: ContextConfig) -> None:
self.config = config
self.token_counter = config.custom_token_counter or EstimateTokenCounter()
self.truncator = ContextTruncator()
self._summary_compressor: LLMSummaryCompressor | None = None
self._discard_compressor: TruncateByTurnsCompressor | None = None
self._unity_compressor = config.custom_compressor
if config.summary_provider:
self._summary_compressor = LLMSummaryCompressor(
provider=config.summary_provider,
keep_recent_ratio=config.retain_percentage,
instruction_text=config.summary_prompt,
compression_threshold=config.token_guard_threshold,
token_counter=self.token_counter,
)
self._discard_compressor = TruncateByTurnsCompressor(
truncate_turns=config.discard_turns,
compression_threshold=config.token_guard_threshold,
)And in process:
if self._unity_compressor is not None:
result = await self._unity_compressor(result)
else:
compressed = await self._try_summary(result)
...This removes dynamic attribute checks and makes compressor presence/ordering clearer.
3. Simplify retention/discard calculation
The discard flow currently touches _compute_discard_limit, _count_turns, discard_turns, and enable_discard across multiple places. You can encapsulate “how many turns may be thrown away” in a single helper that returns 0 when retention or config prevents discard:
def _compute_discard_turns(self, messages: list[Message]) -> int:
if not self.config.enable_discard:
return 0
total_turns = self._count_turns(messages)
if total_turns <= 0:
return 0
method = self.config.retention_method
if method == "turns":
min_keep = self.config.retain_turns
elif method == "percentage":
min_keep = int(total_turns * self.config.retain_percentage)
else: # "null"
min_keep = 0
min_keep = max(0, min_keep)
max_discardable = max(0, total_turns - min_keep)
return min(self.config.discard_turns, max_discardable)Then _try_discard becomes a thin wrapper:
def _try_discard(self, messages: list[Message]) -> list[Message] | None:
requested = self._compute_discard_turns(messages)
if requested <= 0:
return None
return self.truncator.truncate_by_dropping_oldest_turns(
messages, drop_turns=requested,
)This keeps retention behavior intact but centralizes the discard/retention logic into one helper, making the disposal pipeline easier to follow.
Contributor
There was a problem hiding this comment.
Code Review
This pull request redesigns the context management system in AstrBot, replacing the old implicit-value fields with an orthogonal trigger/disposal model. It introduces a new /compact command to manually trigger context compression, updates ContextConfig and ContextManager to support independent trigger checks (turn limits and token guards) and unified disposal entries (LLM summary and discard with retention constraints), and adds migration and validation helpers for backward compatibility. The review feedback highlights two important improvements: first, in conversation.py, msg.content should be handled carefully to preserve structured multimodal list data and None values rather than blindly casting to a string; second, in manager.py, the percentage retention calculation should use math.ceil instead of int to prevent rounding down to zero and violating the retention lower bound.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
Comment on lines
+320
to
+326
| result: list[dict] = [] | ||
| for msg in compressed: | ||
| entry = {"role": msg.role} | ||
| if isinstance(msg.content, str): | ||
| entry["content"] = msg.content | ||
| else: | ||
| entry["content"] = str(msg.content) if msg.content else "" |
Contributor
There was a problem hiding this comment.
对于多模态消息，msg.content 可能会是一个列表（list）。如果直接使用 str(msg.content)，会将其转换为 Python 列表的字符串表示（例如 "[{'type': 'text', ...}]"），从而破坏原有的结构化多模态数据。此外，如果 msg.content 为 None，使用 str(msg.content) if msg.content else "" 会将其转换为空字符串 ""。建议显式判断 None 并支持 list 类型，以完整保留多模态和空内容的原始结构。
result: list[dict] = []
for msg in compressed:
entry = {"role": msg.role}
if msg.content is not None:
if isinstance(msg.content, (str, list)):
entry["content"] = msg.content
else:
entry["content"] = str(msg.content)
else:
entry["content"] = None
Comment on lines
+55
to
+64
| def _compute_discard_limit(self, total_turns: int) -> int: | ||
| """Maximum number of turns that discard may remove, given retention.""" | ||
| method = self.config.retention_method | ||
| if method == "turns": | ||
| return max(0, total_turns - self.config.retain_turns) | ||
| if method == "percentage": | ||
| floor = int(total_turns * self.config.retain_percentage) | ||
| return max(0, total_turns - floor) | ||
| # "null" — no lower bound | ||
| return total_turns |
Contributor
There was a problem hiding this comment.
在按比例保留（percentage）的计算中，使用 int() 进行向下取整可能会导致保留的轮数少于预期比例。例如，当 total_turns 为 3 且 retain_percentage 为 0.3 时，int(3 * 0.3) = 0，这将导致允许丢弃所有轮次（保留 0 轮，即 0%），违背了保留下限的初衷。建议使用 math.ceil 向上取整，以确保至少保留符合比例要求的最小轮数。
Suggested change
| def _compute_discard_limit(self, total_turns: int) -> int: | |
| """Maximum number of turns that discard may remove, given retention.""" | |
| method = self.config.retention_method | |
| if method == "turns": | |
| return max(0, total_turns - self.config.retain_turns) | |
| if method == "percentage": | |
| floor = int(total_turns * self.config.retain_percentage) | |
| return max(0, total_turns - floor) | |
| # "null" — no lower bound | |
| return total_turns | |
| def _compute_discard_limit(self, total_turns: int) -> int: | |
| """Maximum number of turns that discard may remove, given retention.""" | |
| method = self.config.retention_method | |
| if method == "turns": | |
| return max(0, total_turns - self.config.retain_turns) | |
| if method == "percentage": | |
| import math | |
| floor = math.ceil(total_turns * self.config.retain_percentage) | |
| return max(0, total_turns - floor) | |
| # "null" — no lower bound | |
| return total_turns |
…n ceiling - Extract _token_guard_exceeded helper to eliminate duplicated token guard logic between trigger check and post-disposal halving. - Fix content serialization in /compact to preserve None and list[ContentPart] multimodal data instead of casting to str. - Change percentage retention from int() to math.ceil() to prevent rounding down to zero and violating retention lower bound. Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
Contributor
Author
|
/gemini review |
Contributor
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - I've found 6 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="astrbot/core/agent/context/manager.py" line_range="68-76" />
<code_context>
+ # "null" — no lower bound
+ return total_turns
+
+ async def _try_summary(self, messages: list[Message]) -> list[Message] | None:
+ """Attempt LLM summary compression. Returns compressed messages or None."""
+ if not self.config.enable_summary or self._summary_compressor is None:
+ return None
+ try:
+ result = await self._summary_compressor(messages)
+ if result is messages:
+ return None # compressor chose not to compress
+ if len(result) >= len(messages):
+ return None # no effective reduction
+ return result
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Summary compression success is judged only by message count, which may miss real token-size reductions.
The current check `len(result) >= len(messages)` assumes fewer messages always means fewer tokens, which isn’t necessarily true. A summary can reduce token count while keeping the same number of messages, so `_try_summary` may wrongly return `None` and trigger discard logic or warnings even when the context is already within token limits.
Please consider evaluating summary effectiveness using token counts (via `token_counter`) rather than message count, or remove the `len(result)` check and rely on the token-guard recheck after disposal to decide if further halving is needed.
</issue_to_address>
### Comment 2
<location path="astrbot/core/agent/context/manager.py" line_range="96-102" />
<code_context>
+ messages, drop_turns=requested,
+ )
+
+ 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:
+ return False
+ if max_context_tokens <= 0 or tokens <= 0:
+ return False
+ return (tokens / max_context_tokens) > self.config.token_guard_threshold
+
+ def _triggers_fired(
</code_context>
<issue_to_address>
**suggestion:** Token guard uses `max_context_tokens` only from the call site, which may be zero and silently disable the trigger.
In `process`, `_token_guard_exceeded` returns `False` whenever `max_context_tokens <= 0`. In `ToolLoopAgentRunner` and `conversation.compact`, `max_context_tokens` comes from `provider.provider_config.get("max_context_tokens", 0)`, so any provider without this metadata (or without a provider) will silently disable token guarding even when `enable_token_guard` is True.
If that's unintended, consider falling back to a default when `max_context_tokens` is non-positive (e.g., a value from `ContextConfig`), or emit a warning so misconfigured providers are visible to operators.
Suggested implementation:
```python
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 `max_context_tokens` is non-positive, fall back to a configured default.
If no valid maximum can be determined, emit a warning and return False.
"""
if not self.config.enable_token_guard:
return False
if tokens <= 0:
return False
# Prefer the value provided by the caller; fall back to configuration if missing/invalid.
effective_max_context_tokens = max_context_tokens
if effective_max_context_tokens <= 0:
# Fall back to a default from ContextConfig (if available)
effective_max_context_tokens = getattr(self.config, "max_context_tokens", 0)
if effective_max_context_tokens <= 0:
# No usable max context size; token guard is configured but cannot be applied.
# Emit a warning so misconfigured providers are visible to operators.
logger.warning(
"Token guard is enabled but no valid max_context_tokens is configured "
"(received=%s, fallback=%s). Token guarding is effectively disabled.",
max_context_tokens,
getattr(self.config, "max_context_tokens", None),
)
return False
return (tokens / effective_max_context_tokens) > self.config.token_guard_threshold
```
1. Ensure a module-level logger is defined (e.g., `import logging` and `logger = logging.getLogger(__name__)`) if it does not already exist in `astrbot/core/agent/context/manager.py`.
2. Verify that `self.config` (likely `ContextConfig`) has a `max_context_tokens` attribute; if it uses a different name for the default context size, update the fallback access (`getattr(self.config, "max_context_tokens", 0)`) accordingly.
3. Optionally, if there is already a centralized logging or warning mechanism for configuration issues, replace the `logger.warning(...)` call with that existing mechanism to stay consistent with the rest of the codebase.
</issue_to_address>
### Comment 3
<location path="astrbot/core/agent/context/manager.py" line_range="168-174" />
<code_context>
+ )
+
+ # 3. double-check（仅 enable_token_guard）
+ tokens_after = self.token_counter.count_tokens(result)
+ if self._token_guard_exceeded(tokens_after, max_context_tokens):
+ logger.info(
+ "Context still exceeds token guard threshold after disposal, "
+ "applying halving truncation (unconstrained by retention).",
)
-
- if self.compressor.should_compress(
- result,
- total_tokens,
- self.config.max_context_tokens,
- ):
- result = await self._run_compression(result, total_tokens)
+ result = self.truncator.truncate_by_halving(result)
return result
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Double-check token counting ignores `trusted_token_usage`, which may skew guard behavior compared to the initial trigger.
The initial trigger calls `count_tokens(result, trusted_token_usage)`, while the post-disposal check calls `count_tokens(result)` without the hint. If `trusted_token_usage` materially changes the token estimate (e.g., provider vs local counting), the second check may diverge from the first and incorrectly apply or skip halving.
Please either pass `trusted_token_usage` consistently to the double-check, or document why the second pass intentionally uses a different counting strategy, so the guard’s behavior stays consistent.
Suggested implementation:
```python
# 3. double-check（仅 enable_token_guard）
# Use the same trusted_token_usage hint as the initial trigger to keep
# token guard behavior consistent between pre- and post-disposal checks.
tokens_after = self.token_counter.count_tokens(result, trusted_token_usage)
if self._token_guard_exceeded(tokens_after, max_context_tokens):
logger.info(
"Context still exceeds token guard threshold after disposal, "
"applying halving truncation (unconstrained by retention).",
)
result = self.truncator.truncate_by_halving(result)
```
For this change to work, `trusted_token_usage` must be available in the scope where the double-check is performed. If it is currently only used in the initial trigger, you will need to:
1. Ensure `trusted_token_usage` is captured or passed into this method (if it is not already).
2. Thread `trusted_token_usage` through any intermediate helpers so that both the initial trigger and the post-disposal double-check can use the same hint when calling `self.token_counter.count_tokens(...)`.
</issue_to_address>
### Comment 4
<location path="astrbot/builtin_stars/builtin_commands/commands/conversation.py" line_range="250-257" />
<code_context>
+ )
+ return
+
+ # 解析历史记录
+ import json
+
+ raw_history: str = conv.history or "[]"
+ history: list[dict] = (
+ json.loads(raw_history) if isinstance(raw_history, str) else raw_history
</code_context>
<issue_to_address>
**suggestion (bug_risk):** The compact command silently treats non-string `conv.history` as already-parsed JSON, which may be fragile.
`raw_history` is annotated as `str`, but `history` is set to `json.loads(raw_history)` only when it’s a string; otherwise `history = raw_history`. This effectively treats any non-string `conv.history` as already-parsed history and a valid `list[dict]`. If `conv.history` ever contains another shape (e.g., a metadata dict), `history` will have an unexpected type and later code will still try to iterate it as a list of message dicts.
Consider validating `conv.history`’s structure (e.g., checking it’s a list of dicts) and raising or logging a clear error when it isn’t, rather than relying solely on the `isinstance(raw_history, str)` check.
```suggestion
# 解析历史记录
import json
raw_history = conv.history or "[]"
# 将字符串形式的历史记录解析为 JSON；非字符串则直接使用
try:
if isinstance(raw_history, str):
parsed_history = json.loads(raw_history)
else:
parsed_history = raw_history
except Exception:
message.set_result(
MessageEventResult().message("⚠️ Conversation history is invalid JSON and cannot be compacted."),
)
return
# 验证历史记录结构：必须是由 dict 组成的列表
if not isinstance(parsed_history, list) or any(
not isinstance(item, dict) for item in parsed_history
):
message.set_result(
MessageEventResult().message("⚠️ Conversation history has an unexpected structure and cannot be compacted."),
)
return
history: list[dict] = parsed_history
if not history:
```
</issue_to_address>
### Comment 5
<location path="astrbot/core/agent/context/manager.py" line_range="150" />
<code_context>
+ if not self._triggers_fired(result, current_tokens, max_context_tokens):
+ return result
+
+ # 2. 处置入口：custom > summary > discard
+ if hasattr(self, "_unity_compressor"):
+ result = await self._unity_compressor(result)
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the disposal selection and token-guard halving into dedicated helpers to simplify `process` and make the context management flow clearer.
You can centralize the disposal selection and the token‑guard halving into small helpers to reduce branching in `process` while preserving the orthogonal trigger/disposal model.
### 1. Encapsulate disposal ordering into a single helper
Move the “custom > summary > discard” decision into a dedicated helper that returns the chosen compressor (or `None`). This keeps `process` focused on the pipeline and makes the disposal strategy easier to reason about.
```python
async def _select_disposal(self, messages: list[Message]) -> list[Message]:
"""Apply disposal strategy: custom > summary > discard."""
# Custom compressor wins if present
if hasattr(self, "_unity_compressor"):
return await self._unity_compressor(messages)
# Summary first
compressed = await self._try_summary(messages)
if compressed is not None:
return compressed
# Then discard
discarded = self._try_discard(messages)
if discarded is not None:
return discarded
logger.warning(
"Context disposal triggered but both summary and discard "
"are unavailable or disabled. No compression applied.",
)
return messages
```
Then `process` becomes:
```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,
)
if not self._triggers_fired(result, current_tokens, max_context_tokens):
return result
# 处置入口：single call
result = await self._select_disposal(result)
# 3. double-check（仅 enable_token_guard）
result = self._apply_token_guard_halving(result, max_context_tokens)
return result
except Exception:
logger.error("Error during context processing.", exc_info=True)
return messages
```
### 2. Extract token‑guard halving into a small policy helper
Wrap the “double‑check halving” into its own function so the token guard policy is defined in one place and doesn’t interleave with disposal logic:
```python
def _apply_token_guard_halving(
self,
messages: list[Message],
max_context_tokens: int,
) -> list[Message]:
"""If still over token guard, apply halving truncation."""
if not self.config.enable_token_guard:
return messages
tokens_after = self.token_counter.count_tokens(messages)
if self._token_guard_exceeded(tokens_after, max_context_tokens):
logger.info(
"Context still exceeds token guard threshold after disposal, "
"applying halving truncation (unconstrained by retention).",
)
return self.truncator.truncate_by_halving(messages)
return messages
```
This removes duplicated guard checks from `process` and makes the “double‑check + halving” behavior explicit and easier to maintain, without changing any of the existing functionality or thresholds.
</issue_to_address>
### Comment 6
<location path="astrbot/builtin_stars/builtin_commands/commands/conversation.py" line_range="196" />
<code_context>
message.set_result(MessageEventResult().message(ret))
+ async def compact(self, message: AstrMessageEvent) -> None:
+ """手动触发上下文的处置与压缩"""
+ umo = message.unified_msg_origin
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting context config construction, summary provider resolution, history/message conversion, and permission checking into reusable helpers to keep `compact` shorter and more maintainable.
You can reduce complexity here by extracting a few focused helpers, without changing behavior.
### 1. Extract `ContextConfig` construction
You’re manually wiring a lot of context settings; this is reusable and matches behavior in other agents/runners.
```python
# somewhere reusable (e.g., a utils module or on self.context)
def build_context_config_from_settings(settings, summary_provider):
from astrbot.core.agent.context.config import ContextConfig
return ContextConfig(
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider=summary_provider,
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
)
```
Usage in `compact`:
```python
summary_provider = self._resolve_summary_provider(settings, umo)
config = build_context_config_from_settings(settings, summary_provider)
cm = ContextManager(config)
```
### 2. Extract summary provider resolution
This logic is repeated in other places (e.g. `_get_compress_provider`); a shared helper keeps behavior consistent.
```python
def _resolve_summary_provider(self, settings, umo):
summary_provider_id = settings.get("summary_provider_id", "")
if summary_provider_id:
provider = self.context.get_provider_by_id(summary_provider_id)
if provider:
return provider
return self.context.get_using_provider(umo=umo)
```
Then:
```python
summary_provider = self._resolve_summary_provider(settings, umo)
```
### 3. Extract history <-> `Message` conversion
This roundtrip is self-contained and can be reused/tested separately.
```python
from astrbot.core.agent.message import Message, ToolCall
def _history_to_messages(self, history: list[dict]) -> list[Message]:
return [
Message(
role=item.get("role", "user"),
content=item.get("content", ""),
tool_calls=item.get("tool_calls"),
tool_call_id=item.get("tool_call_id"),
)
for item in history
]
def _messages_to_history(self, messages: list[Message]) -> list[dict]:
result: list[dict] = []
for msg in messages:
entry = {"role": msg.role}
if msg.content is None:
entry["content"] = None
elif isinstance(msg.content, (str, list)):
entry["content"] = msg.content
else:
entry["content"] = str(msg.content)
if msg.tool_calls is not None:
entry["tool_calls"] = [
tc.model_dump() if isinstance(tc, ToolCall) else tc
for tc in msg.tool_calls
]
if msg.tool_call_id is not None:
entry["tool_call_id"] = msg.tool_call_id
result.append(entry)
return result
```
Usage in `compact`:
```python
messages = self._history_to_messages(history)
compressed = await cm.process(messages, max_context_tokens=max_context_tokens)
result = self._messages_to_history(compressed)
```
### 4. Optional: permission check helper
To align with `reset` and make the permission logic reusable:
```python
def _check_compact_permission(self, message: AstrMessageEvent, compact_cfg, scene):
required_perm = compact_cfg.get(
scene.key,
"admin" if bool(message.get_group_id()) and not scene.is_unique_session else "member",
)
if required_perm == "admin" and message.role != "admin":
message.set_result(
MessageEventResult().message(
f"Compact command requires admin permission in {scene.name} scenario, "
f"you (ID {message.get_sender_id()}) are not admin, cannot perform this action.",
),
)
return False
return True
```
Then in `compact`:
```python
if not self._check_compact_permission(message, compact_cfg, scene):
return
```
These extractions keep the new feature intact but make `compact` shorter, easier to read/test, and aligned with existing configuration/serialization behavior.
</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
+68
to
+76
| async def _try_summary(self, messages: list[Message]) -> list[Message] | None: | ||
| """Attempt LLM summary compression. Returns compressed messages or None.""" | ||
| if not self.config.enable_summary or self._summary_compressor is None: | ||
| return None | ||
| try: | ||
| result = await self._summary_compressor(messages) | ||
| if result is messages: | ||
| return None # compressor chose not to compress | ||
| if len(result) >= len(messages): |
Contributor
There was a problem hiding this comment.
suggestion (bug_risk): Summary compression success is judged only by message count, which may miss real token-size reductions.
The current check len(result) >= len(messages) assumes fewer messages always means fewer tokens, which isn’t necessarily true. A summary can reduce token count while keeping the same number of messages, so _try_summary may wrongly return None and trigger discard logic or warnings even when the context is already within token limits.
Please consider evaluating summary effectiveness using token counts (via token_counter) rather than message count, or remove the len(result) check and rely on the token-guard recheck after disposal to decide if further halving is needed.
Comment on lines
+96
to
+102
| 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: | ||
| return False | ||
| if max_context_tokens <= 0 or tokens <= 0: | ||
| return False | ||
| return (tokens / max_context_tokens) > self.config.token_guard_threshold |
Contributor
There was a problem hiding this comment.
suggestion: Token guard uses max_context_tokens only from the call site, which may be zero and silently disable the trigger.
In process, _token_guard_exceeded returns False whenever max_context_tokens <= 0. In ToolLoopAgentRunner and conversation.compact, max_context_tokens comes from provider.provider_config.get("max_context_tokens", 0), so any provider without this metadata (or without a provider) will silently disable token guarding even when enable_token_guard is True.
If that's unintended, consider falling back to a default when max_context_tokens is non-positive (e.g., a value from ContextConfig), or emit a warning so misconfigured providers are visible to operators.
Suggested implementation:
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 `max_context_tokens` is non-positive, fall back to a configured default.
If no valid maximum can be determined, emit a warning and return False.
"""
if not self.config.enable_token_guard:
return False
if tokens <= 0:
return False
# Prefer the value provided by the caller; fall back to configuration if missing/invalid.
effective_max_context_tokens = max_context_tokens
if effective_max_context_tokens <= 0:
# Fall back to a default from ContextConfig (if available)
effective_max_context_tokens = getattr(self.config, "max_context_tokens", 0)
if effective_max_context_tokens <= 0:
# No usable max context size; token guard is configured but cannot be applied.
# Emit a warning so misconfigured providers are visible to operators.
logger.warning(
"Token guard is enabled but no valid max_context_tokens is configured "
"(received=%s, fallback=%s). Token guarding is effectively disabled.",
max_context_tokens,
getattr(self.config, "max_context_tokens", None),
)
return False
return (tokens / effective_max_context_tokens) > self.config.token_guard_threshold- Ensure a module-level logger is defined (e.g.,
import loggingandlogger = logging.getLogger(__name__)) if it does not already exist inastrbot/core/agent/context/manager.py. - Verify that
self.config(likelyContextConfig) has amax_context_tokensattribute; if it uses a different name for the default context size, update the fallback access (getattr(self.config, "max_context_tokens", 0)) accordingly. - Optionally, if there is already a centralized logging or warning mechanism for configuration issues, replace the
logger.warning(...)call with that existing mechanism to stay consistent with the rest of the codebase.
Comment on lines
+168
to
+174
| tokens_after = self.token_counter.count_tokens(result) | ||
| if self._token_guard_exceeded(tokens_after, max_context_tokens): | ||
| logger.info( | ||
| "Context still exceeds token guard threshold after disposal, " | ||
| "applying halving truncation (unconstrained by retention).", | ||
| ) | ||
|
|
||
| if self.compressor.should_compress( | ||
| result, | ||
| total_tokens, | ||
| self.config.max_context_tokens, | ||
| ): | ||
| result = await self._run_compression(result, total_tokens) | ||
| result = self.truncator.truncate_by_halving(result) |
Contributor
There was a problem hiding this comment.
suggestion (bug_risk): Double-check token counting ignores trusted_token_usage, which may skew guard behavior compared to the initial trigger.
The initial trigger calls count_tokens(result, trusted_token_usage), while the post-disposal check calls count_tokens(result) without the hint. If trusted_token_usage materially changes the token estimate (e.g., provider vs local counting), the second check may diverge from the first and incorrectly apply or skip halving.
Please either pass trusted_token_usage consistently to the double-check, or document why the second pass intentionally uses a different counting strategy, so the guard’s behavior stays consistent.
Suggested implementation:
# 3. double-check（仅 enable_token_guard）
# Use the same trusted_token_usage hint as the initial trigger to keep
# token guard behavior consistent between pre- and post-disposal checks.
tokens_after = self.token_counter.count_tokens(result, trusted_token_usage)
if self._token_guard_exceeded(tokens_after, max_context_tokens):
logger.info(
"Context still exceeds token guard threshold after disposal, "
"applying halving truncation (unconstrained by retention).",
)
result = self.truncator.truncate_by_halving(result)For this change to work, trusted_token_usage must be available in the scope where the double-check is performed. If it is currently only used in the initial trigger, you will need to:
- Ensure
trusted_token_usageis captured or passed into this method (if it is not already). - Thread
trusted_token_usagethrough any intermediate helpers so that both the initial trigger and the post-disposal double-check can use the same hint when callingself.token_counter.count_tokens(...).
Comment on lines
+250
to
+257
| # 解析历史记录 | ||
| import json | ||
|
|
||
| raw_history: str = conv.history or "[]" | ||
| history: list[dict] = ( | ||
| json.loads(raw_history) if isinstance(raw_history, str) else raw_history | ||
| ) | ||
| if not history: |
Contributor
There was a problem hiding this comment.
suggestion (bug_risk): The compact command silently treats non-string conv.history as already-parsed JSON, which may be fragile.
raw_history is annotated as str, but history is set to json.loads(raw_history) only when it’s a string; otherwise history = raw_history. This effectively treats any non-string conv.history as already-parsed history and a valid list[dict]. If conv.history ever contains another shape (e.g., a metadata dict), history will have an unexpected type and later code will still try to iterate it as a list of message dicts.
Consider validating conv.history’s structure (e.g., checking it’s a list of dicts) and raising or logging a clear error when it isn’t, rather than relying solely on the isinstance(raw_history, str) check.
Suggested change
| # 解析历史记录 | |
| import json | |
| raw_history: str = conv.history or "[]" | |
| history: list[dict] = ( | |
| json.loads(raw_history) if isinstance(raw_history, str) else raw_history | |
| ) | |
| if not history: | |
| # 解析历史记录 | |
| import json | |
| raw_history = conv.history or "[]" | |
| # 将字符串形式的历史记录解析为 JSON；非字符串则直接使用 | |
| try: | |
| if isinstance(raw_history, str): | |
| parsed_history = json.loads(raw_history) | |
| else: | |
| parsed_history = raw_history | |
| except Exception: | |
| message.set_result( | |
| MessageEventResult().message("⚠️ Conversation history is invalid JSON and cannot be compacted."), | |
| ) | |
| return | |
| # 验证历史记录结构：必须是由 dict 组成的列表 | |
| if not isinstance(parsed_history, list) or any( | |
| not isinstance(item, dict) for item in parsed_history | |
| ): | |
| message.set_result( | |
| MessageEventResult().message("⚠️ Conversation history has an unexpected structure and cannot be compacted."), | |
| ) | |
| return | |
| history: list[dict] = parsed_history | |
| if not history: |
| if not self._triggers_fired(result, current_tokens, max_context_tokens): | ||
| return result | ||
|
|
||
| # 2. 处置入口：custom > summary > discard |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider extracting the disposal selection and token-guard halving into dedicated helpers to simplify process and make the context management flow clearer.
You can centralize the disposal selection and the token‑guard halving into small helpers to reduce branching in process while preserving the orthogonal trigger/disposal model.
1. Encapsulate disposal ordering into a single helper
Move the “custom > summary > discard” decision into a dedicated helper that returns the chosen compressor (or None). This keeps process focused on the pipeline and makes the disposal strategy easier to reason about.
async def _select_disposal(self, messages: list[Message]) -> list[Message]:
"""Apply disposal strategy: custom > summary > discard."""
# Custom compressor wins if present
if hasattr(self, "_unity_compressor"):
return await self._unity_compressor(messages)
# Summary first
compressed = await self._try_summary(messages)
if compressed is not None:
return compressed
# Then discard
discarded = self._try_discard(messages)
if discarded is not None:
return discarded
logger.warning(
"Context disposal triggered but both summary and discard "
"are unavailable or disabled. No compression applied.",
)
return messagesThen process becomes:
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,
)
if not self._triggers_fired(result, current_tokens, max_context_tokens):
return result
# 处置入口：single call
result = await self._select_disposal(result)
# 3. double-check（仅 enable_token_guard）
result = self._apply_token_guard_halving(result, max_context_tokens)
return result
except Exception:
logger.error("Error during context processing.", exc_info=True)
return messages2. Extract token‑guard halving into a small policy helper
Wrap the “double‑check halving” into its own function so the token guard policy is defined in one place and doesn’t interleave with disposal logic:
def _apply_token_guard_halving(
self,
messages: list[Message],
max_context_tokens: int,
) -> list[Message]:
"""If still over token guard, apply halving truncation."""
if not self.config.enable_token_guard:
return messages
tokens_after = self.token_counter.count_tokens(messages)
if self._token_guard_exceeded(tokens_after, max_context_tokens):
logger.info(
"Context still exceeds token guard threshold after disposal, "
"applying halving truncation (unconstrained by retention).",
)
return self.truncator.truncate_by_halving(messages)
return messagesThis removes duplicated guard checks from process and makes the “double‑check + halving” behavior explicit and easier to maintain, without changing any of the existing functionality or thresholds.
|
|
||
| message.set_result(MessageEventResult().message(ret)) | ||
|
|
||
| async def compact(self, message: AstrMessageEvent) -> None: |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider extracting context config construction, summary provider resolution, history/message conversion, and permission checking into reusable helpers to keep compact shorter and more maintainable.
You can reduce complexity here by extracting a few focused helpers, without changing behavior.
1. Extract ContextConfig construction
You’re manually wiring a lot of context settings; this is reusable and matches behavior in other agents/runners.
# somewhere reusable (e.g., a utils module or on self.context)
def build_context_config_from_settings(settings, summary_provider):
from astrbot.core.agent.context.config import ContextConfig
return ContextConfig(
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider=summary_provider,
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
)Usage in compact:
summary_provider = self._resolve_summary_provider(settings, umo)
config = build_context_config_from_settings(settings, summary_provider)
cm = ContextManager(config)2. Extract summary provider resolution
This logic is repeated in other places (e.g. _get_compress_provider); a shared helper keeps behavior consistent.
def _resolve_summary_provider(self, settings, umo):
summary_provider_id = settings.get("summary_provider_id", "")
if summary_provider_id:
provider = self.context.get_provider_by_id(summary_provider_id)
if provider:
return provider
return self.context.get_using_provider(umo=umo)Then:
summary_provider = self._resolve_summary_provider(settings, umo)3. Extract history <-> Message conversion
This roundtrip is self-contained and can be reused/tested separately.
from astrbot.core.agent.message import Message, ToolCall
def _history_to_messages(self, history: list[dict]) -> list[Message]:
return [
Message(
role=item.get("role", "user"),
content=item.get("content", ""),
tool_calls=item.get("tool_calls"),
tool_call_id=item.get("tool_call_id"),
)
for item in history
]
def _messages_to_history(self, messages: list[Message]) -> list[dict]:
result: list[dict] = []
for msg in messages:
entry = {"role": msg.role}
if msg.content is None:
entry["content"] = None
elif isinstance(msg.content, (str, list)):
entry["content"] = msg.content
else:
entry["content"] = str(msg.content)
if msg.tool_calls is not None:
entry["tool_calls"] = [
tc.model_dump() if isinstance(tc, ToolCall) else tc
for tc in msg.tool_calls
]
if msg.tool_call_id is not None:
entry["tool_call_id"] = msg.tool_call_id
result.append(entry)
return resultUsage in compact:
messages = self._history_to_messages(history)
compressed = await cm.process(messages, max_context_tokens=max_context_tokens)
result = self._messages_to_history(compressed)4. Optional: permission check helper
To align with reset and make the permission logic reusable:
def _check_compact_permission(self, message: AstrMessageEvent, compact_cfg, scene):
required_perm = compact_cfg.get(
scene.key,
"admin" if bool(message.get_group_id()) and not scene.is_unique_session else "member",
)
if required_perm == "admin" and message.role != "admin":
message.set_result(
MessageEventResult().message(
f"Compact command requires admin permission in {scene.name} scenario, "
f"you (ID {message.get_sender_id()}) are not admin, cannot perform this action.",
),
)
return False
return TrueThen in compact:
if not self._check_compact_permission(message, compact_cfg, scene):
returnThese extractions keep the new feature intact but make compact shorter, easier to read/test, and aligned with existing configuration/serialization behavior.
Contributor
There was a problem hiding this comment.
Code Review
This pull request redesigns the context management system in AstrBot, transitioning from an implicit strategy-based model to an orthogonal trigger/disposal model. Key updates include the restructuring of ContextConfig and ContextManager to decouple triggers (turn limits and token guards) from disposal actions (LLM summary and turn discarding), the addition of a manual /compact command, and a migration helper to transition old configurations. The feedback focuses on improving the robustness of these changes by addressing potential AttributeErrors, TypeErrors, KeyErrors, and JSONDecodeErrors across the command implementation, context manager, and migration helper.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
|
|
||
| # 权限检查（与 reset 相同模式） | ||
| scene = RstScene.get_scene(is_group, is_unique_session) | ||
| alter_cmd_cfg = await sp.get_async("global", "global", "alter_cmd", {}) |
Contributor
There was a problem hiding this comment.
If alter_cmd_cfg is configured as None or empty in the database, calling .get() on it will raise an AttributeError. It is safer to use a fallback dictionary to prevent potential crashes.
Suggested change
| alter_cmd_cfg = await sp.get_async("global", "global", "alter_cmd", {}) | |
| alter_cmd_cfg = await sp.get_async("global", "global", "alter_cmd", {}) or {} |
Comment on lines
+253
to
+256
| raw_history: str = conv.history or "[]" | ||
| history: list[dict] = ( | ||
| json.loads(raw_history) if isinstance(raw_history, str) else raw_history | ||
| ) |
Contributor
There was a problem hiding this comment.
If conv.history contains malformed JSON, json.loads will raise a JSONDecodeError which is not caught outside of the main compression block. Wrapping the history parsing in a try-except block ensures that any corruption in the conversation history is handled gracefully and reported to the user.
Suggested change
| raw_history: str = conv.history or "[]" | |
| history: list[dict] = ( | |
| json.loads(raw_history) if isinstance(raw_history, str) else raw_history | |
| ) | |
| raw_history = conv.history or "[]" | |
| try: | |
| history: list[dict] = ( | |
| json.loads(raw_history) if isinstance(raw_history, str) else raw_history | |
| ) | |
| except json.JSONDecodeError: | |
| message.set_result( | |
| MessageEventResult().message("❌ Failed to parse conversation history. It might be corrupted."), | |
| ) | |
| return |
Comment on lines
+308
to
+309
| max_context_tokens = provider.provider_config.get("max_context_tokens", 0) if provider else 0 | ||
|
|
Contributor
There was a problem hiding this comment.
If provider is not None but provider.provider_config is missing or None, accessing it directly will raise an AttributeError. Adding a check for provider.provider_config makes this lookup more robust.
provider = self.context.get_using_provider(umo=umo)
max_context_tokens = (
provider.provider_config.get("max_context_tokens", 0)
if provider and getattr(provider, "provider_config", None)
else 0
)
Comment on lines
+73
to
+77
| result = await self._summary_compressor(messages) | ||
| if result is messages: | ||
| return None # compressor chose not to compress | ||
| if len(result) >= len(messages): | ||
| return None # no effective reduction |
Contributor
There was a problem hiding this comment.
If self._summary_compressor returns None (e.g., if it fails or decides not to compress), calling len(result) will raise a TypeError. Checking if result is None first prevents this exception.
Suggested change
| result = await self._summary_compressor(messages) | |
| if result is messages: | |
| return None # compressor chose not to compress | |
| if len(result) >= len(messages): | |
| return None # no effective reduction | |
| result = await self._summary_compressor(messages) | |
| if result is None or result is messages: | |
| return None # compressor chose not to compress | |
| if len(result) >= len(messages): | |
| return None # no effective reduction |
Comment on lines
+186
to
+190
| _migra_context_config(astrbot_config["provider_settings"]) | ||
| _validate_context_config(astrbot_config["provider_settings"]) | ||
| for conf in acm.confs.values(): | ||
| _migra_context_config(conf["provider_settings"]) | ||
| _validate_context_config(conf["provider_settings"]) |
Contributor
There was a problem hiding this comment.
If astrbot_config or any of the configurations in acm.confs.values() do not contain the "provider_settings" key, accessing it directly will raise a KeyError and crash the migration process. Checking for the existence of "provider_settings" first ensures a safe and robust migration.
Suggested change
| _migra_context_config(astrbot_config["provider_settings"]) | |
| _validate_context_config(astrbot_config["provider_settings"]) | |
| for conf in acm.confs.values(): | |
| _migra_context_config(conf["provider_settings"]) | |
| _validate_context_config(conf["provider_settings"]) | |
| if "provider_settings" in astrbot_config: | |
| _migra_context_config(astrbot_config["provider_settings"]) | |
| _validate_context_config(astrbot_config["provider_settings"]) | |
| for conf in acm.confs.values(): | |
| if "provider_settings" in conf: | |
| _migra_context_config(conf["provider_settings"]) | |
| _validate_context_config(conf["provider_settings"]) |
Comment on lines
+204
to
+211
| if "max_context_length" in ps: | ||
| old_val = ps.pop("max_context_length") | ||
| migrated = True | ||
| if old_val > 0: | ||
| ps["enable_turn_limit"] = True | ||
| ps["max_turns"] = old_val | ||
| else: | ||
| ps["enable_turn_limit"] = False |
Contributor
There was a problem hiding this comment.
If old_val is None or not a numeric type, the comparison old_val > 0 will raise a TypeError and crash the migration. Checking the type of old_val before comparison makes the migration more resilient.
Suggested change
| if "max_context_length" in ps: | |
| old_val = ps.pop("max_context_length") | |
| migrated = True | |
| if old_val > 0: | |
| ps["enable_turn_limit"] = True | |
| ps["max_turns"] = old_val | |
| else: | |
| ps["enable_turn_limit"] = False | |
| if "max_context_length" in ps: | |
| old_val = ps.pop("max_context_length") | |
| migrated = True | |
| if isinstance(old_val, (int, float)) and old_val > 0: | |
| ps["enable_turn_limit"] = True | |
| ps["max_turns"] = int(old_val) | |
| else: | |
| ps["enable_turn_limit"] = False |
- Wrap json.loads in try/except to handle corrupted conversation history - Validate history structure (list[dict]) before processing in /compact - Pass trusted_token_usage consistently to post-disposal token guard check - Guard against None alter_cmd_cfg and provider_config in /compact - Log warning when token guard is enabled but max_context_tokens is 0 so misconfigured providers are visible to operators Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
- Extract _check_command_permission, _build_context_config, _history_to_messages, _messages_to_history helpers for /compact and /reset, reducing code duplication - Extract _select_disposal method in ContextManager, replace hasattr with explicit is not None checks - Add result is None guard in _try_summary for robustness - Add isinstance type check for max_context_length migration - Fix migration bug: retention_method now matches strategy (percentage for llm_compress, turns for truncate_by_turns) instead of always being overwritten to percentage Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
Contributor
Author
|
/gemini review |
Contributor
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
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
retention_methodand other context-management mode flags are represented as free-form strings; consider switching these to an enum orLiteraltypes to avoid typos and make invalid values fail fast rather than relying only on_validate_context_configwarnings. - In
ContextManager.process, token-guard behavior depends on amax_context_tokensargument being correctly supplied by every caller; it might be more robust to default this from the provider/config (or assert whenenable_token_guard=Truebutmax_context_tokens<=0) so miswired callers don't silently disable guarding. - The context config migration/validation helpers in
migra_helpercurrently log warnings but give no structured feedback to callers; consider returning a summary of applied migrations and validation issues so higher-level code can surface configuration problems more clearly than log inspection alone.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The new `retention_method` and other context-management mode flags are represented as free-form strings; consider switching these to an enum or `Literal` types to avoid typos and make invalid values fail fast rather than relying only on `_validate_context_config` warnings.
- In `ContextManager.process`, token-guard behavior depends on a `max_context_tokens` argument being correctly supplied by every caller; it might be more robust to default this from the provider/config (or assert when `enable_token_guard=True` but `max_context_tokens<=0`) so miswired callers don't silently disable guarding.
- The context config migration/validation helpers in `migra_helper` currently log warnings but give no structured feedback to callers; consider returning a summary of applied migrations and validation issues so higher-level code can surface configuration problems more clearly than log inspection alone.
## Individual Comments
### Comment 1
<location path="astrbot/core/agent/context/manager.py" line_range="98-108" />
<code_context>
- result,
- trusted_token_usage,
- )
+ 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:
+ return False
+ if max_context_tokens <= 0:
+ 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,
+ )
</code_context>
<issue_to_address>
**suggestion:** Consider avoiding repeated warnings when token guard is enabled but `max_context_tokens` is 0.
With `enable_token_guard=True` and `max_context_tokens <= 0`, this method logs the warning on every call. For sessions/providers without a configured max context window, this could flood the logs since `process` may be invoked on every step and via `/compact`. Consider emitting this warning only once per `ContextManager` (or per provider), or downgrading subsequent occurrences to debug level.
```suggestion
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:
return False
if max_context_tokens <= 0:
# Avoid flooding logs: warn once per instance, debug thereafter.
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:
logger.debug(
"Token guard is enabled but max_context_tokens is %s; "
"token guarding remains effectively disabled.",
max_context_tokens,
)
return False
```
</issue_to_address>
### Comment 2
<location path="astrbot/core/utils/migra_helper.py" line_range="264-269" />
<code_context>
+ if ps.get("enable_turn_limit") and ps.get("max_turns", 50) < 2:
+ logger.warning("max_turns should be >= 2 when enable_turn_limit is True.")
+
+ if _has("token_guard_threshold"):
+ val = ps["token_guard_threshold"]
+ if not (0.5 <= val <= 0.99):
+ logger.warning(
+ "token_guard_threshold %.2f is outside recommended range [0.5, 0.99].",
+ val,
+ )
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Type assumptions in `_validate_context_config` can raise at runtime for malformed configs.
Because fields like `token_guard_threshold` and `retain_percentage` are compared as numbers (e.g. `0.5 <= val <= 0.99`), user configs that provide them as strings (common in JSON/YAML) will cause a `TypeError` and abort migration. Since this helper is meant to validate and warn rather than fail, consider guarding these checks with `isinstance(val, (int, float))` (or a `try/except TypeError`) so malformed types only trigger warnings instead of stopping the migration.
</issue_to_address>
### Comment 3
<location path="astrbot/builtin_stars/builtin_commands/commands/conversation.py" line_range="268" />
<code_context>
message.set_result(MessageEventResult().message(ret))
+ async def compact(self, message: AstrMessageEvent) -> None:
+ """手动触发上下文的处置与压缩"""
+ umo = message.unified_msg_origin
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the history parsing and compaction logic into a dedicated helper so that the `compact` command only orchestrates permissions, conversation lookup, and user messaging.
You can reduce complexity by pulling the history/conversation compaction into a dedicated helper and leaving `compact` as orchestration + messaging only. This keeps all behavior but separates responsibilities and makes the command easier to read and test.
### 1. Extract conversation compaction into a helper
Move the JSON/history parsing, `ContextConfig` building, `ContextManager` invocation, and history re‑serialization into a separate class or function. The command then just handles permissions, basic conversation lookup, and user messages.
For example:
```python
# conversation_compactor.py (new module)
from __future__ import annotations
from typing import Tuple, List, Dict
import json
from astrbot.core.agent.context.manager import ContextManager
from astrbot.core.agent.context.config import ContextConfig
from astrbot.core.agent.message import Message, ToolCall
from loguru import logger
class ConversationCompactor:
def __init__(self, context):
self.context = context
def _build_context_config(self, settings: dict, umo: str) -> ContextConfig:
summary_provider_id = settings.get("summary_provider_id", "")
summary_provider = (
self.context.get_provider_by_id(summary_provider_id)
if summary_provider_id
else None
)
if not summary_provider:
summary_provider = self.context.get_using_provider(umo=umo)
return ContextConfig(
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider=summary_provider,
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
)
def _history_to_messages(self, history: List[Dict]) -> List[Message]:
return [
Message(
role=item.get("role", "user"),
content=item.get("content", ""),
tool_calls=item.get("tool_calls"),
tool_call_id=item.get("tool_call_id"),
)
for item in history
]
def _messages_to_history(self, messages) -> List[Dict]:
result: List[Dict] = []
for msg in messages:
entry = {"role": msg.role}
if msg.content is None:
entry["content"] = None
elif isinstance(msg.content, (str, list)):
entry["content"] = msg.content
else:
entry["content"] = str(msg.content)
if msg.tool_calls is not None:
entry["tool_calls"] = [
tc.model_dump() if isinstance(tc, ToolCall) else tc
for tc in msg.tool_calls
]
if msg.tool_call_id is not None:
entry["tool_call_id"] = msg.tool_call_id
result.append(entry)
return result
def _parse_history(self, raw_history) -> List[Dict] | None:
raw_history = raw_history or "[]"
try:
parsed = json.loads(raw_history) if isinstance(raw_history, str) else raw_history
except json.JSONDecodeError:
return None
if not isinstance(parsed, list) or any(not isinstance(i, dict) for i in parsed):
return None
return parsed
async def compact(
self,
umo: str,
settings: dict,
conv,
) -> Tuple[bool, str | None, List[Dict] | None]:
"""
Returns (ok, error_message, new_history).
error_message is user-facing if not ok.
"""
history = self._parse_history(conv.history)
if history is None:
return False, "❌ Conversation history is corrupted or has an unexpected structure and cannot be compacted.", None
if not history:
return False, "ℹ️ Conversation is empty, nothing to compact.", None
original_len = len(history)
messages = self._history_to_messages(history)
config = self._build_context_config(settings, umo)
cm = ContextManager(config)
provider = self.context.get_using_provider(umo=umo)
max_context_tokens = (
provider.provider_config.get("max_context_tokens", 0)
if provider and getattr(provider, "provider_config", None)
else 0
)
try:
compressed = await cm.process(messages, max_context_tokens=max_context_tokens)
except Exception:
logger.error("Context compression failed.", exc_info=True)
return False, "❌ Context compression failed. See logs for details.", None
result = self._messages_to_history(compressed)
removed = original_len - len(result)
success_msg = (
f"✅ Context compressed: {removed} messages removed "
f"({original_len} → {len(result)})."
)
return True, success_msg, result
```
Then your command becomes much simpler and focused:
```python
# in the command class
from .conversation_compactor import ConversationCompactor
async def compact(self, message: AstrMessageEvent) -> None:
"""手动触发上下文的处置与压缩"""
umo = message.unified_msg_origin
cfg = self.context.get_config(umo=umo)
settings = cfg["provider_settings"]
is_unique_session = cfg["platform_settings"]["unique_session"]
is_group = bool(message.get_group_id())
if not await self._check_command_permission("compact", message, is_group, is_unique_session):
return
agent_runner_type = settings.get("agent_runner_type", "")
if agent_runner_type in THIRD_PARTY_AGENT_RUNNER_KEY:
message.set_result(
MessageEventResult().message(
"ℹ️ Compact is not supported for third-party agent runners "
f"({agent_runner_type}). Use /reset instead."
),
)
return
cid = await self.context.conversation_manager.get_curr_conversation_id(umo)
if not cid:
message.set_result(
MessageEventResult().message(
"😕 You are not in a conversation. Use /new to create one.",
),
)
return
conv = await self.context.conversation_manager.get_conversation(umo, cid)
if not conv:
message.set_result(
MessageEventResult().message("😕 Conversation not found."),
)
return
compactor = ConversationCompactor(self.context)
ok, user_msg, new_history = await compactor.compact(umo, settings, conv)
if not ok:
message.set_result(MessageEventResult().message(user_msg))
return
await self.context.conversation_manager.update_conversation(umo, cid, new_history)
message.set_result(MessageEventResult().message(user_msg))
```
This keeps all behavior intact but:
- `compact` is now mostly orchestration and user messaging.
- History parsing, `ContextConfig` wiring, `ContextManager` usage, and message/history conversions are centralized and reusable.
- The command module no longer needs `_build_context_config`, `_history_to_messages`, or `_messages_to_history`; these live in a context/history–focused helper closer to the domain logic.
</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
+98
to
+108
| 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: | ||
| return False | ||
| if max_context_tokens <= 0: | ||
| 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, | ||
| ) |
Contributor
There was a problem hiding this comment.
suggestion: Consider avoiding repeated warnings when token guard is enabled but max_context_tokens is 0.
With enable_token_guard=True and max_context_tokens <= 0, this method logs the warning on every call. For sessions/providers without a configured max context window, this could flood the logs since process may be invoked on every step and via /compact. Consider emitting this warning only once per ContextManager (or per provider), or downgrading subsequent occurrences to debug level.
Suggested change
| 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: | |
| return False | |
| if max_context_tokens <= 0: | |
| 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, | |
| ) | |
| 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: | |
| return False | |
| if max_context_tokens <= 0: | |
| # Avoid flooding logs: warn once per instance, debug thereafter. | |
| 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: | |
| logger.debug( | |
| "Token guard is enabled but max_context_tokens is %s; " | |
| "token guarding remains effectively disabled.", | |
| max_context_tokens, | |
| ) | |
| return False |
Comment on lines
+264
to
+269
| if _has("token_guard_threshold"): | ||
| val = ps["token_guard_threshold"] | ||
| if not (0.5 <= val <= 0.99): | ||
| logger.warning( | ||
| "token_guard_threshold %.2f is outside recommended range [0.5, 0.99].", | ||
| val, |
Contributor
There was a problem hiding this comment.
issue (bug_risk): Type assumptions in _validate_context_config can raise at runtime for malformed configs.
Because fields like token_guard_threshold and retain_percentage are compared as numbers (e.g. 0.5 <= val <= 0.99), user configs that provide them as strings (common in JSON/YAML) will cause a TypeError and abort migration. Since this helper is meant to validate and warn rather than fail, consider guarding these checks with isinstance(val, (int, float)) (or a try/except TypeError) so malformed types only trigger warnings instead of stopping the migration.
|
|
||
| message.set_result(MessageEventResult().message(ret)) | ||
|
|
||
| async def compact(self, message: AstrMessageEvent) -> None: |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider extracting the history parsing and compaction logic into a dedicated helper so that the compact command only orchestrates permissions, conversation lookup, and user messaging.
You can reduce complexity by pulling the history/conversation compaction into a dedicated helper and leaving compact as orchestration + messaging only. This keeps all behavior but separates responsibilities and makes the command easier to read and test.
1. Extract conversation compaction into a helper
Move the JSON/history parsing, ContextConfig building, ContextManager invocation, and history re‑serialization into a separate class or function. The command then just handles permissions, basic conversation lookup, and user messages.
For example:
# conversation_compactor.py (new module)
from __future__ import annotations
from typing import Tuple, List, Dict
import json
from astrbot.core.agent.context.manager import ContextManager
from astrbot.core.agent.context.config import ContextConfig
from astrbot.core.agent.message import Message, ToolCall
from loguru import logger
class ConversationCompactor:
def __init__(self, context):
self.context = context
def _build_context_config(self, settings: dict, umo: str) -> ContextConfig:
summary_provider_id = settings.get("summary_provider_id", "")
summary_provider = (
self.context.get_provider_by_id(summary_provider_id)
if summary_provider_id
else None
)
if not summary_provider:
summary_provider = self.context.get_using_provider(umo=umo)
return ContextConfig(
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider=summary_provider,
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
)
def _history_to_messages(self, history: List[Dict]) -> List[Message]:
return [
Message(
role=item.get("role", "user"),
content=item.get("content", ""),
tool_calls=item.get("tool_calls"),
tool_call_id=item.get("tool_call_id"),
)
for item in history
]
def _messages_to_history(self, messages) -> List[Dict]:
result: List[Dict] = []
for msg in messages:
entry = {"role": msg.role}
if msg.content is None:
entry["content"] = None
elif isinstance(msg.content, (str, list)):
entry["content"] = msg.content
else:
entry["content"] = str(msg.content)
if msg.tool_calls is not None:
entry["tool_calls"] = [
tc.model_dump() if isinstance(tc, ToolCall) else tc
for tc in msg.tool_calls
]
if msg.tool_call_id is not None:
entry["tool_call_id"] = msg.tool_call_id
result.append(entry)
return result
def _parse_history(self, raw_history) -> List[Dict] | None:
raw_history = raw_history or "[]"
try:
parsed = json.loads(raw_history) if isinstance(raw_history, str) else raw_history
except json.JSONDecodeError:
return None
if not isinstance(parsed, list) or any(not isinstance(i, dict) for i in parsed):
return None
return parsed
async def compact(
self,
umo: str,
settings: dict,
conv,
) -> Tuple[bool, str | None, List[Dict] | None]:
"""
Returns (ok, error_message, new_history).
error_message is user-facing if not ok.
"""
history = self._parse_history(conv.history)
if history is None:
return False, "❌ Conversation history is corrupted or has an unexpected structure and cannot be compacted.", None
if not history:
return False, "ℹ️ Conversation is empty, nothing to compact.", None
original_len = len(history)
messages = self._history_to_messages(history)
config = self._build_context_config(settings, umo)
cm = ContextManager(config)
provider = self.context.get_using_provider(umo=umo)
max_context_tokens = (
provider.provider_config.get("max_context_tokens", 0)
if provider and getattr(provider, "provider_config", None)
else 0
)
try:
compressed = await cm.process(messages, max_context_tokens=max_context_tokens)
except Exception:
logger.error("Context compression failed.", exc_info=True)
return False, "❌ Context compression failed. See logs for details.", None
result = self._messages_to_history(compressed)
removed = original_len - len(result)
success_msg = (
f"✅ Context compressed: {removed} messages removed "
f"({original_len} → {len(result)})."
)
return True, success_msg, resultThen your command becomes much simpler and focused:
# in the command class
from .conversation_compactor import ConversationCompactor
async def compact(self, message: AstrMessageEvent) -> None:
"""手动触发上下文的处置与压缩"""
umo = message.unified_msg_origin
cfg = self.context.get_config(umo=umo)
settings = cfg["provider_settings"]
is_unique_session = cfg["platform_settings"]["unique_session"]
is_group = bool(message.get_group_id())
if not await self._check_command_permission("compact", message, is_group, is_unique_session):
return
agent_runner_type = settings.get("agent_runner_type", "")
if agent_runner_type in THIRD_PARTY_AGENT_RUNNER_KEY:
message.set_result(
MessageEventResult().message(
"ℹ️ Compact is not supported for third-party agent runners "
f"({agent_runner_type}). Use /reset instead."
),
)
return
cid = await self.context.conversation_manager.get_curr_conversation_id(umo)
if not cid:
message.set_result(
MessageEventResult().message(
"😕 You are not in a conversation. Use /new to create one.",
),
)
return
conv = await self.context.conversation_manager.get_conversation(umo, cid)
if not conv:
message.set_result(
MessageEventResult().message("😕 Conversation not found."),
)
return
compactor = ConversationCompactor(self.context)
ok, user_msg, new_history = await compactor.compact(umo, settings, conv)
if not ok:
message.set_result(MessageEventResult().message(user_msg))
return
await self.context.conversation_manager.update_conversation(umo, cid, new_history)
message.set_result(MessageEventResult().message(user_msg))This keeps all behavior intact but:
compactis now mostly orchestration and user messaging.- History parsing,
ContextConfigwiring,ContextManagerusage, and message/history conversions are centralized and reusable. - The command module no longer needs
_build_context_config,_history_to_messages, or_messages_to_history; these live in a context/history–focused helper closer to the domain logic.
…al type - Warn once per ContextManager instance when max_context_tokens is 0, then downgrade to debug to avoid log flooding - Add isinstance guards for token_guard_threshold and retain_percentage in _validate_context_config to prevent TypeError from malformed configs - Change retention_method type from str to Literal for static type-checking support Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
Contributor
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - I've found 2 issues, and left some high level feedback:
- The construction of ContextConfig is now spread across several places (e.g. ConversationCommands._build_context_config, ToolLoopAgentRunner.reset, default provider_settings), which makes it easy for defaults or semantics to drift; consider centralizing this into a single factory/helper that derives ContextConfig from provider_settings and provider metadata so all call sites stay consistent.
- The /compact command uses hard-coded English (and emoji) strings for user-visible responses while other parts of the dashboard/config have been wired through i18n; you may want to export these messages through the same i18n mechanism to keep localization behavior consistent.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The construction of ContextConfig is now spread across several places (e.g. ConversationCommands._build_context_config, ToolLoopAgentRunner.reset, default provider_settings), which makes it easy for defaults or semantics to drift; consider centralizing this into a single factory/helper that derives ContextConfig from provider_settings and provider metadata so all call sites stay consistent.
- The /compact command uses hard-coded English (and emoji) strings for user-visible responses while other parts of the dashboard/config have been wired through i18n; you may want to export these messages through the same i18n mechanism to keep localization behavior consistent.
## Individual Comments
### Comment 1
<location path="astrbot/core/agent/runners/tool_loop_agent_runner.py" line_range="242-243" />
<code_context>
self._tool_result_token_counter = EstimateTokenCounter()
+
+ # Provider-level context window size (needed for token guard trigger)
+ self._max_context_tokens: int = provider.provider_config.get(
+ "max_context_tokens", 0,
+ )
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Guard against providers without `provider_config` when reading max_context_tokens.
This line assumes `provider_config` always exists, unlike other paths (e.g. `conversation.compact`) that use `getattr` for safety. A custom/third‑party `Provider` without `provider_config` will raise an `AttributeError` here. Consider matching the safer pattern:
```python
cfg = getattr(provider, "provider_config", {}) or {}
self._max_context_tokens = cfg.get("max_context_tokens", 0)
```
This preserves current behavior when `provider_config` is defined while avoiding hard failures for alternative providers.
</issue_to_address>
### Comment 2
<location path="astrbot/builtin_stars/builtin_commands/commands/conversation.py" line_range="195-200" />
<code_context>
+ for item in history
+ ]
+
+ def _messages_to_history(self, messages):
+ from astrbot.core.agent.message import ToolCall
+
+ result: list[dict] = []
+ for msg in messages:
+ entry = {"role": msg.role}
+ if msg.content is None:
+ entry["content"] = None
+ elif isinstance(msg.content, (str, list)):
+ entry["content"] = msg.content
+ else:
+ entry["content"] = str(msg.content)
+ if msg.tool_calls is not None:
+ entry["tool_calls"] = [
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Preserve structured content types instead of blindly coercing to string.
In `_messages_to_history`, any `msg.content` that isn’t `str` or `list` is coerced to `str`. If providers or tools use richer structures (e.g., dicts with metadata), this will be flattened and may break consumers that rely on the original structure. Consider preserving JSON‑like types instead of coercing them:
```python
elif isinstance(msg.content, (str, list, dict)):
entry["content"] = msg.content
else:
entry["content"] = str(msg.content)
```
or more generally, first try JSON serialization and only fall back to `str` when serialization fails.
```suggestion
if msg.content is None:
entry["content"] = None
elif isinstance(msg.content, (str, list, dict)):
entry["content"] = msg.content
else:
entry["content"] = str(msg.content)
```
</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
+242
to
+243
| self._max_context_tokens: int = provider.provider_config.get( | ||
| "max_context_tokens", 0, |
Contributor
There was a problem hiding this comment.
issue (bug_risk): Guard against providers without provider_config when reading max_context_tokens.
This line assumes provider_config always exists, unlike other paths (e.g. conversation.compact) that use getattr for safety. A custom/third‑party Provider without provider_config will raise an AttributeError here. Consider matching the safer pattern:
cfg = getattr(provider, "provider_config", {}) or {}
self._max_context_tokens = cfg.get("max_context_tokens", 0)This preserves current behavior when provider_config is defined while avoiding hard failures for alternative providers.
Comment on lines
+195
to
+200
| if msg.content is None: | ||
| entry["content"] = None | ||
| elif isinstance(msg.content, (str, list)): | ||
| entry["content"] = msg.content | ||
| else: | ||
| entry["content"] = str(msg.content) |
Contributor
There was a problem hiding this comment.
suggestion (bug_risk): Preserve structured content types instead of blindly coercing to string.
In _messages_to_history, any msg.content that isn’t str or list is coerced to str. If providers or tools use richer structures (e.g., dicts with metadata), this will be flattened and may break consumers that rely on the original structure. Consider preserving JSON‑like types instead of coercing them:
elif isinstance(msg.content, (str, list, dict)):
entry["content"] = msg.content
else:
entry["content"] = str(msg.content)or more generally, first try JSON serialization and only fall back to str when serialization fails.
Suggested change
| if msg.content is None: | |
| entry["content"] = None | |
| elif isinstance(msg.content, (str, list)): | |
| entry["content"] = msg.content | |
| else: | |
| entry["content"] = str(msg.content) | |
| if msg.content is None: | |
| entry["content"] = None | |
| elif isinstance(msg.content, (str, list, dict)): | |
| entry["content"] = msg.content | |
| else: | |
| entry["content"] = str(msg.content) |
- Use getattr guard for provider_config when reading max_context_tokens in ToolLoopAgentRunner, matching the pattern in conversation.py - Preserve dict content in _messages_to_history instead of coercing to str, for robustness against non-standard content types Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
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 migration helpers
_migra_context_config/_validate_context_configassumeprovider_settingsexists onastrbot_configand everyacm.confs[...]; consider usingdict.get(...)with a default and early-return when missing to avoid KeyErrors during migration on partially configured or older installations. - The
/compactcommand’s_build_context_confighard-codes default values for the new context fields, which can easily drift fromContextConfig/default.py; it would be more robust to derive these defaults from a single source (e.g. instantiateContextConfigand override fromsettings) instead of duplicating them here.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The migration helpers `_migra_context_config` / `_validate_context_config` assume `provider_settings` exists on `astrbot_config` and every `acm.confs[...]`; consider using `dict.get(...)` with a default and early-return when missing to avoid KeyErrors during migration on partially configured or older installations.
- The `/compact` command’s `_build_context_config` hard-codes default values for the new context fields, which can easily drift from `ContextConfig`/`default.py`; it would be more robust to derive these defaults from a single source (e.g. instantiate `ContextConfig` and override from `settings`) instead of duplicating them here.
## Individual Comments
### Comment 1
<location path="astrbot/core/agent/context/manager.py" line_range="46" />
<code_context>
- Returns:
- The processed message list.
+ def _count_turns(self, messages: list[Message]) -> int:
+ """Count the number of conversation turns (non-system rounds)."""
+ rounds = split_into_rounds(messages)
</code_context>
<issue_to_address>
**issue (complexity):** Consider computing turn counts once in `process` and passing them into helpers so triggers and disposals work from explicit inputs rather than recomputing state internally.
You can reduce some of the current indirection/coupling by computing turn counts once in `process` and threading them through, instead of having each helper recalculate internally. This keeps the orthogonal trigger/disposal model and all existing features, but simplifies control flow and avoids repeated `split_into_rounds` calls.
Concretely:
- `_count_turns` is called from both `_triggers_fired` and `_try_discard`.
- Each helper has to know how to compute turns, which makes reasoning about the lifecycle harder.
- Computing turns once and passing `total_turns` down makes the dataflow more explicit and trims helper responsibilities.
For example:
```python
# process: compute total_turns once and pass it down
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)
if not self._triggers_fired(
total_turns=total_turns,
current_tokens=current_tokens,
max_context_tokens=max_context_tokens,
):
return result
result = await self._select_disposal(result, total_turns)
tokens_after = self.token_counter.count_tokens(result, trusted_token_usage)
if self._token_guard_exceeded(tokens_after, max_context_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
```
Then adjust the helpers to accept `total_turns` instead of recomputing:
```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._token_guard_exceeded(current_tokens, max_context_tokens):
return True
return False
```
```python
def _try_discard(
self,
messages: list[Message],
total_turns: int,
) -> list[Message] | None:
"""Discard oldest turns, bounded by retention."""
if not self.config.enable_discard:
return None
max_discardable = self._compute_discard_limit(total_turns)
if max_discardable <= 0:
return None
requested = min(self.config.discard_turns, max_discardable)
return self.truncator.truncate_by_dropping_oldest_turns(
messages,
drop_turns=requested,
)
```
And update `_select_disposal` to thread `total_turns` into `_try_discard`:
```python
async def _select_disposal(
self,
messages: list[Message],
total_turns: int,
) -> list[Message]:
"""Apply disposal strategy: custom > summary > discard."""
if self._unity_compressor is not None:
return await self._unity_compressor(messages)
compressed = await self._try_summary(messages)
if compressed is not None:
return compressed
discarded = self._try_discard(messages, total_turns=total_turns)
if discarded is not None:
return discarded
logger.warning(
"Context disposal triggered but both summary and discard "
"are unavailable or disabled. No compression applied.",
)
return messages
```
This keeps the orthogonal trigger/disposal design intact but makes the flow more declarative:
- `process` becomes the single place where tokens and turns are computed.
- Triggers and disposals now operate on explicit inputs (tokens, turns) rather than re-deriving them, which reduces both complexity and runtime overhead.
</issue_to_address>
### Comment 2
<location path="astrbot/builtin_stars/builtin_commands/commands/conversation.py" line_range="149" />
<code_context>
+ return False
+ return True
+
+ def _build_context_config(self, settings: dict, umo: str):
+ from astrbot.core.agent.context.config import ContextConfig
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the new history conversion and context wiring logic into shared utilities so this command handler stays a thin orchestrator over the context engine.
You can reduce the new complexity by pushing the low‑level “context engine” wiring into a small reusable utility module and keeping this command handler focused on orchestration.
### 1. Extract history ↔ Message conversion
`_history_to_messages` and `_messages_to_history` are generic and don’t need access to `self`. Moving them to a shared utility makes them reusable and reduces the cognitive load in this class.
**Before (methods on the command class):**
```python
def _history_to_messages(self, history: list[dict]):
from astrbot.core.agent.message import Message
return [
Message(
role=item.get("role", "user"),
content=item.get("content", ""),
tool_calls=item.get("tool_calls"),
tool_call_id=item.get("tool_call_id"),
)
for item in history
]
def _messages_to_history(self, messages):
from astrbot.core.agent.message import ToolCall
result: list[dict] = []
for msg in messages:
entry = {"role": msg.role}
if msg.content is None:
entry["content"] = None
elif isinstance(msg.content, (str, list, dict)):
entry["content"] = msg.content
else:
entry["content"] = str(msg.content)
if msg.tool_calls is not None:
entry["tool_calls"] = [
tc.model_dump() if isinstance(tc, ToolCall) else tc
for tc in msg.tool_calls
]
if msg.tool_call_id is not None:
entry["tool_call_id"] = msg.tool_call_id
result.append(entry)
return result
```
**After (shared util):**
```python
# astrbot/core/agent/context/history_utils.py
from typing import List, Dict, Any
from astrbot.core.agent.message import Message, ToolCall
def history_to_messages(history: List[Dict[str, Any]]) -> List[Message]:
return [
Message(
role=item.get("role", "user"),
content=item.get("content", ""),
tool_calls=item.get("tool_calls"),
tool_call_id=item.get("tool_call_id"),
)
for item in history
]
def messages_to_history(messages: List[Message]) -> List[Dict[str, Any]]:
result: List[Dict[str, Any]] = []
for msg in messages:
entry: Dict[str, Any] = {"role": msg.role}
if msg.content is None:
entry["content"] = None
elif isinstance(msg.content, (str, list, dict)):
entry["content"] = msg.content
else:
entry["content"] = str(msg.content)
if msg.tool_calls is not None:
entry["tool_calls"] = [
tc.model_dump() if isinstance(tc, ToolCall) else tc
for tc in msg.tool_calls
]
if msg.tool_call_id is not None:
entry["tool_call_id"] = msg.tool_call_id
result.append(entry)
return result
```
Usage in `compact` becomes simpler:
```python
from astrbot.core.agent.context.history_utils import (
history_to_messages,
messages_to_history,
)
# ...
messages = history_to_messages(history)
compressed = await cm.process(messages, max_context_tokens=max_context_tokens)
result = messages_to_history(compressed)
```
### 2. Extract ContextConfig / ContextManager wiring
`_build_context_config` + `ContextManager` construction + token‑window extraction is effectively context-engine wiring. Extracting that into a helper keeps `compact` from being a mini engine.
**Before (inside command class):**
```python
def _build_context_config(self, settings: dict, umo: str):
from astrbot.core.agent.context.config import ContextConfig
summary_provider_id = settings.get("summary_provider_id", "")
summary_provider = (
self.context.get_provider_by_id(summary_provider_id)
if summary_provider_id
else None
)
if not summary_provider:
summary_provider = self.context.get_using_provider(umo=umo)
return ContextConfig(
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider=summary_provider,
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
)
```
**After (utility function):**
```python
# astrbot/core/agent/context/context_utils.py
from astrbot.core.agent.context.config import ContextConfig
def build_context_config(context, settings: dict, umo: str) -> ContextConfig:
summary_provider_id = settings.get("summary_provider_id", "")
summary_provider = (
context.get_provider_by_id(summary_provider_id)
if summary_provider_id
else None
)
if not summary_provider:
summary_provider = context.get_using_provider(umo=umo)
return ContextConfig(
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider=summary_provider,
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
)
```
Then in `compact`:
```python
from astrbot.core.agent.context.manager import ContextManager
from astrbot.core.agent.context.context_utils import build_context_config
# ...
config = build_context_config(self.context, settings, umo)
cm = ContextManager(config)
provider = self.context.get_using_provider(umo=umo)
max_context_tokens = (
provider.provider_config.get("max_context_tokens", 0)
if provider and getattr(provider, "provider_config", None)
else 0
)
compressed = await cm.process(messages, max_context_tokens=max_context_tokens)
```
### 3. Keep `_check_command_permission` focused
You’ve already improved reuse between `reset` and `compact`; the remaining coupling is to `message.set_result`. If you want further decoupling without changing behavior, you can return an error message instead of mutating `message` in the helper:
```python
async def _check_command_permission(... ) -> tuple[bool, str | None]:
# ... compute required_perm ...
if required_perm == "admin" and message.role != "admin":
return False, (
f"{command.capitalize()} command requires admin permission in {scene.name} scenario, "
f"you (ID {message.get_sender_id()}) are not admin, cannot perform this action."
)
return True, None
```
Usage:
```python
ok, err = await self._check_command_permission("compact", message, is_group, is_unique_session)
if not ok:
message.set_result(MessageEventResult().message(err))
return
```
This keeps permission logic reusable while making the side effect (how the error is surfaced) explicit in the command handler.
All of these changes preserve behavior while making the command file less of a context‑management orchestrator and more of a thin coordinator over shared utilities.
</issue_to_address>
### Comment 3
<location path="astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py" line_range="118" />
<code_context>
if self.dequeue_context_length <= 0:
self.dequeue_context_length = 1
+
+ # New orthogonal fields
+ self.enable_turn_limit: bool = settings.get("enable_turn_limit", False)
+ self.max_turns: int = settings.get("max_turns", 50)
</code_context>
<issue_to_address>
**issue (complexity):** Consider encapsulating the new context-related parameters into a dedicated config object to avoid a growing list of separate attributes and constructor arguments.
You can tame the parameter/attribute explosion by grouping the new orthogonal context fields into a dedicated config object, built once from `settings` and then passed through. This keeps `initialize` lean and reduces the chance of desync between attributes and constructor args.
### 1. Introduce a `ContextConfig` (or similar) object
For example:
```python
from dataclasses import dataclass
@dataclass
class ContextConfig:
max_context_length: int
dequeue_context_length: int
enable_turn_limit: bool
max_turns: int
enable_token_guard: bool
token_guard_threshold: float
enable_summary: bool
enable_discard: bool
discard_turns: int
summary_prompt: str
summary_provider_id: str
retention_method: str
retain_turns: int
retain_percentage: float
fallback_max_context_tokens: int
```
### 2. Build this object in `initialize` instead of many instance vars
```python
max_context_length = settings.get("max_context_length", -1)
dequeue_context_length = min(
max(1, settings.get("dequeue_context_length", 1)),
max(1, max_context_length - 1),
)
context_config = ContextConfig(
max_context_length=max_context_length,
dequeue_context_length=dequeue_context_length,
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider_id=settings.get("summary_provider_id", ""),
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
fallback_max_context_tokens=settings.get("fallback_max_context_tokens", 128000),
)
self.context_config = context_config # if needed elsewhere
```
You still preserve all existing behavior, but `initialize` now has a single place where context-related settings are defined.
### 3. Pass the config object instead of many independent args
If you can adjust `MainAgentBuildConfig`, prefer:
```python
agent = MainAgentBuildConfig(
# ... existing args ...
context_config=context_config,
llm_safety_mode=self.llm_safety_mode,
safety_mode_strategy=self.safety_mode_strategy,
# ...
)
```
If you must keep the flat signature for now, you can still reduce duplication by unpacking:
```python
agent = MainAgentBuildConfig(
# ... existing args ...
**asdict(context_config),
llm_safety_mode=self.llm_safety_mode,
safety_mode_strategy=self.safety_mode_strategy,
# ...
)
```
This isolates the “context management” surface area into one well-typed structure, makes future additions/changes localized, and prevents `initialize` from turning into a long list of orthogonal attributes and mirrored constructor parameters.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
|
||
| Returns: | ||
| The processed message list. | ||
| def _count_turns(self, messages: list[Message]) -> int: |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider computing turn counts once in process and passing them into helpers so triggers and disposals work from explicit inputs rather than recomputing state internally.
You can reduce some of the current indirection/coupling by computing turn counts once in process and threading them through, instead of having each helper recalculate internally. This keeps the orthogonal trigger/disposal model and all existing features, but simplifies control flow and avoids repeated split_into_rounds calls.
Concretely:
_count_turnsis called from both_triggers_firedand_try_discard.- Each helper has to know how to compute turns, which makes reasoning about the lifecycle harder.
- Computing turns once and passing
total_turnsdown makes the dataflow more explicit and trims helper responsibilities.
For example:
# process: compute total_turns once and pass it down
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)
if not self._triggers_fired(
total_turns=total_turns,
current_tokens=current_tokens,
max_context_tokens=max_context_tokens,
):
return result
result = await self._select_disposal(result, total_turns)
tokens_after = self.token_counter.count_tokens(result, trusted_token_usage)
if self._token_guard_exceeded(tokens_after, max_context_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 messagesThen adjust the helpers to accept total_turns instead of recomputing:
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._token_guard_exceeded(current_tokens, max_context_tokens):
return True
return Falsedef _try_discard(
self,
messages: list[Message],
total_turns: int,
) -> list[Message] | None:
"""Discard oldest turns, bounded by retention."""
if not self.config.enable_discard:
return None
max_discardable = self._compute_discard_limit(total_turns)
if max_discardable <= 0:
return None
requested = min(self.config.discard_turns, max_discardable)
return self.truncator.truncate_by_dropping_oldest_turns(
messages,
drop_turns=requested,
)And update _select_disposal to thread total_turns into _try_discard:
async def _select_disposal(
self,
messages: list[Message],
total_turns: int,
) -> list[Message]:
"""Apply disposal strategy: custom > summary > discard."""
if self._unity_compressor is not None:
return await self._unity_compressor(messages)
compressed = await self._try_summary(messages)
if compressed is not None:
return compressed
discarded = self._try_discard(messages, total_turns=total_turns)
if discarded is not None:
return discarded
logger.warning(
"Context disposal triggered but both summary and discard "
"are unavailable or disabled. No compression applied.",
)
return messagesThis keeps the orthogonal trigger/disposal design intact but makes the flow more declarative:
processbecomes the single place where tokens and turns are computed.- Triggers and disposals now operate on explicit inputs (tokens, turns) rather than re-deriving them, which reduces both complexity and runtime overhead.
| return False | ||
| return True | ||
|
|
||
| def _build_context_config(self, settings: dict, umo: str): |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider extracting the new history conversion and context wiring logic into shared utilities so this command handler stays a thin orchestrator over the context engine.
You can reduce the new complexity by pushing the low‑level “context engine” wiring into a small reusable utility module and keeping this command handler focused on orchestration.
1. Extract history ↔ Message conversion
_history_to_messages and _messages_to_history are generic and don’t need access to self. Moving them to a shared utility makes them reusable and reduces the cognitive load in this class.
Before (methods on the command class):
def _history_to_messages(self, history: list[dict]):
from astrbot.core.agent.message import Message
return [
Message(
role=item.get("role", "user"),
content=item.get("content", ""),
tool_calls=item.get("tool_calls"),
tool_call_id=item.get("tool_call_id"),
)
for item in history
]
def _messages_to_history(self, messages):
from astrbot.core.agent.message import ToolCall
result: list[dict] = []
for msg in messages:
entry = {"role": msg.role}
if msg.content is None:
entry["content"] = None
elif isinstance(msg.content, (str, list, dict)):
entry["content"] = msg.content
else:
entry["content"] = str(msg.content)
if msg.tool_calls is not None:
entry["tool_calls"] = [
tc.model_dump() if isinstance(tc, ToolCall) else tc
for tc in msg.tool_calls
]
if msg.tool_call_id is not None:
entry["tool_call_id"] = msg.tool_call_id
result.append(entry)
return resultAfter (shared util):
# astrbot/core/agent/context/history_utils.py
from typing import List, Dict, Any
from astrbot.core.agent.message import Message, ToolCall
def history_to_messages(history: List[Dict[str, Any]]) -> List[Message]:
return [
Message(
role=item.get("role", "user"),
content=item.get("content", ""),
tool_calls=item.get("tool_calls"),
tool_call_id=item.get("tool_call_id"),
)
for item in history
]
def messages_to_history(messages: List[Message]) -> List[Dict[str, Any]]:
result: List[Dict[str, Any]] = []
for msg in messages:
entry: Dict[str, Any] = {"role": msg.role}
if msg.content is None:
entry["content"] = None
elif isinstance(msg.content, (str, list, dict)):
entry["content"] = msg.content
else:
entry["content"] = str(msg.content)
if msg.tool_calls is not None:
entry["tool_calls"] = [
tc.model_dump() if isinstance(tc, ToolCall) else tc
for tc in msg.tool_calls
]
if msg.tool_call_id is not None:
entry["tool_call_id"] = msg.tool_call_id
result.append(entry)
return resultUsage in compact becomes simpler:
from astrbot.core.agent.context.history_utils import (
history_to_messages,
messages_to_history,
)
# ...
messages = history_to_messages(history)
compressed = await cm.process(messages, max_context_tokens=max_context_tokens)
result = messages_to_history(compressed)2. Extract ContextConfig / ContextManager wiring
_build_context_config + ContextManager construction + token‑window extraction is effectively context-engine wiring. Extracting that into a helper keeps compact from being a mini engine.
Before (inside command class):
def _build_context_config(self, settings: dict, umo: str):
from astrbot.core.agent.context.config import ContextConfig
summary_provider_id = settings.get("summary_provider_id", "")
summary_provider = (
self.context.get_provider_by_id(summary_provider_id)
if summary_provider_id
else None
)
if not summary_provider:
summary_provider = self.context.get_using_provider(umo=umo)
return ContextConfig(
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider=summary_provider,
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
)After (utility function):
# astrbot/core/agent/context/context_utils.py
from astrbot.core.agent.context.config import ContextConfig
def build_context_config(context, settings: dict, umo: str) -> ContextConfig:
summary_provider_id = settings.get("summary_provider_id", "")
summary_provider = (
context.get_provider_by_id(summary_provider_id)
if summary_provider_id
else None
)
if not summary_provider:
summary_provider = context.get_using_provider(umo=umo)
return ContextConfig(
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider=summary_provider,
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
)Then in compact:
from astrbot.core.agent.context.manager import ContextManager
from astrbot.core.agent.context.context_utils import build_context_config
# ...
config = build_context_config(self.context, settings, umo)
cm = ContextManager(config)
provider = self.context.get_using_provider(umo=umo)
max_context_tokens = (
provider.provider_config.get("max_context_tokens", 0)
if provider and getattr(provider, "provider_config", None)
else 0
)
compressed = await cm.process(messages, max_context_tokens=max_context_tokens)3. Keep _check_command_permission focused
You’ve already improved reuse between reset and compact; the remaining coupling is to message.set_result. If you want further decoupling without changing behavior, you can return an error message instead of mutating message in the helper:
async def _check_command_permission(... ) -> tuple[bool, str | None]:
# ... compute required_perm ...
if required_perm == "admin" and message.role != "admin":
return False, (
f"{command.capitalize()} command requires admin permission in {scene.name} scenario, "
f"you (ID {message.get_sender_id()}) are not admin, cannot perform this action."
)
return True, NoneUsage:
ok, err = await self._check_command_permission("compact", message, is_group, is_unique_session)
if not ok:
message.set_result(MessageEventResult().message(err))
returnThis keeps permission logic reusable while making the side effect (how the error is surfaced) explicit in the command handler.
All of these changes preserve behavior while making the command file less of a context‑management orchestrator and more of a thin coordinator over shared utilities.
| if self.dequeue_context_length <= 0: | ||
| self.dequeue_context_length = 1 | ||
|
|
||
| # New orthogonal fields |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider encapsulating the new context-related parameters into a dedicated config object to avoid a growing list of separate attributes and constructor arguments.
You can tame the parameter/attribute explosion by grouping the new orthogonal context fields into a dedicated config object, built once from settings and then passed through. This keeps initialize lean and reduces the chance of desync between attributes and constructor args.
1. Introduce a ContextConfig (or similar) object
For example:
from dataclasses import dataclass
@dataclass
class ContextConfig:
max_context_length: int
dequeue_context_length: int
enable_turn_limit: bool
max_turns: int
enable_token_guard: bool
token_guard_threshold: float
enable_summary: bool
enable_discard: bool
discard_turns: int
summary_prompt: str
summary_provider_id: str
retention_method: str
retain_turns: int
retain_percentage: float
fallback_max_context_tokens: int2. Build this object in initialize instead of many instance vars
max_context_length = settings.get("max_context_length", -1)
dequeue_context_length = min(
max(1, settings.get("dequeue_context_length", 1)),
max(1, max_context_length - 1),
)
context_config = ContextConfig(
max_context_length=max_context_length,
dequeue_context_length=dequeue_context_length,
enable_turn_limit=settings.get("enable_turn_limit", False),
max_turns=settings.get("max_turns", 50),
enable_token_guard=settings.get("enable_token_guard", True),
token_guard_threshold=settings.get("token_guard_threshold", 0.82),
enable_summary=settings.get("enable_summary", True),
enable_discard=settings.get("enable_discard", True),
discard_turns=settings.get("discard_turns", 1),
summary_prompt=settings.get("summary_prompt", ""),
summary_provider_id=settings.get("summary_provider_id", ""),
retention_method=settings.get("retention_method", "turns"),
retain_turns=settings.get("retain_turns", 20),
retain_percentage=settings.get("retain_percentage", 0.3),
fallback_max_context_tokens=settings.get("fallback_max_context_tokens", 128000),
)
self.context_config = context_config # if needed elsewhereYou still preserve all existing behavior, but initialize now has a single place where context-related settings are defined.
3. Pass the config object instead of many independent args
If you can adjust MainAgentBuildConfig, prefer:
agent = MainAgentBuildConfig(
# ... existing args ...
context_config=context_config,
llm_safety_mode=self.llm_safety_mode,
safety_mode_strategy=self.safety_mode_strategy,
# ...
)If you must keep the flat signature for now, you can still reduce duplication by unpacking:
agent = MainAgentBuildConfig(
# ... existing args ...
**asdict(context_config),
llm_safety_mode=self.llm_safety_mode,
safety_mode_strategy=self.safety_mode_strategy,
# ...
)This isolates the “context management” surface area into one well-typed structure, makes future additions/changes localized, and prevents initialize from turning into a long list of orthogonal attributes and mirrored constructor parameters.
Rail1bc
marked this pull request as ready for review
July 22, 2026 02:46
Contributor
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
Instead of having _triggers_fired and _try_discard each call _count_turns internally, compute total_turns once in process() and thread it through _select_disposal to both helpers. Eliminates redundant split_into_rounds calls. Also fix(metadata): update CONFIG_METADATA_3 context_management to use new orthogonal trigger/disposal field names, matching the i18n locale files. Co-Authored-By: deepseek-v4-flash <deepseek-ai@claude-code-best.win>
dosubot
Bot
added
size:XXL
Contributor
Author
|
@sourcery-ai review |
Contributor
|
Sorry @Rail1bc, your pull request is larger than the review limit of 150000 diff characters |
Contributor
Author
|
重新整理历史 |
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 / 动机
重构上下文管理配置模型，使其更加正交、可组合，并支持主动触发压缩。解决 #9252、#8348、#9281 等 issue 的实际需求。
核心改动：
summary_provider字段_migra_context_config()+_validate_context_config()接入migra()，覆盖全部旧到新的字段映射，自动迁移，向后兼容context_management结构，共 13 个新字段翻译/compact命令：新增手动触发压缩，读取对话历史 → 调用ContextManager.process()→ 写回 DB。第三方 runner 跳过，异常保留上下文，群聊非隔离默认仅管理员/compact回归测试。108 passed, 0 failedModifications / 改动点
重构图（ContextConfig 正交化）：
astrbot/core/agent/context/config.py— 新增ContextConfig12 个正交字段，废弃旧字段astrbot/core/agent/context/manager.py— 重写process()，统一 trigger → disposal → retention → double-check 流程astrbot/core/agent/context/__init__.py— 导出更新astrbot/core/agent/runners/tool_loop_agent_runner.py— 使用新ContextManager替换旧压缩逻辑astrbot/core/main_agent_build_config.py—MainAgentBuildConfig新增正交字段astrbot/core/astr_main_agent.py— 构建新ContextConfig，_get_compress_provider()解析 summary providerastrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py—initialize()读取 12 个新字段，构建MainAgentBuildConfigastrbot/core/config/default.py— provider_settings 默认值替换astrbot/core/config/migration.py—_migra_context_config()旧→新字段映射迁移,_validate_context_config()校验国际化：
astrbot/dashboard/api/i18n/zh-CN.json— 更新 context_management 翻译（13 字段）astrbot/dashboard/api/i18n/en-US.json— 更新astrbot/dashboard/api/i18n/ru-RU.json— 更新主动触发（
/compact命令）：astrbot/builtin_stars/builtin_commands/commands/conversation.py— 新增compact()方法astrbot/builtin_stars/builtin_commands/main.py— 注册@filter.command("compact")[] This is NOT a breaking change. / 这不是一个破坏性变更。
Screenshots or Test Results / 运行截图或测试结果
测试结果：108 passed, 0 failed, 9 skipped（旧 API 跳过）
tests/agent/test_context_config_new.pytests/agent/test_context_manager_new.pytests/unit/test_context_migration.pytests/unit/test_main_agent_build_config_new.pytests/agent/test_context_manager.py覆盖路径：
Checklist / 检查清单
😊 If there are new features added in the PR, I have discussed it with the authors through issues/emails, etc.
/ 如果 PR 中有新加入的功能，已经通过 Issue / 邮件等方式和作者讨论过。
[] 👀 My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
/ 我的更改经过了良好的测试，并已在上方提供了"验证步骤"和"运行截图"。
🤓 I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations in
requirements.txtandpyproject.toml./ 我确保没有引入新依赖库，或者引入了新依赖库的同时将其添加到
requirements.txt和pyproject.toml文件相应位置。😮 My changes do not introduce malicious code.
/ 我的更改没有引入恶意代码。
Summary by Sourcery
Redesign context management into an orthogonal trigger/disposal model and add a manually-invoked /compact command for conversation compression.
New Features:
Bug Fixes:
Enhancements:
Tests: