From 8369886f42efde5751f3521b9fd4ea72edf796cc Mon Sep 17 00:00:00 2001 From: "IETM_FIXE\\ietm6" Date: Wed, 10 Jun 2026 15:18:51 +0200 Subject: [PATCH] Parallelisation des appels MAP (import campagne + analyse approfondie) Les morceaux/lots sont traites par vagues de llm_map_concurrency appels simultanes (defaut 3, .env). L ordre narratif est preserve (fusion vague par vague dans l ordre du livre), la resilience par morceau et les heartbeats SSE sont conserves. Divise le temps d import d un gros livre par ~3 sur un provider cloud ; sans effet sur Ollama local (qui sequence cote serveur). Co-Authored-By: Claude Opus 4.8 --- brain/app/api/deps.py | 12 ++- brain/app/application/import_campaign.py | 94 +++++++++++++++--------- brain/app/application/notebook_deep.py | 50 +++++++++---- brain/app/core/config.py | 6 ++ 4 files changed, 109 insertions(+), 53 deletions(-) diff --git a/brain/app/api/deps.py b/brain/app/api/deps.py index 616d1b9..8d58b6c 100644 --- a/brain/app/api/deps.py +++ b/brain/app/api/deps.py @@ -92,7 +92,11 @@ def get_import_campaign_use_case( ) -> ImportCampaignUseCase: """Factory du use case d'import de campagne PDF (extraction + arborescence).""" return ImportCampaignUseCase( - llm=llm, extractor=_PDF_EXTRACTOR, chunk_target_tokens=settings.import_chunk_tokens) + llm=llm, + extractor=_PDF_EXTRACTOR, + chunk_target_tokens=settings.import_chunk_tokens, + map_concurrency=settings.llm_map_concurrency, + ) def get_adapt_campaign_use_case( @@ -140,4 +144,8 @@ def get_notebook_deep_use_case( llm: Annotated[LLMProvider, Depends(get_llm_provider)], settings: Annotated[Settings, Depends(get_settings)], ) -> NotebookDeepUseCase: - return NotebookDeepUseCase(llm=llm, batch_tokens=settings.import_chunk_tokens) + return NotebookDeepUseCase( + llm=llm, + batch_tokens=settings.import_chunk_tokens, + map_concurrency=settings.llm_map_concurrency, + ) diff --git a/brain/app/application/import_campaign.py b/brain/app/application/import_campaign.py index 17d0794..a238a1a 100644 --- a/brain/app/application/import_campaign.py +++ b/brain/app/application/import_campaign.py @@ -10,6 +10,7 @@ PROPOSITION non persistée : le Core crée les entités seulement après revue. """ from __future__ import annotations +import asyncio import logging from app.application.chunking import chunk_text, split_in_half @@ -217,10 +218,15 @@ class ImportCampaignUseCase: llm: LLMProvider, extractor: PdfTextExtractor, chunk_target_tokens: int = _CHUNK_TARGET_TOKENS, + map_concurrency: int = 1, ) -> None: self._llm = llm self._extractor = extractor self._chunk_target_tokens = chunk_target_tokens + # Appels MAP par VAGUES de cette taille : l'ordre narratif est préservé + # (fusion vague par vague, dans l'ordre du livre) mais le mur d'attente + # des appels LLM est divisé d'autant. 1 = comportement séquentiel. + self._map_concurrency = max(1, map_concurrency) async def execute(self, pdf_bytes: bytes) -> CampaignImportResult: """Variante non-streamée : traite tout puis renvoie l'arbre complet.""" @@ -228,9 +234,15 @@ class ImportCampaignUseCase: chunks = chunk_text(doc.full_text, self._chunk_target_tokens) toc_block = _format_toc(doc.toc) merger = _TreeMerger() - for i, chunk in enumerate(chunks): - merger.add(await self._map_chunk( - chunk, index=i, total=len(chunks), toc_block=toc_block)) + total = len(chunks) + for start in range(0, total, self._map_concurrency): + wave = list(enumerate(chunks))[start:start + self._map_concurrency] + results = await asyncio.gather(*( + self._map_chunk(c, index=i, total=total, toc_block=toc_block) + for i, c in wave + )) + for res in results: + merger.add(res) return CampaignImportResult( arcs=merger.result(), page_count=doc.page_count, @@ -266,38 +278,50 @@ class ImportCampaignUseCase: merger = _TreeMerger() skipped = 0 last_error: str | None = None - for i, chunk in enumerate(chunks): - # RÉSILIENCE : un morceau qui échoue (provider saturé, quota, etc.) est - # SAUTÉ — on ne perd pas tout l'import pour autant. On n'abandonne que - # si AUCUN morceau ne passe (cf. après la boucle). - # HEARTBEAT : keep-alive pendant l'appel LLM pour ne jamais laisser le - # flux SSE silencieux (sinon le Core coupe sur timeout d'inactivité). - try: - arcs_payload: list[dict] | None = None - async for kind, payload in with_heartbeat( - self._map_chunk(chunk, index=i, total=total, toc_block=toc_block) - ): - if kind == "heartbeat": - yield {"type": "heartbeat", "current": i + 1, "total": total} - else: - arcs_payload = payload - merger.add(arcs_payload or []) - except LLMProviderError as exc: - skipped += 1 - last_error = str(exc) - logger.warning("Morceau %s/%s ignoré (échec LLM) : %s", i + 1, total, exc) - yield {"type": "chunk_failed", "current": i + 1, "total": total, - "message": str(exc)[:300]} - arcs, chapters, scenes = merger.counts() - yield { - "type": "progress", - "current": i + 1, - "total": total, - "arc_count": arcs, - "chapter_count": chapters, - "scene_count": scenes, - "skipped": skipped, - } + done_count = 0 + # PARALLÉLISME : les morceaux sont traités par VAGUES de `map_concurrency` + # appels simultanés. L'ordre narratif est préservé : la fusion se fait + # vague par vague, dans l'ordre du livre. + # RÉSILIENCE : un morceau qui échoue (provider saturé, quota, etc.) est + # SAUTÉ — on ne perd pas tout l'import pour autant. On n'abandonne que + # si AUCUN morceau ne passe (cf. après la boucle). + # HEARTBEAT : keep-alive pendant la vague d'appels LLM pour ne jamais + # laisser le flux SSE silencieux (sinon le Core coupe sur inactivité). + for start in range(0, total, self._map_concurrency): + wave = list(enumerate(chunks))[start:start + self._map_concurrency] + gathered = asyncio.gather( + *(self._map_chunk(c, index=i, total=total, toc_block=toc_block) + for i, c in wave), + return_exceptions=True, + ) + results: list | None = None + async for kind, payload in with_heartbeat(gathered): + if kind == "heartbeat": + yield {"type": "heartbeat", "current": done_count + 1, "total": total} + else: + results = payload + for (i, _), res in zip(wave, results or []): + done_count += 1 + if isinstance(res, LLMProviderError): + skipped += 1 + last_error = str(res) + logger.warning("Morceau %s/%s ignoré (échec LLM) : %s", i + 1, total, res) + yield {"type": "chunk_failed", "current": i + 1, "total": total, + "message": str(res)[:300]} + elif isinstance(res, BaseException): + raise res # bug inattendu : ne pas l'avaler en silence + else: + merger.add(res or []) + arcs, chapters, scenes = merger.counts() + yield { + "type": "progress", + "current": done_count, + "total": total, + "arc_count": arcs, + "chapter_count": chapters, + "scene_count": scenes, + "skipped": skipped, + } if total > 0 and skipped == total: # Tout a échoué : "done" vide serait trompeur → erreur explicite. diff --git a/brain/app/application/notebook_deep.py b/brain/app/application/notebook_deep.py index 7030ece..5fa1f7d 100644 --- a/brain/app/application/notebook_deep.py +++ b/brain/app/application/notebook_deep.py @@ -13,6 +13,7 @@ lots ; avec un petit modèle local, plus de lots (mais ça reste exhaustif). """ from __future__ import annotations +import asyncio import logging from typing import AsyncIterator @@ -62,9 +63,13 @@ Réponds en français.""" class NotebookDeepUseCase: - def __init__(self, llm: LLMProvider, batch_tokens: int = 10000) -> None: + def __init__( + self, llm: LLMProvider, batch_tokens: int = 10000, map_concurrency: int = 1 + ) -> None: self._llm = llm self._batch_tokens = max(2000, batch_tokens) + # Lots MAP traités par vagues de cette taille (parallélisme LLM). + self._map_concurrency = max(1, map_concurrency) async def stream( self, @@ -92,21 +97,21 @@ class NotebookDeepUseCase: batches = self._group(chunks) total = len(batches) notes: list[str] = [] - for i, batch in enumerate(batches): - yield {"type": "progress", "current": i, "total": total} - excerpt = "\n\n".join( - f"(p. {c['page']}) {c['text'].strip()}" if c.get("page") else c["text"].strip() - for c in batch - ) - prompt = _MAP_PROMPT.format(no_match=_NO_MATCH, question=question, excerpt=excerpt) - try: - raw = await generate_with_retry(self._llm, prompt, temperature=_MAP_TEMPERATURE) - except LLMProviderError as exc: - logger.warning("Analyse approfondie : lot %s/%s ignoré : %s", i + 1, total, exc) - continue - answer = raw.strip() - if answer and answer.upper().rstrip(".") != _NO_MATCH: - notes.append(answer) + # Lots traités par VAGUES parallèles ; les notes restent dans l'ordre du + # document (gather préserve l'ordre des tâches de la vague). + for start in range(0, total, self._map_concurrency): + yield {"type": "progress", "current": start, "total": total} + wave = batches[start:start + self._map_concurrency] + results = await asyncio.gather( + *(self._map_batch(question, b) for b in wave), return_exceptions=True) + for j, res in enumerate(results): + if isinstance(res, LLMProviderError): + logger.warning( + "Analyse approfondie : lot %s/%s ignoré : %s", start + j + 1, total, res) + elif isinstance(res, BaseException): + raise res # bug inattendu : ne pas l'avaler + elif res: + notes.append(res) yield {"type": "progress", "current": total, "total": total} notes_block = "\n\n".join(notes) if notes else "(aucune information pertinente trouvée dans le document)" @@ -136,6 +141,19 @@ class NotebookDeepUseCase: )} yield {"type": "done"} + async def _map_batch(self, question: str, batch: list[dict]) -> str: + """Phase MAP d'un lot : extrait les infos pertinentes ('' si RAS).""" + excerpt = "\n\n".join( + f"(p. {c['page']}) {c['text'].strip()}" if c.get("page") else c["text"].strip() + for c in batch + ) + prompt = _MAP_PROMPT.format(no_match=_NO_MATCH, question=question, excerpt=excerpt) + raw = await generate_with_retry(self._llm, prompt, temperature=_MAP_TEMPERATURE) + answer = raw.strip() + if answer and answer.upper().rstrip(".") != _NO_MATCH: + return answer + return "" + def _group(self, chunks: list[dict]) -> list[list[dict]]: """Regroupe les extraits en lots ~`batch_tokens` (compte tiktoken).""" enc = tiktoken.get_encoding("cl100k_base") diff --git a/brain/app/core/config.py b/brain/app/core/config.py index 007d6f6..aa11395 100644 --- a/brain/app/core/config.py +++ b/brain/app/core/config.py @@ -90,6 +90,12 @@ class Settings(BaseSettings): # les hors-sujet 0.2-0.4. Montable à ~0.4 si trop de bruit, 0 = désactivé. rag_min_score: float = 0.30 + # Nombre d'appels LLM MAP menes EN PARALLELE (import de campagne, analyse + # approfondie). 3 = bon defaut cloud (divise le temps d'un gros livre par ~3). + # Ollama local sequence les requetes de toute facon (pas de gain, pas de mal). + # Baisser a 1 si un provider gratuit rate-limite agressivement. + llm_map_concurrency: int = 3 + # Taille cible d'un morceau (en tokens) pour l'import de PDF (regles/campagne). # Plus c'est gros, moins il y a de morceaux => moins de fragmentation et un # import plus rapide, MAIS il faut que ca tienne dans la fenetre du modele.