yj> B d Z ddlmZ ddlmZ dgZ G d d ZdS )u Stateful scrubber for reasoning/thinking blocks in streamed assistant text.
``run_agent._strip_think_blocks`` is regex-based and correct for a complete
string, but when it runs *per-delta* in ``_fire_stream_delta`` it destroys
the state that downstream consumers (CLI ``_stream_delta``, gateway
``GatewayStreamConsumer._filter_and_accumulate``) rely on.
Concretely, when MiniMax-M2.7 streams
delta1 = ""
delta2 = "Let me check their config"
delta3 = ""
the per-delta regex erases delta1 entirely (case 2: unterminated-open at
boundary matches ``^...``), so the downstream state machine never
sees the open tag, treats delta2 as regular content, and leaks reasoning
to the user. Consumers that don't run their own state machine (ACP,
api_server, TTS) never had any defence at all — they just emitted
whatever survived the upstream regex.
This module centralises the tag-suppression state machine at the
upstream layer so every stream_delta_callback sees text that has
already had reasoning blocks removed. Partial tags at delta
boundaries are held back until the next delta resolves them, and
end-of-stream flushing surfaces any held-back prose that turned out
not to be a real tag.
Usage::
scrubber = StreamingThinkScrubber()
for delta in stream:
visible = scrubber.feed(delta)
if visible:
emit(visible)
tail = scrubber.flush() # at end of stream
if tail:
emit(tail)
The scrubber is re-entrant per agent instance. Call ``reset()`` at
the top of each new turn so a hung block from an interrupted prior
stream cannot taint the next turn's output.
Tag variants handled (case-insensitive):
````, ````, ````, ````,
````.
Block-boundary rule for opens: an opening tag is only treated as a
reasoning-block opener when it appears at the start of the stream,
after a newline (optionally followed by whitespace), or when only
whitespace has been emitted on the current line. This prevents prose
that *mentions* the tag name (e.g. ``"use tags here"``) from
being incorrectly suppressed. Closed pairs (``X``) are
always suppressed regardless of boundary; a closed pair is an
intentional, bounded construct.
) annotations)TupleStreamingThinkScrubberc D e Zd ZU dZdZded< ed eD Zded< ed eD Zded< e d eez D Z
d
ed < d"dZ d"dZ d#dZ
d$dZed%d Zd&dZd'dZd(d Zed)d Zed#d Zd!S )*r an Stateful scrubber for streaming reasoning/thinking blocks.
State machine:
- ``_in_block``: True while inside an opened block, waiting for
a close tag. All text inside is discarded.
- ``_buf``: held-back partial-tag tail. Emitted / discarded on
the next ``feed()`` call or by ``flush()``.
- ``_last_emitted_ended_newline``: True iff the most recent
emission to the consumer ended with ``\n``, or nothing has
been emitted yet (start-of-stream counts as a boundary). Used
to decide whether an open tag at buffer position 0 is at a
block boundary.
)thinkthinking reasoningthoughtREASONING_SCRATCHPADTuple[str, ...]_OPEN_TAG_NAMESc # " K | ]
}d | dV dS )<>N .0names :/home/ragecks/.hermes/hermes-agent/agent/think_scrubber.py z StreamingThinkScrubber.Y s* 'P'P D 'P'P'P'P'P'P
_OPEN_TAGSc # " K | ]
}d | dV dS )Z s* (R(R$ d (R(R(R(R(R(Rr _CLOSE_TAGSc # 4 K | ]}t | V d S )N)len)r tags r r z StreamingThinkScrubber.] s( IICHHIIIIIIr int _MAX_TAG_LENreturnNonec 0 d| _ d| _ d| _ d S )NF T _in_block_buf_last_emitted_ended_newlineselfs r __init__z StreamingThinkScrubber.__init___ s $ 15(((r c 0 d| _ d| _ d| _ dS )z4Reset all state. Call at the top of every new turn.Fr$ TNr% r) s r resetz StreamingThinkScrubber.resetd s +/(((r textstrc * |sdS | j |z }d| _ g }|re| j r~| || j \ }}|dk rD| || j }|r || d nd| _ d | S |||z d }d| _ n| | }| || \ }} |u|dk s |d |k rc|\ }
} |d|
} | rF| | } | r/| | |
d | _ || d }-|dk rh|d| } | rF| | } | r/| | |
d | _ d| _ ||| z d }| || j }| || j }
t ||
}|r |d| }|| d | _ n |}d| _ |rF| | }|r/| | |
d | _ d | S |ed | S )zFeed one delta; return the scrubbed visible portion.
May return an empty string when the entire delta is reasoning
content or is being held back pending resolution of a partial
tag at the boundary.
r$ NFr
T)r' r& _find_first_tagr _max_partial_suffixjoin_find_earliest_closed_pair_find_open_at_boundary_strip_orphan_close_tagsappendendswithr( r max)r* r. bufout close_idx close_lenheldpairopen_idxopen_len start_idxend_idx preceding
held_close emit_texts r feedzStreamingThinkScrubber.feedj s 2i$ Q $~ P
$'+';';)( ($ 9 ?? 33C9IJJD/3 ;TEFF DI 773<<')i/001!& 66s;; &*%@%@& &"(
# NNd1g&9&9)-&Iw #JYJI $($A$A)$L$L $ JJy 1 1 1 ) 2 24 8 8 ! < ghh-C r>> !$IXII $($A$A)$L$L $ JJy 1 1 1 ) 2 24 8 8 ! < &*DN h 1 2 23C
//T_EE ! 5 5)
4,, # #FdUF I #TEFF DII #I "DI $ = =i H HI
9--- % . .t 4 4 8 wws||#c Q $f wws||r c | j rd| _ d| _ d| _ dS | j }d| _ d| _ |sdS | | S )u End-of-stream flush.
If still inside an unterminated block, held-back content is
discarded — leaking partial reasoning is worse than a
truncated answer. Otherwise the held-back partial-tag tail is
emitted verbatim (it turned out not to be a real tag prefix).
Always treats the next ``feed()`` as a fresh stream boundary.
Intra-turn retries (thinking-only prefill, empty-response
retry) flush then stream again without calling ``reset()``;
leaving ``_last_emitted_ended_newline`` False made a new
stream's opening ```` look mid-line and leak into the
visible reply.
r$ FT)r& r' r( r8 )r* tails r flushz StreamingThinkScrubber.flush sb > DI "DN/3D ,2y ,0( 2,,T222r r< tagsTuple[int, int]c | }d}d}|D ]L}| | }|dk r |dk s||k r|}t | }M||fS )zfReturn (earliest_index, tag_length) over *tags*, or (-1, 0).
Case-insensitive match.
r1 r )lowerfindr )r< rM buf_lowerbest_idxbest_lenr idxs r r3 z&StreamingThinkScrubber._find_first_tag sx IIKK $ $C..--Cbyyh"nnh s88!!r c | }d}t | j | j D ]\ }}| }| }| | }|dk rI| ||t | z } | dk rv| t | z }
| ||d k r||
f}|S )a Return (start_idx, end_idx) of the earliest closed pair, else None.
A closed pair is ``...`` of any variant. Matches are
case-insensitive and non-greedy (the closest close tag after
an open tag wins), matching the regex ``.*?``
semantics of ``_strip_think_blocks`` case 1. When two tag
variants could both match, the one whose open tag appears
earlier wins.
Nr1 r )rP zipr r rQ r ) r* r< rR bestopen_tag close_tag
open_lower close_lowerrB r> rE s r r6 z1StreamingThinkScrubber._find_earliest_closed_pair s IIKK )-#&t8H#I#I
+
+ Hi! ))J#//++K ~~j11H2~~!XJ 7 I B #k"2"22G|x$q' 1 1 '* r already_emitted list[str]c , | }d}d}| j D ]q}| }d} | || } | dk rn;| || | r |dk s| |k r| }t | }n| dz }Xr||fS )zReturn the earliest block-boundary open-tag (idx, len).
Returns (-1, 0) if no boundary-legal opener is present.
r1 r T )rP r rQ _is_block_boundaryr )
r* r< r] rR rS rT r tag_lower search_startrU s
r r7 z-StreamingThinkScrubber._find_open_at_boundary s IIKK ? ' 'C I L
' nnY =="99**3_EE 2~~x #&#&s88 "Qw
' !!r rU boolc d |dk r$|r|d d S | j S |d| }| d }|dk r?|r |d d }n| j }|o| dk S ||dz d dk S )a True iff position *idx* in *buf* is a block boundary.
A block boundary is:
- buf position 0 AND the most recent emission ended with
a newline (or nothing has been emitted yet)
- any position whose preceding text on the current line
(since the last newline in buf) is whitespace-only, AND
if there is no newline in the preceding buf portion, the
most recent prior emission ended with a newline
r r1 r2 Nr$ r` )r: r( rfindstrip)r* r< rU r] rF last_nl
prior_newlines r ra z)StreamingThinkScrubber._is_block_boundary4 s !88
:&r*33D9993 3I //$'' b==
A / 3 < >A%%)*>*>v*F*F% HHHHH
qr c . d|vr|S | }g }d}|t | k rd}|||dz dk r| j D ]}| }t | }||||z |k rJ||z } | t | k r,|| dv r"| dz
} | t | k r
|| dv "| }d} n|s | || |dz
}|t | k d | S ) a Remove any close tags from *text* (orphan-close handling).
An orphan close tag has no matching open in the current
scrubber state; it's always noise, stripped with any trailing
whitespace so the surrounding prose flows naturally.
r r F z
r` Tr$ )rP r r r9 r5 )
rn r.
text_lowerr= rp matchedr rb tag_lenjs
r r8 z/StreamingThinkScrubber._strip_orphan_close_tagsm sH t KZZ\\
#d))mmG!AE'"d** ? C # I !)nnG!!AK-0I== K #d))mmQ90D0D FA #d))mmQ90D0D "& >
47###Q# #d))mm$ wws||r N)r! r" )r. r/ r! r/ )r! r/ )r< r/ rM r r! rN )r< r/ )r< r/ r] r^ r! rN )r< r/ rU r r] r^ r! rd )r< r/ rM r r! r )__name__
__module__ __qualname____doc__r
__annotations__tupler r r; r r+ r- rI rL staticmethodr3 r6 r7 ra classmethodr4 r8 r r r r r @ s (O #(%'P'P'P'P'P"P"PJPPPP#(5(R(R/(R(R(R#R#RKRRRR II
[0HIIIIILIIII6 6 6 6
0 0 0 0 ` ` ` ` D 3 3 3 3B " " " \"" 8" " " "2!5 !5 !5 !5F [* [ r N)r{
__future__r typingr __all__r r r r r sz 6 6p # " " " " " #
$L L L L L L L L L L r