"""
Module de parseurs de flux textuels temps reel pour LLM et agents conversationnels.
Fournit :
- StreamingTextFilter : separation en continu des canaux de pensee (<think>),
des artefacts documentaires (<result>, <code>, <code_view>) et du flux oral.
- FinalAnswerStreamParser : extraction live de l'argument final_answer sous smolagents.
"""
from __future__ import annotations
import re
from collections.abc import Callable
from .logger import getMirzaLogger
logger = getMirzaLogger("mirza.stream")
# Expressions regulieres des canaux de pensee, de document et de brut
RE_DOC_END = re.compile(r"</(?:result|code|code_view)>", re.IGNORECASE)
RE_DOC_START = re.compile(r"<(result|code|code_view)\b[^>]*>", re.IGNORECASE)
RE_XML_STRIP = re.compile(r"<[^>]+>")
RE_THINKING_START = re.compile(
r"<\|channel>thought\b|<(?:thought|thinking|think)\b[^>]*>|(?:^|\n)thought\n|"
r"(?:^|\n)\s*[*•-]\s*(?:Constraint|Persona|Idea|Draft|Need|The previous|Option|Step|Critique|Analysis)\b|"
r"(?:^|\n)\s*(?:Draft\s*\d+|Idea\s*\d+|Option\s*\d+):|"
r"(?:^|\n)\s*Let['\x27]s (?:see|think|analyze|go with)\b",
re.IGNORECASE,
)
RE_THINKING_TAG = re.compile(
r"^(?:<\|channel>thought\b|<(?:thought|thinking|think)\b[^>]*>|(?:^|\n)thought\n)",
re.IGNORECASE,
)
RE_HEADER_TAGS = re.compile(r"<\|(?:turn>model|channel>text)>?\n?|<[^>]+>")
RE_OPEN_THINK_TAG = re.compile(
r"^\s*(?:<\|channel>thought\b|<(?:thought|thinking|think)\b[^>]*>|thought\n?)",
re.IGNORECASE,
)
RE_THINKING_END = re.compile(
r"<channel\|>|</(?:thought|thinking|think)>|<\|channel>text\b",
re.IGNORECASE,
)
#: Marqueur de puce en debut de ligne. Un seul caractere de puce suivi d'un
#: espace : `**` en debut de ligne est du gras, pas une puce.
RE_LIST_MARKER = re.compile(r"^[-*•]\s+|#\s*")
RE_DRAFT_MATCHES = re.compile(
r"(?:Revised|Draft\s*\d+|Sentence\s*1)[^:]*:\*?\s*([^\n]+(?:\n[^\n*#]+)*)",
re.IGNORECASE,
)
RE_CLEAN_PREFIX = re.compile(r"^[`\s*•\-]+")
RE_IMPLICIT_DOC_START = re.compile(
r"(?:\n\s*#{1,3}\s+|\n\s*\*\*(?:Dossier|Patient|Consultation|Bilan|Observation|Traitement|Recommandation|Rapport|Synthèse|Synthese|Historique)\b)",
re.IGNORECASE,
)
[docs]
class StreamingTextFilter:
"""
Filtre en flux continu les balises XML/Markdown et les blocs de pensee.
Isole en temps reel les flux CoT, les documents riches et le texte parlable.
"""
[docs]
def __init__(
self,
onThought: Callable[[str], None] | None = None,
onResultDoc: Callable[[str], None] | None = None,
onDocActive: Callable[[bool], None] | None = None,
maxThoughtTokens: int = 0,
initialThinkingActive: bool = False,
) -> None:
"""
Initialise les tampons et callbacks du filtre de flux textuel.
Parameters
----------
onThought : Callable[[str], None] | None, optional
Fonction appelee a chaque token de pensee CoT detecte.
onResultDoc : Callable[[str], None] | None, optional
Fonction appelee avec le contenu documentaire extrait des balises <result>, <code> ou <code_view>.
onDocActive : Callable[[bool], None] | None, optional
Fonction appelee avec un booleen signalant le debut/fin de generation d'un bloc de resultat/code.
maxThoughtTokens : int, optional
Nombre maximum de tokens de reflexion autorises. Defaut : 0.
initialThinkingActive : bool, optional
True si le LLM demarre directement en mode pensee, sans balise
ouvrante explicite. Defaut : False.
"""
self.inThinkingBlock = initialThinkingActive
self.inDocBlock = False
self.inResultBlock = False
self.isMarkdownFenceDoc = False
self.currentDocTag: str | None = None
self.buffer = ""
self.resultBuffer = ""
self.onThought = onThought
self.onResultDoc = onResultDoc
self.onDocActive = onDocActive
self.maxThoughtTokens = maxThoughtTokens
self.thoughtTokensCount = 0
self._backtickParity = 0
self._insideInlineCode = False
def _emitDocument(self, texte: str) -> None:
"""
Envoie le document au client, une fois normalise.
Ce parseur est le premier a voir la reponse complete, avant l'arbitrage
de `agentOutput`. Sans cette normalisation, deux enveloppes protocolaires
arrivaient jusqu'a l'ecran : celle de l'appel d'outil, que le modele
ecrit encore, et le `<code>` que reactParser ajoute. Le badge affichait
alors le code d'appel au lieu du document, avec le titre "extrait de
code" declenche par les backticks ajoutes ici meme.
Le retrait de l'appel se fait avant celui du `<code>` : l'inverse
laisserait une enveloppe vide autour du document.
Parameters
----------
texte : str
Le document extrait de la balise, encore brut.
"""
from ..pipeline.agentOutput import ACTION_WRAPPER, TOOL_CALL_WRAPPER
if not texte.strip():
return
enveloppe = ACTION_WRAPPER.search(texte)
if enveloppe:
texte = f"{texte[:enveloppe.start()]} {enveloppe.group(1)} {texte[enveloppe.end():]}"
appel = TOOL_CALL_WRAPPER.match(texte.strip())
if appel:
texte = texte.strip()[appel.start("body") : appel.end("body")]
if self.currentDocTag in ("code", "code_view") and not texte.startswith("```"):
texte = f"```\n{texte}\n```"
if self.onResultDoc and texte.strip():
self.onResultDoc(texte.strip())
def _trackInlineCode(self) -> None:
"""
Met a jour la parite des accents graves du fragment courant.
Un accent grave non fermant signifie qu'un code en ligne a ete coupe en
deux par le streaming. Les fences de trois accents n'entrent pas dans le
compte : elles ouvrent et ferment un bloc de code, pas un fragment.
"""
fences = self.buffer.count("```")
singles = self.buffer.replace("```", "").count("`")
self._backtickParity = (self._backtickParity + singles) % 2
self._insideInlineCode = self._backtickParity == 1 and not fences
[docs]
def processChunk(self, chunk: str) -> str:
"""
Traite un fragment de texte en continu et retourne le texte parlable filtre.
Parameters
----------
chunk : str
Fragment ou jeton textuel brut emis par le LLM.
Returns
-------
str
Texte epure des balises de pensee et de resultats, pret pour la voix ou l'affichage.
"""
self.buffer += chunk
output: list[str] = []
while self.buffer:
# 0. Bloc <result>, <code>, <code_view> ou fence markdown en cours
if self.inDocBlock or self.inResultBlock:
if self.isMarkdownFenceDoc:
if self.buffer.startswith("```"):
newlineIdx = self.buffer.find("\n")
if newlineIdx != -1:
self.resultBuffer += self.buffer[:newlineIdx + 1]
self.buffer = self.buffer[newlineIdx + 1:]
else:
break
closeMatch = re.search(r"\n```\s*(?:\n|$)|```$", self.buffer)
if closeMatch:
docPart = self.buffer[:closeMatch.end()]
self.resultBuffer += docPart
self._emitDocument(self.resultBuffer)
if self.onDocActive:
try:
self.onDocActive(False)
except Exception as exc:
logger.debug("Erreur onDocActive(False): %s", exc)
self.inDocBlock = False
self.inResultBlock = False
self.isMarkdownFenceDoc = False
self.currentDocTag = None
self.buffer = self.buffer[closeMatch.end():]
continue
else:
if len(self.buffer) > 15:
safeDoc = self.buffer[:-10]
self.resultBuffer += safeDoc
self.buffer = self.buffer[-10:]
break
else:
endMatch = RE_DOC_END.search(self.buffer)
if endMatch:
docPart = self.buffer[:endMatch.start()]
self.resultBuffer += docPart
self._emitDocument(self.resultBuffer)
if self.onDocActive:
try:
self.onDocActive(False)
except Exception as exc:
logger.debug("Erreur onDocActive(False): %s", exc)
self.inDocBlock = False
self.inResultBlock = False
self.currentDocTag = None
self.buffer = self.buffer[endMatch.end():]
continue
else:
if len(self.buffer) > 20:
safeDoc = self.buffer[:-16]
self.resultBuffer += safeDoc
self.buffer = self.buffer[-16:]
break
if not self.inThinkingBlock:
# 0 bis. Ouverture d'un bloc <result>, <code> ou <code_view>
docMatch = RE_DOC_START.search(self.buffer)
if docMatch:
tagActuel = docMatch.group(1).lower()
# Balise <code> smolagents : transport d'appel, a ignorer pour le document
if tagActuel == "code":
avant = self.buffer[:docMatch.start()]
apres = self.buffer[docMatch.end():]
if avant:
cleanPre = RE_XML_STRIP.sub("", avant)
if cleanPre:
output.append(cleanPre)
self.buffer = apres
continue
pre = self.buffer[:docMatch.start()]
if pre:
cleanPre = RE_XML_STRIP.sub("", pre)
if cleanPre:
output.append(cleanPre)
self.inDocBlock = True
self.inResultBlock = True
self.isMarkdownFenceDoc = False
self.currentDocTag = tagActuel
self.resultBuffer = ""
self.buffer = self.buffer[docMatch.end():]
if self.onDocActive:
try:
self.onDocActive(True)
except Exception as exc:
logger.debug("Erreur onDocActive(True): %s", exc)
continue
# 0 ter. Fence markdown ``` non encadre par une balise
if "```" in self.buffer:
fenceIdx = self.buffer.find("```")
pre = self.buffer[:fenceIdx]
if pre:
cleanPre = RE_XML_STRIP.sub("", pre)
if cleanPre:
output.append(cleanPre)
self.inDocBlock = True
self.inResultBlock = True
self.isMarkdownFenceDoc = True
self.currentDocTag = "code_view"
self.resultBuffer = ""
self.buffer = self.buffer[fenceIdx:]
if self.onDocActive:
try:
self.onDocActive(True)
except Exception as exc:
logger.debug("Erreur onDocActive(True): %s", exc)
continue
# 0 quater. Détection anticipée d'une section documentaire implicite (## Titre ou **Dossier)
# afin d'éviter la lecture orale de comptes-rendus ou dossiers cliniques non balisés
implicitMatch = RE_IMPLICIT_DOC_START.search(self.buffer)
if implicitMatch:
pre = self.buffer[:implicitMatch.start()]
if pre:
cleanPre = RE_XML_STRIP.sub("", pre)
if cleanPre:
output.append(cleanPre)
self.inDocBlock = True
self.inResultBlock = True
self.isMarkdownFenceDoc = False
self.currentDocTag = "result"
self.resultBuffer = ""
self.buffer = self.buffer[implicitMatch.start() + 1:] # sauter le '\n' initial
if self.onDocActive:
try:
self.onDocActive(True)
except Exception as exc:
logger.debug("Erreur onDocActive(True): %s", exc)
continue
# 1. Balise incomplete en fin de buffer
if "<" in self.buffer:
lastLt = self.buffer.rfind("<")
if ">" not in self.buffer[lastLt:] and (len(self.buffer) - lastLt) < 25:
pre = self.buffer[:lastLt]
if pre:
cleanPre = RE_XML_STRIP.sub("", pre)
if cleanPre:
output.append(cleanPre)
self.buffer = self.buffer[lastLt:]
break
# 2. Identifiant de canal incomplet (<|channel>...)
channelIdx = self.buffer.find("<|channel>")
if channelIdx != -1:
afterChannel = self.buffer[channelIdx + len("<|channel>"):]
if not any(sep in afterChannel for sep in ["\n", " ", ":", ">"]) and len(afterChannel) < 10:
pre = self.buffer[:channelIdx]
if pre:
cleanPre = RE_XML_STRIP.sub("", pre)
if cleanPre:
output.append(cleanPre)
self.buffer = self.buffer[channelIdx:]
break
# 3. Detection du debut d'un bloc de pensee
startMatch = RE_THINKING_START.search(self.buffer)
if startMatch:
pre = self.buffer[:startMatch.start()]
if pre:
cleanPre = RE_XML_STRIP.sub("", pre)
if cleanPre:
output.append(cleanPre)
self.inThinkingBlock = True
self.buffer = self.buffer[startMatch.start():]
tagMatch = RE_THINKING_TAG.match(self.buffer)
if tagMatch:
self.buffer = self.buffer[tagMatch.end():]
continue
# 4. Nettoyage des entetes de canal texte ou balises orphelines
headerMatch = RE_HEADER_TAGS.search(self.buffer)
if headerMatch:
pre = self.buffer[:headerMatch.start()]
if pre:
output.append(pre)
self.buffer = self.buffer[headerMatch.end():]
continue
# 5. Garde-fou semantique : suppression des puces de brouillon hors code inline
if not self._insideInlineCode:
bullet = RE_LIST_MARKER.match(self.buffer)
if bullet and not self.buffer.startswith("```"):
self.buffer = self.buffer[bullet.end():]
self._trackInlineCode()
output.append(self.buffer)
self.buffer = ""
else:
openTag = RE_OPEN_THINK_TAG.match(self.buffer)
if openTag:
self.buffer = self.buffer[openTag.end():]
self.thoughtTokensCount += max(1, len(chunk) // 4)
endMatch = None
for m in RE_THINKING_END.finditer(self.buffer):
matchStart = m.start()
preText = self.buffer[max(0, matchStart - 30):matchStart]
postText = self.buffer[m.end():m.end() + 20]
if m.group(0).lower() == "<channel|>":
if postText.startswith(("<|channel>text", "\n<|channel>text")):
endMatch = m
break
if any(q in preText for q in ['"', "'", "«", "avec "]):
continue
endMatch = m
break
if endMatch:
thoughtText = self.buffer[:endMatch.start()]
if thoughtText and self.onThought:
self.onThought(thoughtText)
self.inThinkingBlock = False
remaining = self.buffer[endMatch.end():]
remaining = re.sub(r"^(?:<\|channel>text\n?|<\|channel>|text\n?)", "", remaining, flags=re.IGNORECASE)
remaining = re.sub(r"^[`\.\*\-#]+", "", remaining)
self.buffer = remaining
continue
else:
if len(self.buffer) > 20:
safeEmit = self.buffer[:-18]
self.buffer = self.buffer[-18:]
if self.onThought and safeEmit:
self.onThought(safeEmit)
break
return "".join(output)
[docs]
def finalize(self) -> str:
"""
Vide les residus du buffer en fin de generation.
Returns
-------
str
Texte residuel nettoye pour la synthese vocale ou vide si aucun residu utile.
"""
output: list[str] = []
if self.inDocBlock or self.inResultBlock:
self.resultBuffer += self.buffer
extracted = self.resultBuffer.strip()
if self.isMarkdownFenceDoc and not extracted.endswith("```"):
extracted += "\n```"
self._emitDocument(extracted)
if self.onDocActive:
try:
self.onDocActive(False)
except Exception as exc:
logger.debug("Erreur onDocActive(False) lors de finalize: %s", exc)
self.inDocBlock = False
self.inResultBlock = False
self.isMarkdownFenceDoc = False
self.currentDocTag = None
self.buffer = ""
return ""
if self.inThinkingBlock:
thoughtContent = self.buffer
if self.buffer and self.onThought:
self.onThought(self.buffer)
self.inThinkingBlock = False
self.buffer = ""
draftMatches = RE_DRAFT_MATCHES.findall(thoughtContent)
if draftMatches:
rescued = draftMatches[-1].strip()
rescued = RE_CLEAN_PREFIX.sub("", rescued).strip()
if len(rescued) > 15:
output.append(rescued)
else:
paras = [p.strip() for p in thoughtContent.split("\n\n") if p.strip() and not p.strip().startswith(("*", "-", "#"))]
if paras:
rescued = paras[-1].strip()
rescued = RE_CLEAN_PREFIX.sub("", rescued).strip()
if len(rescued) > 15:
output.append(rescued)
else:
if self.buffer:
clean = RE_XML_STRIP.sub("", self.buffer)
clean = RE_CLEAN_PREFIX.sub("", clean).strip()
if clean:
output.append(clean)
self.buffer = ""
return "".join(output)
[docs]
class FinalAnswerStreamParser:
"""
Parseur temps reel des tokens du LLM lors de l'etape de synthese agentique smolagents.
"""
[docs]
def __init__(self, on_speech_chunk: Callable[[str], None], on_thought_chunk: Callable[[str], None] | None = None) -> None:
"""
Initialise le parseur de tokens agentiques.
Parameters
----------
on_speech_chunk : Callable[[str], None]
Fonction appelee avec les morceaux de texte oral detectes.
on_thought_chunk : Callable[[str], None] | None, optional
Fonction appelee avec les morceaux de raisonnement ou appels d'outils.
"""
self.on_speech_chunk = on_speech_chunk
self.on_thought_chunk = on_thought_chunk
self.buffer = ""
self.in_final_answer = False
self.delimiter: str | None = None
self.finished = False
self.has_streamed = False
[docs]
def process_chunk(self, chunk: str) -> None:
"""
Analyse un jeton du flux agentique pour detecter la reponse finale.
Parameters
----------
chunk : str
Jeton ou bloc textuel emis par le LLM lors de l'appel de final_answer.
"""
if self.finished:
return
self.buffer += chunk
if not self.in_final_answer:
if self.on_thought_chunk:
self.on_thought_chunk(chunk)
m = re.search(r'final_answer\s*\(\s*', self.buffer)
if m:
after_paren = self.buffer[m.end():]
if len(after_paren) >= 3:
if after_paren.startswith(('"""', "'''")):
self.delimiter = after_paren[:3]
elif after_paren.startswith(('"', "'")):
self.delimiter = after_paren[0]
else:
return
self.in_final_answer = True
self.buffer = after_paren[len(self.delimiter):]
else:
return
else:
return
if self.in_final_answer and self.delimiter is not None:
# 1. Detection de balise documentaire
doc_match = re.search(r"<(?:result|code|code_view)\b|```", self.buffer, re.IGNORECASE)
if doc_match:
raw_speech = self.buffer[:doc_match.start()]
cleaned = self._extract_clean_speech(raw_speech, is_doc=True)
if cleaned:
self.on_speech_chunk(cleaned)
self.has_streamed = True
self.finished = True
self.buffer = ""
return
# 2. Detection de la fermeture finale
delim_idx = -1
i = 0
while i <= len(self.buffer) - len(self.delimiter):
if self.buffer[i:i + len(self.delimiter)] == self.delimiter:
num_slashes = 0
k = i - 1
while k >= 0 and self.buffer[k] == "\\":
num_slashes += 1
k -= 1
if num_slashes % 2 == 0:
delim_idx = i
break
i += 1
if delim_idx != -1:
raw_speech = self.buffer[:delim_idx]
has_markdown = bool(re.search(r'^\s*#{1,3}\s+', raw_speech, re.MULTILINE)) or len(raw_speech) > 250
cleaned = self._extract_clean_speech(raw_speech, is_doc=has_markdown)
if cleaned:
self.on_speech_chunk(cleaned)
self.has_streamed = True
self.finished = True
self.buffer = ""
return
# 3. Garde-fou document non balise
has_markdown = bool(re.search(r'^\s*#{1,3}\s+', self.buffer, re.MULTILINE))
if has_markdown and len(self.buffer) >= 180:
cleaned = self._extract_clean_speech(self.buffer, is_doc=True)
if cleaned:
self.on_speech_chunk(cleaned)
self.has_streamed = True
self.finished = True
self.buffer = ""
return
[docs]
def finalize(self) -> None:
"""
Traite les residus si la generation s'est arretee avant la fermeture.
"""
if self.in_final_answer and not self.finished and self.buffer:
doc_match = re.search(r"<(?:result|code|code_view)\b|```", self.buffer, re.IGNORECASE)
raw_text = self.buffer[:doc_match.start()] if doc_match else self.buffer
raw_text = re.sub(r'["\')\]\n\r]+$', '', raw_text)
has_markdown = bool(re.search(r'^\s*#{1,3}\s+', raw_text, re.MULTILINE)) or len(raw_text) > 250
cleaned = self._extract_clean_speech(raw_text, is_doc=has_markdown)
if cleaned:
self.on_speech_chunk(cleaned)
self.has_streamed = True
self.buffer = ""
self.finished = True
@classmethod
def _extract_clean_speech(cls, text: str, is_doc: bool = False) -> str:
"""
Nettoie et extrait 1 a 2 phrases orales fluides en eliminant les titres markdown et listes.
Parameters
----------
text : str
Texte source a nettoyer.
is_doc : bool, optional
Indique si le texte contient une structure documentaire. Defaut : False.
Returns
-------
str
Resume oral concis pour TTS.
"""
text = cls._clean_str_escapes(text).strip()
if not is_doc:
return text
clean_lines = []
for line in text.split("\n"):
stripped = line.strip()
if stripped and not stripped.startswith(("#", "-", "*", "|", ">")):
stripped = re.sub(r"\*\*([^*]+)\*\*", r"\1", stripped)
stripped = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", stripped)
clean_lines.append(stripped)
joined = " ".join(clean_lines).strip()
sentences = [s.strip() for s in re.split(r"(?<=[.!?])\s+", joined) if s.strip()]
if not sentences:
from .prompts import ARCHIVE_ANNOUNCEMENT
return ARCHIVE_ANNOUNCEMENT
summary = sentences[0]
if len(summary) > 250:
last_space = summary.rfind(" ", 0, 245)
summary = (summary[:last_space] if last_space != -1 else summary[:245]) + "..."
if summary and not summary.endswith((".", "!", "?", "...")):
summary = summary.rstrip(",;:") + "."
if not any(k in summary.lower() for k in ("archiver", "archivé", "archiving", "archived")):
from .prompts import ARCHIVE_ANNOUNCEMENT
summary = f"{summary} {ARCHIVE_ANNOUNCEMENT}"
return summary
[docs]
def hasStreamedSpeech(self) -> bool:
"""
Indique si des fragments de discours ont ete transmis a la synthese.
Returns
-------
bool
True si au moins un fragment vocal a ete emis, False sinon.
"""
return self.has_streamed
@staticmethod
def _clean_str_escapes(s: str) -> str:
"""
Nettoie les caracteres d'echappement usuels des chaines Python brutes.
Parameters
----------
s : str
Chaine de caracteres a traiter.
Returns
-------
str
Chaine nettoye avec retours a la ligne et guillemets de-echappes.
"""
return s.replace('\\"', '"').replace("\\'", "'").replace('\\n', ' ').replace('\\t', ' ')