butler/gateway/progressive_stream.py

Rama main · réplica de lectura del repositorio original.

"""Optional progressive WeChat replies during long streams (Hermes GatewayStream subset)."""

from __future__ import annotations

import time
from typing import cast

from butler.utilities.env_parse import env_truthy, int_env, float_env
from butler.defaults.env_defaults import (
    GATEWAY_PROGRESSIVE_STREAM_DEFAULT,
    GATEWAY_PROGRESSIVE_MIN_CHARS,
    GATEWAY_PROGRESSIVE_INTERVAL_SECONDS,
)


def progressive_stream_enabled() -> bool:
    return cast(bool, env_truthy("BUTLER_GATEWAY_PROGRESSIVE_STREAM", default=GATEWAY_PROGRESSIVE_STREAM_DEFAULT))


def progressive_stream_min_chars() -> int:
    try:
        result: int = int_env("BUTLER_GATEWAY_PROGRESSIVE_MIN_CHARS", GATEWAY_PROGRESSIVE_MIN_CHARS, min=80)
        return result
    except ValueError:
        return int(GATEWAY_PROGRESSIVE_MIN_CHARS)


def progressive_stream_interval_seconds() -> float:
    try:
        result: float = float_env("BUTLER_GATEWAY_PROGRESSIVE_INTERVAL", GATEWAY_PROGRESSIVE_INTERVAL_SECONDS, min=15.0)
        return result
    except ValueError:
        return float(GATEWAY_PROGRESSIVE_INTERVAL_SECONDS)


def format_progressive_chunk(preview: str) -> str:
    body = (preview or "").strip()
    if len(body) > 320:
        body = body[:317] + "..."
    if not body:
        return ""
    return f"⏳ 处理中…\n{body}"


def maybe_schedule_progressive_reply(bridge: object | None, preview: str) -> None:
    """Rate-limited supplementary message while the main reply is still generating."""
    if bridge is None or not progressive_stream_enabled():
        return
    text = format_progressive_chunk(preview)
    if not text:
        return
    now = time.monotonic()
    last = float(getattr(bridge, "_progressive_last_at", 0.0) or 0.0)
    if now - last < progressive_stream_interval_seconds():
        return
    chars = int(getattr(bridge, "_stream_chars", 0) or 0)
    if chars < progressive_stream_min_chars():
        return
    schedule = getattr(bridge, "schedule_supplementary_reply", None)
    if not callable(schedule):
        return
    if schedule(text, kind="progressive_stream"):
        bridge._progressive_last_at = now  # type: ignore[attr-defined]