ewannhugging commited on
Commit
50fe1f5
·
1 Parent(s): 3dd506d

test_change

Browse files
Files changed (2) hide show
  1. app.py +75 -22
  2. ingest_train_data.py +209 -105
app.py CHANGED
@@ -36,6 +36,8 @@ from chromadb.config import Settings
36
  from langchain_text_splitters import RecursiveCharacterTextSplitter
37
  from pypdf import PdfReader
38
 
 
 
39
  from llm import call_llm as call_llm_with_metrics
40
 
41
  logging.basicConfig(level=logging.INFO)
@@ -55,9 +57,25 @@ DATA_DIR = Path("/data") if Path("/data").is_dir() else Path("./data")
55
  CHROMA_PERSIST_DIR = str(DATA_DIR / "chroma_db")
56
  TRAIN_DOCS_DIR = Path("./train_data")
57
  COLLECTION_NAME = "rag_documents"
58
- CHUNK_SIZE = 512
59
- CHUNK_OVERLAP = 50
60
- TOP_K_RESULTS = 5
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
61
 
62
  # Settings loaded from data/config.json (runtime config).
63
  # Falls back to the root config.json which serves as the example/template.
@@ -116,6 +134,17 @@ collection = chroma_client.get_or_create_collection(
116
  )
117
  logger.info(f"ChromaDB collection '{COLLECTION_NAME}' ready. Documents: {collection.count()}")
118
 
 
 
 
 
 
 
 
 
 
 
 
119
  # ---------------------------------------------------------------------------
120
  # Step 3: LLM Configuration (Azure Foundry GPT-5)
121
  # ---------------------------------------------------------------------------
@@ -228,14 +257,19 @@ def add_documents_to_vectorstore(documents: list[dict]) -> int:
228
  return len(documents)
229
 
230
 
231
- def retrieve_relevant_context(query: str, top_k: int = TOP_K_RESULTS) -> list[dict]:
 
 
 
 
 
232
  """
233
  Retrieve the most relevant document chunks for a given query.
234
 
235
- This is the "Retrieval" step in RAG:
236
- 1. Embed the user's query using the same embedding model
237
- 2. Search the vector store for the nearest neighbors
238
- 3. Return the top-k most similar chunks as context
239
  """
240
  if collection.count() == 0:
241
  return []
@@ -256,7 +290,19 @@ def retrieve_relevant_context(query: str, top_k: int = TOP_K_RESULTS) -> list[di
256
  "similarity_score": 1 - results["distances"][0][i],
257
  })
258
 
259
- return contexts
 
 
 
 
 
 
 
 
 
 
 
 
260
 
261
 
262
  def build_rag_prompt(query: str, contexts: list[dict]) -> str:
@@ -276,23 +322,30 @@ def build_rag_prompt(query: str, contexts: list[dict]) -> str:
276
  return prompt
277
 
278
 
279
- def rag_query(query: str, top_k: int = TOP_K_RESULTS) -> dict:
 
 
 
 
 
 
 
 
280
  """
281
  End-to-end RAG pipeline: Query → Retrieve → Generate.
282
 
283
- This demonstrates Step 2d: How to make a RAG system end-to-end.
284
-
285
- Pipeline steps:
286
- 1. Receive user query
287
- 2. Retrieve relevant context from vector store
288
- 3. Build augmented prompt with context
289
- 4. Call LLM to generate answer
290
- 5. Return answer with source metadata, explanation, token count, and timing
291
  """
292
  start_time = time.perf_counter()
293
 
294
  # Step 1: Retrieve relevant document chunks
295
- contexts = retrieve_relevant_context(query, top_k=top_k)
 
 
 
 
 
296
 
297
  if not contexts:
298
  elapsed_ms = round((time.perf_counter() - start_time) * 1000, 2)
@@ -318,8 +371,8 @@ def rag_query(query: str, top_k: int = TOP_K_RESULTS) -> dict:
318
  endpoint_url=LLM_ENDPOINT_URL,
319
  api_key=AZURE_API_KEY,
320
  model=LLM_MODEL_NAME,
321
- max_completion_tokens=LLM_MAX_TOKENS,
322
- temperature=LLM_TEMPERATURE,
323
  top_p=LLM_TOP_P,
324
  )
325
  raw_content = llm_result["content"]
@@ -406,7 +459,7 @@ app = FastAPI(
406
  class QueryRequest(BaseModel):
407
  """Request schema for the /query endpoint."""
408
  query: str
409
- top_k: Optional[int] = TOP_K_RESULTS
410
 
411
 
412
  class IngestRequest(BaseModel):
 
36
  from langchain_text_splitters import RecursiveCharacterTextSplitter
37
  from pypdf import PdfReader
38
 
39
+ from sentence_transformers import CrossEncoder
40
+
41
  from llm import call_llm as call_llm_with_metrics
42
 
43
  logging.basicConfig(level=logging.INFO)
 
57
  CHROMA_PERSIST_DIR = str(DATA_DIR / "chroma_db")
58
  TRAIN_DOCS_DIR = Path("./train_data")
59
  COLLECTION_NAME = "rag_documents"
60
+ CHUNK_SIZE = 1000 # était 512
61
+ CHUNK_OVERLAP = 150 # était 50
62
+
63
+ # ---------------------------------------------------------------------------
64
+ # Reranking configuration
65
+ # ---------------------------------------------------------------------------
66
+ # Mettre à True pour activer le reranking, False pour le désactiver
67
+ ENABLE_RERANKING = True
68
+
69
+ # Nombre de chunks récupérés lors de la première étape (recherche vectorielle)
70
+ # Plus ce nombre est élevé, meilleur est le rappel — mais plus le reranker est lent
71
+ TOP_K_RETRIEVAL = 20
72
+
73
+ # Nombre de chunks conservés après le reranking (envoyés au LLM)
74
+ # Doit être inférieur ou égal à TOP_K_RETRIEVAL
75
+ TOP_K_RERANKED = 2
76
+
77
+ # Modèle cross-encoder multilingue (supporte le français)
78
+ RERANKER_MODEL_NAME = "cross-encoder/mmarco-mMiniLMv2-L12-H384-v1"
79
 
80
  # Settings loaded from data/config.json (runtime config).
81
  # Falls back to the root config.json which serves as the example/template.
 
134
  )
135
  logger.info(f"ChromaDB collection '{COLLECTION_NAME}' ready. Documents: {collection.count()}")
136
 
137
+ # ---------------------------------------------------------------------------
138
+ # Step 2b: Initialize Cross-Encoder Reranker
139
+ # ---------------------------------------------------------------------------
140
+ reranker = None
141
+ if ENABLE_RERANKING:
142
+ logger.info(f"Loading reranker model: {RERANKER_MODEL_NAME}")
143
+ reranker = CrossEncoder(RERANKER_MODEL_NAME)
144
+ logger.info("Reranker ready.")
145
+ else:
146
+ logger.info("Reranking disabled (ENABLE_RERANKING = False).")
147
+
148
  # ---------------------------------------------------------------------------
149
  # Step 3: LLM Configuration (Azure Foundry GPT-5)
150
  # ---------------------------------------------------------------------------
 
257
  return len(documents)
258
 
259
 
260
+ def retrieve_relevant_context(
261
+ query: str,
262
+ top_k: int = TOP_K_RETRIEVAL,
263
+ enable_reranking: bool = ENABLE_RERANKING,
264
+ top_k_reranked: int = TOP_K_RERANKED,
265
+ ) -> list[dict]:
266
  """
267
  Retrieve the most relevant document chunks for a given query.
268
 
269
+ Pipeline :
270
+ 1. Recherche vectorielle : récupère top_k candidats
271
+ 2. Si enable_reranking : score chaque paire (query, chunk) avec le cross-encoder
272
+ et conserve les top_k_reranked meilleurs
273
  """
274
  if collection.count() == 0:
275
  return []
 
290
  "similarity_score": 1 - results["distances"][0][i],
291
  })
292
 
293
+ if not enable_reranking or reranker is None:
294
+ return contexts
295
+
296
+ # Étape 2 : reranking avec le cross-encoder
297
+ pairs = [[query, ctx["text"]] for ctx in contexts]
298
+ rerank_scores = reranker.predict(pairs)
299
+
300
+ for ctx, score in zip(contexts, rerank_scores):
301
+ ctx["rerank_score"] = float(score)
302
+
303
+ contexts.sort(key=lambda x: x["rerank_score"], reverse=True)
304
+
305
+ return contexts[:top_k_reranked]
306
 
307
 
308
  def build_rag_prompt(query: str, contexts: list[dict]) -> str:
 
322
  return prompt
323
 
324
 
325
+ def rag_query(
326
+ query: str,
327
+ top_k: int = TOP_K_RETRIEVAL,
328
+ *,
329
+ temperature: float = LLM_TEMPERATURE,
330
+ max_tokens: int = LLM_MAX_TOKENS,
331
+ enable_reranking: bool = ENABLE_RERANKING,
332
+ top_k_reranked: int = TOP_K_RERANKED,
333
+ ) -> dict:
334
  """
335
  End-to-end RAG pipeline: Query → Retrieve → Generate.
336
 
337
+ Tous les paramètres ont des valeurs par défaut issues de la configuration globale
338
+ et peuvent être surchargés lors de l'appel (ex : sensitivity analysis).
 
 
 
 
 
 
339
  """
340
  start_time = time.perf_counter()
341
 
342
  # Step 1: Retrieve relevant document chunks
343
+ contexts = retrieve_relevant_context(
344
+ query,
345
+ top_k=top_k,
346
+ enable_reranking=enable_reranking,
347
+ top_k_reranked=top_k_reranked,
348
+ )
349
 
350
  if not contexts:
351
  elapsed_ms = round((time.perf_counter() - start_time) * 1000, 2)
 
371
  endpoint_url=LLM_ENDPOINT_URL,
372
  api_key=AZURE_API_KEY,
373
  model=LLM_MODEL_NAME,
374
+ max_completion_tokens=max_tokens,
375
+ temperature=temperature,
376
  top_p=LLM_TOP_P,
377
  )
378
  raw_content = llm_result["content"]
 
459
  class QueryRequest(BaseModel):
460
  """Request schema for the /query endpoint."""
461
  query: str
462
+ top_k: Optional[int] = TOP_K_RETRIEVAL
463
 
464
 
465
  class IngestRequest(BaseModel):
ingest_train_data.py CHANGED
@@ -1,24 +1,29 @@
1
  """
2
- Ingest Training Data Script
3
- ============================
4
 
5
- Standalone script to convert all documents in ./train_data into embeddings
6
- and store them in the ChromaDB vector store.
 
 
7
 
8
  Usage:
9
- python ingest_train_data.py
 
10
 
11
- Requires:
12
- - AZURE_API_KEY environment variable set
13
- - data/config.json (or root config.json as fallback) with endpoint URLs
14
- - Documents (PDF/TXT) in ./train_data/<category>/
15
  """
16
 
17
- import os
18
- import sys
19
  import json
20
  import logging
 
 
21
  import time
 
22
  from pathlib import Path
23
 
24
  os.environ.setdefault("ANONYMIZED_TELEMETRY", "False")
@@ -32,40 +37,56 @@ from pypdf import PdfReader
32
  logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
33
  logger = logging.getLogger(__name__)
34
 
35
- # ---------------------------------------------------------------------------
36
- # Configuration
37
- # ---------------------------------------------------------------------------
38
 
39
- PROJECT_ROOT = Path(__file__).parent
40
- DATA_DIR = Path("/data") if Path("/data").is_dir() else PROJECT_ROOT / "data"
41
  TRAIN_DATA_DIR = Path("/train_data") if Path("/train_data").is_dir() else PROJECT_ROOT / "train_data"
42
 
43
  CHROMA_PERSIST_DIR = str(DATA_DIR / "chroma_db")
44
- COLLECTION_NAME = "rag_documents"
45
- CHUNK_SIZE = 512
46
- CHUNK_OVERLAP = 50
47
- EMBEDDING_BATCH_SIZE = 16
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
48
 
49
  _CONFIG_PATH = DATA_DIR / "config.json"
50
  if not _CONFIG_PATH.exists():
51
  _CONFIG_PATH = PROJECT_ROOT / "config.json"
52
- logger.warning(f"No config.json in {DATA_DIR} — using root config.json as fallback.")
53
 
54
  with open(_CONFIG_PATH, encoding="utf-8") as _f:
55
  _config = json.load(_f)
56
 
57
  EMBEDDING_ENDPOINT_URL = _config["embedding"]["endpoint_url"]
58
- EMBEDDING_MODEL_NAME = _config["embedding"]["model"]
59
 
60
  AZURE_API_KEY = os.environ.get("AZURE_API_KEY")
61
  if not AZURE_API_KEY:
62
- logger.error("AZURE_API_KEY is not set. Export it before running this script.")
63
  sys.exit(1)
64
 
65
- # ---------------------------------------------------------------------------
66
  # Helpers
67
- # ---------------------------------------------------------------------------
68
-
69
 
70
  def extract_text_from_pdf(pdf_path: Path) -> str:
71
  reader = PdfReader(str(pdf_path))
@@ -88,109 +109,192 @@ def chunk_text(text: str, source: str) -> list[dict]:
88
 
89
 
90
  def generate_embeddings(texts: list[str]) -> list[list[float]]:
 
91
  headers = {"api-key": AZURE_API_KEY, "Content-Type": "application/json"}
92
  payload = {"input": texts, "model": EMBEDDING_MODEL_NAME}
93
  resp = http_requests.post(EMBEDDING_ENDPOINT_URL, headers=headers, json=payload, timeout=120)
94
  resp.raise_for_status()
95
- data = resp.json()
96
- return [item["embedding"] for item in data["data"]]
97
-
98
-
99
- def generate_embeddings_batched(texts: list[str]) -> list[list[float]]:
100
- all_embeddings = []
101
- for i in range(0, len(texts), EMBEDDING_BATCH_SIZE):
102
- batch = texts[i:i + EMBEDDING_BATCH_SIZE]
103
- logger.info(f" Embedding batch {i // EMBEDDING_BATCH_SIZE + 1} ({len(batch)} chunks)...")
104
- embeddings = generate_embeddings(batch)
105
- all_embeddings.extend(embeddings)
106
- time.sleep(0.5) # Rate-limit courtesy
107
- return all_embeddings
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
108
 
109
 
110
- # ---------------------------------------------------------------------------
111
  # Main
112
- # ---------------------------------------------------------------------------
113
-
114
 
115
  def main():
 
 
 
 
 
 
 
116
  if not TRAIN_DATA_DIR.exists():
117
- logger.error(f"Training data directory not found: {TRAIN_DATA_DIR}")
118
  sys.exit(1)
119
 
120
- logger.info(f"Training data directory: {TRAIN_DATA_DIR}")
121
- logger.info(f"ChromaDB path: {CHROMA_PERSIST_DIR}")
122
- logger.info(f"Embedding model: {EMBEDDING_MODEL_NAME}")
123
-
124
  chroma_client = chromadb.PersistentClient(
125
  path=CHROMA_PERSIST_DIR,
126
  settings=Settings(anonymized_telemetry=False),
127
  )
 
 
 
 
 
 
128
  collection = chroma_client.get_or_create_collection(
129
  name=COLLECTION_NAME,
130
  metadata={"hnsw:space": "cosine"},
131
  )
132
- logger.info(f"ChromaDB collection '{COLLECTION_NAME}' — existing documents: {collection.count()}")
133
-
134
- # Gather all PDF and TXT files recursively
135
- files = sorted(
136
- list(TRAIN_DATA_DIR.rglob("*.pdf")) + list(TRAIN_DATA_DIR.rglob("*.txt"))
137
- )
138
- logger.info(f"Found {len(files)} files to process.")
139
-
140
- total_chunks_added = 0
141
-
142
- for idx, file_path in enumerate(files, start=1):
143
- relative = file_path.relative_to(TRAIN_DATA_DIR)
144
- # Use category/filename as the source identifier
145
- source = str(relative)
146
- logger.info(f"[{idx}/{len(files)}] Processing: {source}")
147
-
148
- try:
149
- if file_path.suffix.lower() == ".pdf":
150
- text = extract_text_from_pdf(file_path)
151
- else:
152
- text = file_path.read_text(encoding="utf-8")
153
- except Exception as e:
154
- logger.warning(f" Skipped (read error): {e}")
155
- continue
156
-
157
- if not text.strip():
158
- logger.warning(f" Skipped (no extractable text)")
159
- continue
160
-
161
- chunks = chunk_text(text, source=source)
162
- if not chunks:
163
- logger.warning(f" Skipped (no chunks produced)")
164
- continue
165
-
166
- logger.info(f" {len(chunks)} chunks extracted")
167
-
168
- texts = [c["text"] for c in chunks]
169
- try:
170
- embeddings = generate_embeddings_batched(texts)
171
- except http_requests.exceptions.HTTPError as e:
172
- logger.error(f" Embedding failed: {e}")
173
- continue
174
- except Exception as e:
175
- logger.error(f" Unexpected embedding error: {e}")
176
- continue
177
 
178
- existing_count = collection.count()
179
- ids = [f"doc_{existing_count + i}" for i in range(len(chunks))]
180
- metadatas = [{"source": c["source"], "chunk_index": c["chunk_index"]} for c in chunks]
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
181
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
182
  collection.add(
183
- ids=ids,
184
- embeddings=embeddings,
185
- documents=texts,
186
- metadatas=metadatas,
187
  )
188
- total_chunks_added += len(chunks)
189
- logger.info(f" Added {len(chunks)} chunks (total in store: {collection.count()})")
190
-
191
- logger.info("=" * 60)
192
- logger.info(f"Ingestion complete. Chunks added this run: {total_chunks_added}")
193
- logger.info(f"Total documents in vector store: {collection.count()}")
 
 
 
 
 
 
194
 
195
 
196
  if __name__ == "__main__":
 
1
  """
2
+ Ingest Training Data Script — Version parallélisée
3
+ =====================================================
4
 
5
+ Trois phases :
6
+ 1. Extraction + chunking de tous les fichiers (ThreadPoolExecutor, I/O + CPU)
7
+ 2. Embedding via Azure OpenAI (ThreadPoolExecutor, I/O pur → gros gain)
8
+ 3. Insertion bulk dans ChromaDB (1 seul appel par tranche de 5000)
9
 
10
  Usage:
11
+ python ingest_train_data.py # ingestion normale (skip si déjà fait)
12
+ python ingest_train_data.py --reset # supprime la collection et recommence
13
 
14
+ Paramètres de parallélisme (ajuster selon les limites de l'API Azure) :
15
+ EXTRACTION_WORKERS : workers pour la lecture/parsing des PDFs
16
+ EMBEDDING_WORKERS : workers pour les appels HTTP à l'API d'embedding
17
+ EMBEDDING_BATCH_SIZE: chunks par requête (max ~2048 pour Azure OpenAI)
18
  """
19
 
20
+ import argparse
 
21
  import json
22
  import logging
23
+ import os
24
+ import sys
25
  import time
26
+ from concurrent.futures import ThreadPoolExecutor, as_completed
27
  from pathlib import Path
28
 
29
  os.environ.setdefault("ANONYMIZED_TELEMETRY", "False")
 
37
  logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
38
  logger = logging.getLogger(__name__)
39
 
40
+ # ──────────────────────────────────────────────────────────────────────────────
41
+ # Configuration test de la configuration
42
+ # ──────────────────────────────────────────────────────────────────────────────
43
 
44
+ PROJECT_ROOT = Path(__file__).parent
45
+ DATA_DIR = Path("/data") if Path("/data").is_dir() else PROJECT_ROOT / "data"
46
  TRAIN_DATA_DIR = Path("/train_data") if Path("/train_data").is_dir() else PROJECT_ROOT / "train_data"
47
 
48
  CHROMA_PERSIST_DIR = str(DATA_DIR / "chroma_db")
49
+ COLLECTION_NAME = "rag_documents"
50
+
51
+ # Chunking — mis à jour pour de meilleures performances RAG
52
+ CHUNK_SIZE = 1000 # était 512
53
+ CHUNK_OVERLAP = 150 # était 50
54
+
55
+ # Parallélisme
56
+ EMBEDDING_BATCH_SIZE = 64 # chunks par requête API (était 16)
57
+ EMBEDDING_WORKERS = 6 # appels API simultanés — réduire si rate-limit Azure
58
+ EXTRACTION_WORKERS = 8 # workers pour l'extraction/parsing PDF
59
+
60
+ # Insertion ChromaDB par tranches pour éviter les problèmes mémoire
61
+ CHROMA_INSERT_BATCH = 5_000
62
+
63
+ # Retry sur les appels embedding
64
+ EMBEDDING_MAX_RETRIES = 3
65
+ EMBEDDING_RETRY_DELAY = 2.0 # secondes entre deux tentatives
66
+
67
+ # ──────────────────────────────────────────────────────────────────────────────
68
+ # Chargement de la config
69
+ # ──────────────────────────────────────────────────────────────────────────────
70
 
71
  _CONFIG_PATH = DATA_DIR / "config.json"
72
  if not _CONFIG_PATH.exists():
73
  _CONFIG_PATH = PROJECT_ROOT / "config.json"
74
+ logger.warning(f"Pas de config.json dans {DATA_DIR} — utilisation de root config.json.")
75
 
76
  with open(_CONFIG_PATH, encoding="utf-8") as _f:
77
  _config = json.load(_f)
78
 
79
  EMBEDDING_ENDPOINT_URL = _config["embedding"]["endpoint_url"]
80
+ EMBEDDING_MODEL_NAME = _config["embedding"]["model"]
81
 
82
  AZURE_API_KEY = os.environ.get("AZURE_API_KEY")
83
  if not AZURE_API_KEY:
84
+ logger.error("AZURE_API_KEY n'est pas défini. Définissez la variable d'environnement avant de lancer.")
85
  sys.exit(1)
86
 
87
+ # ────────────────────────────���─────────────────────────────────────────────────
88
  # Helpers
89
+ # ──────────────────────────────────────────────────────────────────────────────
 
90
 
91
  def extract_text_from_pdf(pdf_path: Path) -> str:
92
  reader = PdfReader(str(pdf_path))
 
109
 
110
 
111
  def generate_embeddings(texts: list[str]) -> list[list[float]]:
112
+ """Un seul appel HTTP — appelé en parallèle par plusieurs workers."""
113
  headers = {"api-key": AZURE_API_KEY, "Content-Type": "application/json"}
114
  payload = {"input": texts, "model": EMBEDDING_MODEL_NAME}
115
  resp = http_requests.post(EMBEDDING_ENDPOINT_URL, headers=headers, json=payload, timeout=120)
116
  resp.raise_for_status()
117
+ return [item["embedding"] for item in resp.json()["data"]]
118
+
119
+
120
+ # ──────────────────────────────────────────────────────────────────────────────
121
+ # Workers (appelés depuis les ThreadPoolExecutors)
122
+ # ──────────────────────────────────────────────────────────────────────────────
123
+
124
+ def _process_file(file_path: Path) -> tuple[str, list[dict]]:
125
+ """Phase 1 — Extrait et chunke un fichier. Retourne (source, chunks)."""
126
+ source = str(file_path.relative_to(TRAIN_DATA_DIR))
127
+ try:
128
+ text = extract_text_from_pdf(file_path) if file_path.suffix.lower() == ".pdf" \
129
+ else file_path.read_text(encoding="utf-8")
130
+ except Exception as e:
131
+ logger.warning(f" Ignoré (erreur lecture) {source}: {e}")
132
+ return source, []
133
+ if not text.strip():
134
+ logger.warning(f" Ignoré (pas de texte) {source}")
135
+ return source, []
136
+ chunks = chunk_text(text, source=source)
137
+ return source, chunks
138
+
139
+
140
+ def _embed_batch(batch_idx: int, texts: list[str]) -> tuple[int, list[list[float]]]:
141
+ """Phase 2 — Embedde un batch avec retry. Retourne (batch_idx, embeddings)."""
142
+ for attempt in range(1, EMBEDDING_MAX_RETRIES + 1):
143
+ try:
144
+ return batch_idx, generate_embeddings(texts)
145
+ except Exception as e:
146
+ if attempt == EMBEDDING_MAX_RETRIES:
147
+ logger.error(f" Batch {batch_idx} échoué après {EMBEDDING_MAX_RETRIES} tentatives : {e}")
148
+ raise
149
+ logger.warning(f" Batch {batch_idx} — tentative {attempt} échouée ({e}), retry dans {EMBEDDING_RETRY_DELAY}s...")
150
+ time.sleep(EMBEDDING_RETRY_DELAY)
151
 
152
 
153
+ # ──────────────────────────────────────────────────────────────────────────────
154
  # Main
155
+ # ──────────────────────────────────────────────────────────────────────────────
 
156
 
157
  def main():
158
+ parser = argparse.ArgumentParser(description="Ingestion parallélisée des documents d'entraînement")
159
+ parser.add_argument(
160
+ "--reset", action="store_true",
161
+ help="Supprime la collection ChromaDB existante avant de réingérer (nécessaire après changement de chunking)",
162
+ )
163
+ args = parser.parse_args()
164
+
165
  if not TRAIN_DATA_DIR.exists():
166
+ logger.error(f"Dossier train_data introuvable : {TRAIN_DATA_DIR}")
167
  sys.exit(1)
168
 
169
+ # ── ChromaDB ──────────────────────────────────���───────────────────────────
 
 
 
170
  chroma_client = chromadb.PersistentClient(
171
  path=CHROMA_PERSIST_DIR,
172
  settings=Settings(anonymized_telemetry=False),
173
  )
174
+ if args.reset:
175
+ try:
176
+ chroma_client.delete_collection(COLLECTION_NAME)
177
+ logger.info("Collection supprimée (--reset).")
178
+ except Exception:
179
+ pass
180
  collection = chroma_client.get_or_create_collection(
181
  name=COLLECTION_NAME,
182
  metadata={"hnsw:space": "cosine"},
183
  )
184
+ if collection.count() > 0 and not args.reset:
185
+ logger.info(f"Collection déjà peuplée ({collection.count()} chunks). Utilisez --reset pour réingérer.")
186
+ sys.exit(0)
187
+
188
+ # ── Découverte des fichiers ───────────────────────────────────────────────
189
+ files = sorted(list(TRAIN_DATA_DIR.rglob("*.pdf")) + list(TRAIN_DATA_DIR.rglob("*.txt")))
190
+ if not files:
191
+ logger.error(f"Aucun fichier PDF/TXT trouvé dans {TRAIN_DATA_DIR}")
192
+ sys.exit(1)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
193
 
194
+ t_start = time.perf_counter()
195
+ logger.info(f"{'='*60}")
196
+ logger.info(f"Fichiers trouvés : {len(files)}")
197
+ logger.info(f"Chunk size / overlap: {CHUNK_SIZE} / {CHUNK_OVERLAP}")
198
+ logger.info(f"Batch size embedding: {EMBEDDING_BATCH_SIZE}")
199
+ logger.info(f"Workers extraction : {EXTRACTION_WORKERS}")
200
+ logger.info(f"Workers embedding : {EMBEDDING_WORKERS}")
201
+ logger.info(f"{'='*60}")
202
+
203
+ # ── Phase 1 : Extraction + chunking en parallèle ─────────────────────────
204
+ logger.info(f"\nPhase 1/3 — Extraction et chunking ({EXTRACTION_WORKERS} workers)...")
205
+ all_chunks: list[dict] = []
206
+ errors_extraction = 0
207
+
208
+ with ThreadPoolExecutor(max_workers=EXTRACTION_WORKERS) as executor:
209
+ futures = {executor.submit(_process_file, f): f for f in files}
210
+ done = 0
211
+ for future in as_completed(futures):
212
+ done += 1
213
+ try:
214
+ source, chunks = future.result()
215
+ all_chunks.extend(chunks)
216
+ logger.info(f" [{done}/{len(files)}] {source} → {len(chunks)} chunks")
217
+ except Exception as e:
218
+ errors_extraction += 1
219
+ logger.error(f" [{done}/{len(files)}] Erreur inattendue : {e}")
220
+
221
+ t_phase1 = time.perf_counter() - t_start
222
+ logger.info(f"Phase 1 terminée en {t_phase1:.1f}s — {len(all_chunks)} chunks ({errors_extraction} erreurs)")
223
+
224
+ if not all_chunks:
225
+ logger.error("Aucun chunk produit. Vérifiez les fichiers dans train_data/.")
226
+ sys.exit(1)
227
 
228
+ # ── Phase 2 : Embedding en parallèle ─────────────────────────────────────
229
+ logger.info(f"\nPhase 2/3 — Embedding ({len(all_chunks)} chunks, batch={EMBEDDING_BATCH_SIZE}, workers={EMBEDDING_WORKERS})...")
230
+
231
+ texts = [c["text"] for c in all_chunks]
232
+ batches = [(i, texts[i : i + EMBEDDING_BATCH_SIZE]) for i in range(0, len(texts), EMBEDDING_BATCH_SIZE)]
233
+ embeddings_map: dict[int, list] = {}
234
+ errors_embedding = 0
235
+
236
+ t_phase2_start = time.perf_counter()
237
+ with ThreadPoolExecutor(max_workers=EMBEDDING_WORKERS) as executor:
238
+ futures = {executor.submit(_embed_batch, idx, batch): idx for idx, batch in batches}
239
+ done = 0
240
+ for future in as_completed(futures):
241
+ done += 1
242
+ try:
243
+ batch_idx, embeddings = future.result()
244
+ embeddings_map[batch_idx] = embeddings
245
+ except Exception as e:
246
+ errors_embedding += 1
247
+ logger.error(f" Batch échoué, {errors_embedding} erreurs totales")
248
+ if done % 20 == 0 or done == len(batches):
249
+ elapsed = time.perf_counter() - t_phase2_start
250
+ rate = done / elapsed if elapsed > 0 else 0
251
+ eta = (len(batches) - done) / rate if rate > 0 else 0
252
+ logger.info(f" {done}/{len(batches)} batches — {elapsed:.0f}s écoulés, ETA ~{eta:.0f}s")
253
+
254
+ t_phase2 = time.perf_counter() - t_phase2_start
255
+ logger.info(f"Phase 2 terminée en {t_phase2:.1f}s ({errors_embedding} erreurs)")
256
+
257
+ if errors_embedding > 0:
258
+ logger.warning(f"{errors_embedding} batches ont échoué — les chunks correspondants seront absents de la base.")
259
+
260
+ # Reconstruction dans l'ordre original
261
+ flat_embeddings: list = []
262
+ failed_indices: set[int] = set()
263
+ valid_chunks: list[dict] = []
264
+
265
+ for batch_start in range(0, len(texts), EMBEDDING_BATCH_SIZE):
266
+ if batch_start in embeddings_map:
267
+ flat_embeddings.extend(embeddings_map[batch_start])
268
+ batch_end = min(batch_start + EMBEDDING_BATCH_SIZE, len(all_chunks))
269
+ valid_chunks.extend(all_chunks[batch_start:batch_end])
270
+ else:
271
+ failed_indices.add(batch_start)
272
+
273
+ # ── Phase 3 : Insertion bulk dans ChromaDB ───────────────────────────────
274
+ logger.info(f"\nPhase 3/3 — Insertion dans ChromaDB ({len(valid_chunks)} chunks)...")
275
+ t_phase3_start = time.perf_counter()
276
+
277
+ for i in range(0, len(valid_chunks), CHROMA_INSERT_BATCH):
278
+ batch_chunks = valid_chunks[i : i + CHROMA_INSERT_BATCH]
279
+ batch_embeddings = flat_embeddings[i : i + CHROMA_INSERT_BATCH]
280
  collection.add(
281
+ ids = [f"doc_{i + j}" for j in range(len(batch_chunks))],
282
+ embeddings = batch_embeddings,
283
+ documents = [c["text"] for c in batch_chunks],
284
+ metadatas = [{"source": c["source"], "chunk_index": c["chunk_index"]} for c in batch_chunks],
285
  )
286
+ logger.info(f" Inséré {min(i + CHROMA_INSERT_BATCH, len(valid_chunks))}/{len(valid_chunks)} chunks")
287
+
288
+ t_phase3 = time.perf_counter() - t_phase3_start
289
+ t_total = time.perf_counter() - t_start
290
+
291
+ logger.info(f"\n{'='*60}")
292
+ logger.info(f"Ingestion terminée en {t_total:.1f}s ({t_total/60:.1f} min)")
293
+ logger.info(f" Phase 1 (extraction) : {t_phase1:.1f}s")
294
+ logger.info(f" Phase 2 (embedding) : {t_phase2:.1f}s")
295
+ logger.info(f" Phase 3 (ChromaDB) : {t_phase3:.1f}s")
296
+ logger.info(f"Chunks dans la base : {collection.count()}")
297
+ logger.info(f"{'='*60}")
298
 
299
 
300
  if __name__ == "__main__":