Skip to content

fix: segmented reply not applied to send_message_to_user messages #9111

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
tangtaizong666 wants to merge 2 commits into from
Closed

fix: segmented reply not applied to send_message_to_user messages #9111

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b0d92c1
fix: apply segmented reply to send_message_to_user direct sends
tangtaizong666 Jul 2, 2026
ea6cf5c
fix: harden segmented reply config handling
tangtaizong666 Jul 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 13 additions & 28 deletions astrbot/core/pipeline/respond/stage.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,4 @@
import asyncio
import math
import random
from collections.abc import AsyncGenerator

import astrbot.core.message.components as Comp
Expand All @@ -10,6 +8,11 @@
from astrbot.core.platform.astr_message_event import AstrMessageEvent
from astrbot.core.star.star_handler import EventType
from astrbot.core.utils.path_util import path_Mapping
from astrbot.core.utils.segmented_reply import (
SEGMENTED_REPLY_UNSUPPORTED_PLATFORMS,
calc_segment_interval,
parse_interval_range,
)

from ..context import PipelineContext, call_event_hook
from ..stage import Stage, register_stage
Expand Down Expand Up @@ -80,31 +83,17 @@ async def initialize(self, ctx: PipelineContext) -> None:
interval_str: str = ctx.astrbot_config["platform_settings"][
"segmented_reply"
]["interval"]
interval_str_ls = interval_str.replace(" ", "").split(",")
try:
self.interval = [float(t) for t in interval_str_ls]
except BaseException as e:
logger.error(f"解析分段回复的间隔时间失败。{e}")
self.interval = list(parse_interval_range(interval_str))
logger.info(f"分段回复间隔时间：{self.interval}")

async def _word_cnt(self, text: str) -> int:
"""分段回复 统计字数"""
if all(ord(c) < 128 for c in text):
word_count = len(text.split())
else:
word_count = len([c for c in text if c.isalnum()])
return word_count

async def _calc_comp_interval(self, comp: BaseMessageComponent) -> float:
"""分段回复 计算间隔时间"""
if self.interval_method == "log":
if isinstance(comp, Comp.Plain):
wc = await self._word_cnt(comp.text)
i = math.log(wc + 1, self.log_base)
return random.uniform(i, i + 0.5)
return random.uniform(1, 1.75)
# random
return random.uniform(self.interval[0], self.interval[1])
return calc_segment_interval(
comp.text if isinstance(comp, Comp.Plain) else None,
self.interval_method,
(self.interval[0], self.interval[1]),
self.log_base,
)

async def _is_empty_message_chain(self, chain: list[BaseMessageComponent]) -> bool:
"""检查消息链是否为空
Expand Down Expand Up @@ -137,11 +126,7 @@ def is_seg_reply_required(self, event: AstrMessageEvent) -> bool:
if self.only_llm_result and not result.is_model_result():
return False

if event.get_platform_name() in [
"qq_official_webhook",
"weixin_official_account",
"dingtalk",
]:
if event.get_platform_name() in SEGMENTED_REPLY_UNSUPPORTED_PLATFORMS:
return False

return True
Expand Down
86 changes: 27 additions & 59 deletions astrbot/core/pipeline/result_decorate/stage.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
import random
import re
import time
import traceback
from collections.abc import AsyncGenerator
Expand All @@ -13,6 +12,13 @@
from astrbot.core.star.session_llm_manager import SessionServiceManager
from astrbot.core.star.star import star_map
from astrbot.core.star.star_handler import EventType, star_handlers_registry
from astrbot.core.utils.segmented_reply import (
SEGMENTED_REPLY_UNSUPPORTED_PLATFORMS,
cleanup_segments,
compile_split_words_pattern,
split_text_by_regex,
split_text_by_words,
)

from ..context import PipelineContext
from ..stage import Stage, register_stage, registered_stages
Expand Down Expand Up @@ -74,15 +80,7 @@ async def initialize(self, ctx: PipelineContext) -> None:
self.split_words = ctx.astrbot_config["platform_settings"][
"segmented_reply"
].get("split_words", ["。", "？", "！", "~", "…"])
if self.split_words:
escaped_words = sorted(
[re.escape(word) for word in self.split_words], key=len, reverse=True
)
self.split_words_pattern = re.compile(
f"(.*?({'|'.join(escaped_words)})|.+$)", re.DOTALL
)
else:
self.split_words_pattern = None
self.split_words_pattern = compile_split_words_pattern(self.split_words)
self.content_cleanup_rule = ctx.astrbot_config["platform_settings"][
"segmented_reply"
]["content_cleanup_rule"]
Expand All @@ -101,28 +99,6 @@ async def initialize(self, ctx: PipelineContext) -> None:
provider_cfg = ctx.astrbot_config.get("provider_settings", {})
self.show_reasoning = provider_cfg.get("display_reasoning_text", False)

def _split_text_by_words(self, text: str) -> list[str]:
"""使用分段词列表分段文本"""
if not self.split_words_pattern:
return [text]

segments = self.split_words_pattern.findall(text)
result = []
for seg in segments:
if isinstance(seg, tuple):
content = seg[0]
if not isinstance(content, str):
continue
for word in self.split_words:
if content.endswith(word):
content = content[: -len(word)]
break
if content.strip():
result.append(content)
elif seg and seg.strip():
result.append(seg)
return result if result else [text]

async def process(
self,
event: AstrMessageEvent,
Expand Down Expand Up @@ -202,11 +178,11 @@ async def process(
break

# 分段回复
if self.enable_segmented_reply and event.get_platform_name() not in [
"qq_official_webhook",
"weixin_official_account",
"dingtalk",
]:
if (
self.enable_segmented_reply
and event.get_platform_name()
not in SEGMENTED_REPLY_UNSUPPORTED_PLATFORMS
):
if (
self.only_llm_result and result.is_model_result()
) or not self.only_llm_result:
Expand All @@ -220,33 +196,25 @@ async def process(

# 根据 split_mode 选择分段方式
if self.split_mode == "words":
split_response = self._split_text_by_words(comp.text)
split_response = split_text_by_words(
comp.text,
self.split_words,
self.split_words_pattern,
)
else: # regex 模式
try:
split_response = re.findall(
self.regex,
comp.text,
re.DOTALL | re.MULTILINE,
)
except re.error:
logger.error(
f"分段回复正则表达式错误，使用默认分段方式: {traceback.format_exc()}",
)
split_response = re.findall(
r".*?[。？！~…]+|.+$",
comp.text,
re.DOTALL | re.MULTILINE,
)
split_response = split_text_by_regex(
comp.text,
self.regex,
)

if not split_response:
new_chain.append(comp)
continue
for seg in split_response:
if self.content_cleanup_rule:
seg = re.sub(self.content_cleanup_rule, "", seg)
seg = seg.strip()
if seg:
new_chain.append(Plain(seg))
for seg in cleanup_segments(
split_response,
self.content_cleanup_rule,
):
new_chain.append(Plain(seg))
else:
# 非 Plain 类型的消息段不分段
new_chain.append(comp)
Expand Down
166 changes: 165 additions & 1 deletion astrbot/core/tools/message_tools.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import asyncio
import json
import os
import shlex
Expand Down Expand Up @@ -26,6 +27,17 @@
get_astrbot_system_tmp_path,
get_astrbot_temp_path,
)
from astrbot.core.utils.segmented_reply import (
DEFAULT_SPLIT_REGEX,
DEFAULT_SPLIT_WORDS,
SEGMENTED_REPLY_UNSUPPORTED_PLATFORMS,
calc_segment_interval,
cleanup_segments,
compile_split_words_pattern,
parse_interval_range,
split_text_by_regex,
split_text_by_words,
)


def _file_send_allowed_roots(umo: str | None) -> tuple[Path, ...]:
Expand Down Expand Up @@ -187,6 +199,146 @@ async def _resolve_path_from_sandbox(

raise FileNotFoundError(f"{component_type} path does not exist: {path}")

def _get_segmented_reply_conf(

@sourcery-ai sourcery-ai Bot Jul 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (complexity): Consider extracting the platform support checks, chain-building logic, and async send loop into shared segmented_reply utilities so this tool remains thin and mirrors the pipeline’s behavior.

You can keep the new behavior but move most of the orchestration into segmented_reply utilities so this tool stays thin and aligned with the pipeline.

1. Centralize platform support logic

Right now _get_segmented_reply_conf knows too much about platform details. Move that logic into segmented_reply.py:

# astrbot/core/utils/segmented_reply.py

SEGMENTED_REPLY_UNSUPPORTED_PLATFORMS = {...}

def is_segmented_reply_supported(
    platform_id: str,
    platform_manager: Any | None,
) -> bool:
    if platform_id in SEGMENTED_REPLY_UNSUPPORTED_PLATFORMS:
        return False

    if not platform_manager:
        return True

    try:
        for inst in platform_manager.platform_insts:
            meta = inst.meta()
            if (
                meta.id == platform_id
                and meta.name in SEGMENTED_REPLY_UNSUPPORTED_PLATFORMS
            ):
                return False
    except Exception:
        # be conservative, don’t break sending
        return True

    return True

Then _get_segmented_reply_conf becomes flatter and only does config plumbing:

# in the tool

from astrbot.core.utils.segmented_reply import is_segmented_reply_supported

def _get_segmented_reply_conf(
    self,
    context: ContextWrapper[AstrAgentContext],
    target_session: MessageSession,
) -> dict | None:
    try:
        cfg = context.context.context.get_config(umo=str(target_session))
    except Exception:
        return None
    if not isinstance(cfg, dict):
        return None

    seg_conf = cfg.get("platform_settings", {}).get("segmented_reply", {})
    if not seg_conf.get("enable", False):
        return None

    platform_manager = getattr(context.context.context, "platform_manager", None)
    if not is_segmented_reply_supported(target_session.platform_id, platform_manager):
        return None

    return seg_conf

2. Extract generic chain-building into a shared utility

_build_segmented_chains is generic (“components → chains”) and mostly mirrors pipeline behavior. Move it into segmented_reply.py and keep it pure:

# astrbot/core/utils/segmented_reply.py

import astrbot.core.message.components as Comp

def build_segmented_chains(
    components: list[Comp.BaseMessageComponent],
    seg_conf: dict,
) -> list[list[Comp.BaseMessageComponent]] | None:
    try:
        words_count_threshold = int(seg_conf.get("words_count_threshold", 150))
    except (TypeError, ValueError):
        words_count_threshold = 150

    split_mode = seg_conf.get("split_mode", "regex")
    regex = seg_conf.get("regex", DEFAULT_SPLIT_REGEX)
    split_words = seg_conf.get("split_words", DEFAULT_SPLIT_WORDS)
    content_cleanup_rule = seg_conf.get("content_cleanup_rule", "")

    split_words_pattern = (
        compile_split_words_pattern(split_words) if split_mode == "words" else None
    )

    header_comps = [
        comp for comp in components if isinstance(comp, (Comp.At, Comp.Reply))
    ]
    body_comps = [
        comp for comp in components if not isinstance(comp, (Comp.At, Comp.Reply))
    ]
    if not body_comps:
        return None

    units: list[Comp.BaseMessageComponent] = []
    for comp in body_comps:
        if not isinstance(comp, Comp.Plain) or len(comp.text) > words_count_threshold:
            units.append(comp)
            continue

        if split_mode == "words":
            split_response = split_text_by_words(
                comp.text,
                split_words,
                split_words_pattern,
            )
        else:
            split_response = split_text_by_regex(comp.text, regex)

        if not split_response:
            units.append(comp)
            continue

        segments = cleanup_segments(split_response, content_cleanup_rule)
        if segments:
            units.extend(Comp.Plain(text=seg) for seg in segments)
        else:
            units.append(comp)

    chains: list[list[Comp.BaseMessageComponent]] = []
    pending_header = list(header_comps)
    for unit in units:
        if isinstance(unit, Comp.Record):
            chains.append([unit])
        else:
            chains.append([*pending_header, unit])
            pending_header = []

    if pending_header:
        chains.insert(0, pending_header)

    return chains

Then the tool doesn’t need _build_segmented_chains at all:

from astrbot.core.utils.segmented_reply import build_segmented_chains

seg_conf = self._get_segmented_reply_conf(context, target_session)
seg_chains = build_segmented_chains(components, seg_conf) if seg_conf else None

3. Generalize the async sending loop

The sleep/scheduling logic in _send_segmented_chains can also be a shared helper, since it already uses shared primitives (parse_interval_range, calc_segment_interval):

# astrbot/core/utils/segmented_reply.py

from astrbot.core.message import MessageChain

async def send_segmented_chains(
    context: Any,
    target_session: Any,
    chains: list[list[Comp.BaseMessageComponent]],
    seg_conf: dict,
) -> None:
    interval_method = seg_conf.get("interval_method", "random")
    interval_range = parse_interval_range(str(seg_conf.get("interval", "1.5,3.5")))
    try:
        log_base = float(seg_conf.get("log_base", 2.6))
    except (TypeError, ValueError):
        log_base = 2.6

    for chain in chains:
        main_comp = chain[-1]
        interval = calc_segment_interval(
            main_comp.text if isinstance(main_comp, Comp.Plain) else None,
            interval_method,
            interval_range,
            log_base,
        )
        await asyncio.sleep(interval)
        await context.send_message(target_session, MessageChain(chain=chain))

And in the tool you only adapt the context type:

from astrbot.core.utils.segmented_reply import (
    build_segmented_chains,
    send_segmented_chains,
)

seg_conf = self._get_segmented_reply_conf(context, target_session)
seg_chains = build_segmented_chains(components, seg_conf) if seg_conf else None

if seg_chains:
    await send_segmented_chains(
        context.context.context,  # existing ContextImpl
        target_session,
        seg_chains,
        seg_conf,
    )
else:
    await context.context.context.send_message(target_session, message_chain)

This keeps all current behavior (including header/body handling, Record special casing, and interval logic), but:

  • removes three tightly-coupled private helpers from the tool,
  • pushes orchestration and platform rules into segmented_reply.py,
  • leaves the tool’s call() flow close to “get config → maybe segment → send”.

self,
context: ContextWrapper[AstrAgentContext],
target_session: MessageSession,
) -> dict | None:
"""获取目标会话生效的分段回复配置；不需要分段时返回 None。

send_message_to_user 直接调用 send_message 发送，绕过了
ResultDecorateStage/RespondStage，因此需要在这里应用与 pipeline
一致的分段回复行为。See #8325.
"""
try:
cfg = context.context.context.get_config(umo=str(target_session))
except Exception:
return None
if not isinstance(cfg, dict):
return None
seg_conf = cfg.get("platform_settings", {}).get("segmented_reply", {})
# 工具消息由 LLM 主动发出，属于 LLM 结果，无需检查 only_llm_result。
if not seg_conf.get("enable", False):
return None
platform_id = target_session.platform_id
if platform_id in SEGMENTED_REPLY_UNSUPPORTED_PLATFORMS:
return None
platform_manager = getattr(
context.context.context,
"platform_manager",
None,
)
if platform_manager is not None:
try:
for inst in platform_manager.platform_insts:
meta = inst.meta()
if (
meta.id == platform_id
and meta.name in SEGMENTED_REPLY_UNSUPPORTED_PLATFORMS
):
return None
except Exception:
pass
return seg_conf

def _build_segmented_chains(
self,
components: list[Comp.BaseMessageComponent],
seg_conf: dict,
) -> list[list[Comp.BaseMessageComponent]] | None:
"""按分段回复配置将消息组件拆分为多条消息链；返回 None 表示原样发送。

与 pipeline 行为对齐：Plain 文本按配置分段，At/Reply 附加在第一条
消息上（Record 需要单独发送，不附加）。
"""
try:
words_count_threshold = int(seg_conf.get("words_count_threshold", 150))
except (TypeError, ValueError):
words_count_threshold = 150
split_mode = seg_conf.get("split_mode", "regex")
regex = seg_conf.get("regex", DEFAULT_SPLIT_REGEX)
split_words = seg_conf.get("split_words", DEFAULT_SPLIT_WORDS)
content_cleanup_rule = seg_conf.get("content_cleanup_rule", "")
split_words_pattern = (
compile_split_words_pattern(split_words) if split_mode == "words" else None
)

header_comps = [
comp for comp in components if isinstance(comp, (Comp.At, Comp.Reply))
]
body_comps = [
comp for comp in components if not isinstance(comp, (Comp.At, Comp.Reply))
]
if not body_comps:
return None

units: list[Comp.BaseMessageComponent] = []
for comp in body_comps:
if (
not isinstance(comp, Comp.Plain)
or len(comp.text) > words_count_threshold
):
# 与 ResultDecorateStage 一致：非 Plain 或超过阈值的长文本不分段
units.append(comp)
continue
if split_mode == "words":
split_response = split_text_by_words(
comp.text,
split_words,
split_words_pattern,
)
else:
split_response = split_text_by_regex(comp.text, regex)
if not split_response:
units.append(comp)
continue
segments = cleanup_segments(split_response, content_cleanup_rule)
if segments:
units.extend(Comp.Plain(text=seg) for seg in segments)
else:
units.append(comp)

chains: list[list[Comp.BaseMessageComponent]] = []
pending_header = list(header_comps)
for unit in units:
if isinstance(unit, Comp.Record):
# Record 需要单独发送，与 RespondStage 一致
chains.append([unit])
else:
chains.append([*pending_header, unit])
pending_header = []
if pending_header:
# 全部是 Record 时，单独发送 At/Reply，避免丢失 LLM 显式要求的提及
chains.insert(0, pending_header)
return chains

async def _send_segmented_chains(
self,
context: ContextWrapper[AstrAgentContext],
target_session: MessageSession,
chains: list[list[Comp.BaseMessageComponent]],
seg_conf: dict,
) -> None:
interval_method = seg_conf.get("interval_method", "random")
interval_range = parse_interval_range(str(seg_conf.get("interval", "1.5,3.5")))
try:
log_base = float(seg_conf.get("log_base", 2.6))
except (TypeError, ValueError):
log_base = 2.6
for chain in chains:
main_comp = chain[-1]
interval = calc_segment_interval(
main_comp.text if isinstance(main_comp, Comp.Plain) else None,
interval_method,
interval_range,
log_base,
)
await asyncio.sleep(interval)
await context.context.context.send_message(
target_session,
MessageChain(chain=chain),
)

async def call(
self, context: ContextWrapper[AstrAgentContext], **kwargs
) -> ToolExecResult:
Expand Down Expand Up @@ -318,7 +470,19 @@ async def call(
return f"error: invalid session: {session}"

message_chain = MessageChain(chain=components)
await context.context.context.send_message(target_session, message_chain)
seg_conf = self._get_segmented_reply_conf(context, target_session)
seg_chains = (
self._build_segmented_chains(components, seg_conf) if seg_conf else None
)
if seg_chains:
await self._send_segmented_chains(
context,
target_session,
seg_chains,
seg_conf,
)
else:
await context.context.context.send_message(target_session, message_chain)
if str(target_session) == current_session:
context.context.event._has_send_oper = True
sent_plain_text = message_chain.get_plain_text().strip()
Expand Down
Loading