`gj> : d Z ddlmZ ddlmZ dgZ G d d Zy)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 @ e Zd ZU dZdZded< ed eD Zded< ed eD Zded< e d eez D Z
d
ed < dd Z dd
Z ddZ
ddZe dd ZddZ d dZ d dZe d d Zedd Zy) 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 | d yw)<>N .0names G/root/.hermes/venv/lib/python3.12/site-packages/agent/think_scrubber.py z StreamingThinkScrubber.Y s 'P!D6 'P
_OPEN_TAGSc # ( K | ]
}d | d yw)Z s (R$2dV1 (Rr _CLOSE_TAGSc # 2 K | ] }t | y w)N)len)r tags r r z StreamingThinkScrubber.] s ICHIs int _MAX_TAG_LENc . d| _ d| _ d| _ y )NF T _in_block_buf_last_emitted_ended_newlineselfs r __init__z StreamingThinkScrubber.__init___ s $ 15( c . d| _ d| _ d| _ y)z4Reset all state. Call at the top of every new turn.Fr" TNr# r' s r resetz StreamingThinkScrubber.resetd s +/(r* c p |sy| j |z }d| _ g }|r| j rq| j || j \ }}|dk( r<| j || j }|r|| d nd| _ dj | S |||z d }d| _ n| j
| }| j || \ }} |Z|dk( s|d |k rM|\ }
} |d|
} | r:| j | } | r'|j | | j d | _ || d }|dk7 rR|d| } | r:| j | } | r'|j | | j d | _ d| _ ||| z d }Y| j || j }| j || j }
t ||
}|r|d| }|| d | _ n |}d| _ |r:| j | }|r'|j | |j d | _ dj | S |rdj | 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( textbufout close_idx close_lenheldpairopen_idxopen_len start_idxend_idx preceding
held_close emit_texts r feedzStreamingThinkScrubber.feedj s ii$ ~~'+';';))($ 9 ? 33C9I9IJD/3TEF DI 773<')i/01!& 66s; &*%@%@&"(
# Nd1g&9)-&Iw #JYI $($A$A)$L $ JJy 1 ) 2 24 8 ! < gh-C r> !$IXI $($A$A)$L $ JJy 1 ) 2 24 8 ! < &*DN h 1 23C
//T__E ! 5 5))
4, #FdU I #TEF DI #I "DI $ = =i HI
9- % . .t 4 8 wws|#c f wws|r* c | j rd| _ d| _ d| _ y| j }d| _ d| _ |sy| j | 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& r5 )r( tails r flushz StreamingThinkScrubber.flush sU >>DI "DN/3D ,yy ,0(,,T22r* c | j }d}d}|D ]? }|j |j }|dk7 s(|dk( s||k s3|}t | }A ||fS )zfReturn (earliest_index, tag_length) over *tags*, or (-1, 0).
Case-insensitive match.
r. r )lowerfindr )r: tags buf_lowerbest_idxbest_lenr idxs r r0 z&StreamingThinkScrubber._find_first_tag sj IIK $C..-Cbyh"nh s8 $
!!r* c j |j }d}t | j | j D ]} \ }}|j }|j }|j | }|dk( r=|j ||t | z } | dk( ra| t | z }
| ||d k sz||
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.
Nr. r )rL zipr r rM r ) r( r: rO bestopen_tag close_tag
open_lower close_lowerr@ r< rC s r r3 z1StreamingThinkScrubber._find_earliest_closed_pair s IIK )-#&t8H8H#I
+ Hi! )J#//+K ~~j1H2~!XJ 7I B #k"22G|x$q' 1 '*
+ r* c
|j }d}d}| j D ]] }|j }d} |j || } | dk( r.| j || | r|dk( s| |k r
| }t | }Y| dz }I ||fS )zReturn the earliest block-boundary open-tag (idx, len).
Returns (-1, 0) if no boundary-legal opener is present.
r. r )rL r rM _is_block_boundaryr )
r( r: already_emittedrO rP rQ r tag_lower search_startrR s
r r4 z-StreamingThinkScrubber._find_open_at_boundary s IIK ?? 'C I L nnY ="9**3_E 2~x #&#&s8 "Qw ' !!r* c , |dk( r"|r|d j d S | j S |d| }|j d }|dk( r:|r|d j d }n | j }|xr |j dk( S ||dz d j 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 r. r/ Nr" r[ )r7 r& rfindstrip)r( r: rR r] rD last_nl
prior_newlines r r\ z)StreamingThinkScrubber._is_block_boundary4 s !8 &r*33D9933 3I //$' b= / 3 < A%)*>*>v*F H
r* c d|vr|S |j }g }d}|t | k rd}|||dz dk( rr| j D ]c }|j }t | }||||z |k( s*||z } | t | k r"|| dv r| dz
} | t | k r|| dv r| }d} n |s|j || |dz
}|t | k rdj | 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" )rL r r r6 r2 )
ri r9
text_lowerr; rk matchedr r^ tag_lenjs
r r5 z/StreamingThinkScrubber._strip_orphan_close_tagsm s
t KZZ\
#d)mG!AE"d* ?? C # I !)nG!!AK0I= K #d)mQ90D FA #d)mQ90D "&
47#Q# #d)m$ wws|r* N)returnNone)r9 strrs ru )rs ru )r: ru rN r rs Tuple[int, int])r: ru )r: ru r] list[str]rs rv )r: ru rR r r] rw rs bool)r: ru rN r rs r )__name__
__module__ __qualname____doc__r
__annotations__tupler r r8 r r) r, rG rJ staticmethodr0 r3 r4 r\ classmethodr1 r5 r r* r r r @ s( (O_ #('P'P"PJP#((R/(R#RK R I
[0HIIL#I6
0 ` D 3B "
"'" " ""8"")2" "2!5!5 !53<!5
!5F , * r* N)r|
__future__r typingr __all__r r r* r r s' 6p # #
$L L r*