--- title: OpenWebUI mise à jours base de connaissance author: Frederic AOUSTIN version: 1.0 last_date: 21/08/2026 --- # OpenWebUI mise à jours base de connaissance ![category](developpement) ![tag](python) ![tag](ia) Je mets très souvent en place *des bases de connaissance* dans **OpenWebUI** Cette fonctionnalité permet d'avoir accès rapidement à un **RAG** et si les documents sont de qualités, d'obtenir alors une aide précieuse dans les recherches. Afin que cette aide perdure dans le temps il faut mettre à jours cette base de connaissance. J'ai pris pour habitude d'avoir un flux de ce type ```mermaid flowchart LR chunking["Extraction au format md"] subgraph db["Mise à jours Base de Connaissance"] del["Suppression fichier si déjà existant"] add["upload fichier"] attach["link entre fichier et Base de Connaissance"] del --> add add --> attach end chunking --> del ``` Je propose un petit script qui réalise la partie upload et attachement des fichiers (avec suppression au besoin) en python en utilisant les **APIs** d'*OpenWebUI*. Il faut bien sur avoir un utilisateur qui a les accès à la base de connaissance et une clé api active. ```python """ upload_to_openwebui.py ----------------------- Uploade des fichiers Markdown dans la base de connaissance d'OpenWebUI. Si un fichier du même nom existe déjà dans la KB, il est supprimé avant le nouvel upload. Usage : python upload_to_openwebui.py --kb mykb --files fichier1.md fichier2.md python upload_to_openwebui.py --kb mykb --folder ./mes_fiches/ Configuration : Définir les variables d'environnement : OPENWEBUI_URL ex: http://localhost:3000 OPENWEBUI_API_KEY votre clé API OpenWebUI """ import os import sys import time import argparse import requests from pathlib import Path # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- BASE_URL = os.environ.get("OPENWEBUI_URL", "http://localhost:3000").rstrip("/") API_KEY = os.environ.get("OPENWEBUI_API_KEY", "") KB_NAME = os.environ.get("OPENWEBUI_KB_NAME", "") HEADERS = { "Authorization": f"Bearer {API_KEY}", } ERRORS = [] SUCCESSS = [] REPLACEDS = [] # --------------------------------------------------------------------------- # Parallelism # --------------------------------------------------------------------------- import sys import time import threading, queue class Action(): def __init__(self, func, **kw): if not callable(func): raise ValueError("%s is not callable" % func) self.func = func self.kw = kw def run(self): return self.func(**self.kw) class ParallelismThread(threading.Thread): def __init__(self, para): threading.Thread.__init__(self) self.para = para def run(self): while self.para.qin.qsize() > 0: act = self.para.qin.get() self.para.qout.put(act.run()) class Parallelism(): def __init__(self, thread=1): self.threads = [] for i in range(0, thread): self.threads.append(ParallelismThread(self)) self.qin = queue.Queue() self.qout = queue.Queue() def append(self, func, **kw): self.qin.put(Action(func, **kw)) def run(self): size = self.qin.qsize() for thread in self.threads: thread.start() while self.qout.qsize() < size: time.sleep(0.1) return [self.qout.get() for elt in range(0, size)] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def get_kb_id(kb_name: str) -> str: """Récupère l'ID de la knowledge base par son nom.""" resp = requests.get(f"{BASE_URL}/api/v1/knowledge/", headers=HEADERS) resp.raise_for_status() for kb in resp.json()['items']: if kb.get("name", "").strip().lower() == kb_name.strip().lower(): return kb["id"] raise ValueError(f"Knowledge base '{kb_name}' introuvable. Vérifiez le nom dans OpenWebUI.") def list_kb_files(kb_id: str) -> list[dict]: """Retourne la liste complète des fichiers présents dans la KB (pagination automatique).""" all_files = [] page = 1 limit = 100 # maximum par page, à ajuster selon ce que l'API accepte while True: resp = requests.get( f"{BASE_URL}/api/v1/knowledge/{kb_id}/files", headers={**HEADERS, "Content-Type": "application/json"}, params={"page": page, "limit": limit} ) resp.raise_for_status() data = resp.json() # Récupère les fichiers de cette page items = data.get("items", []) or [] for item in items: del item['data'] all_files.extend(items) # Conditions d'arrêt if len(items) < limit: # Dernière page : moins de résultats que le limit break # Vérifie s'il y a une page suivante via les métadonnées has_next = ( data.get("has_more") # format {"has_more": true} or data.get("next") # format {"next": "url..."} or (data.get("total", 0) > page * limit) # format {"total": 3500} ) if not has_next: break page += 1 print(f"✅ Total fichiers récupérés : {len(all_files)}") return all_files def find_existing_file(kb_files: list[dict], filename: str): """Cherche un fichier par nom dans la liste des fichiers de la KB. Retourne le file_id s'il existe, None sinon.""" for f in kb_files: # Le nom peut être dans f["meta"]["name"] ou f["filename"] meta_name = f.get("meta", {}).get("name", "") file_name = f.get("filename", "") if meta_name == filename or file_name == filename: return f["id"] return None def remove_file_from_kb(kb_id: str, file_id: str, filename: str): """Supprime un fichier de la KB (détachement + suppression du fichier).""" # 1. Détacher le fichier de la KB resp = requests.post( f"{BASE_URL}/api/v1/knowledge/{kb_id}/file/remove", headers={**HEADERS, "Content-Type": "application/json"}, json={"file_id": file_id}, ) if resp.status_code not in (200, 204): print(f" ⚠️ Impossible de détacher '{filename}' (status {resp.status_code})") return # 2. Supprimer le fichier du stockage resp = requests.delete( f"{BASE_URL}/api/v1/files/{file_id}", headers=HEADERS, ) if resp.status_code in (200, 204): print(f" 🗑️ Fichier existant supprimé : {filename}") else: print(f" ⚠️ Détaché mais non supprimé du stockage (status {resp.status_code})") def upload_file(filepath: Path) -> str: """Uploade un fichier vers OpenWebUI et retourne son file_id.""" with open(filepath, "rb") as f: resp = requests.post( f"{BASE_URL}/api/v1/files/", headers=HEADERS, files={"file": (filepath.name, f, "text/markdown")}, ) try: resp.raise_for_status() except requests.exceptions.HTTPError as e: print(f"❌ Erreur HTTP {resp.status_code}") print(" ", resp.text) # ou resp.json() si tu sais que c'est du JSON raise return resp.json()["id"] def attach_file_to_kb(kb_id: str, file_id: str): """Attache un fichier uploadé à la knowledge base.""" resp = requests.post( f"{BASE_URL}/api/v1/knowledge/{kb_id}/file/add", headers={**HEADERS, "Content-Type": "application/json"}, json={"file_id": file_id}, ) try: resp.raise_for_status() except requests.exceptions.HTTPError as e: print(f"❌ Erreur HTTP {resp.status_code}") print(" ", resp.text) # ou resp.json() si tu sais que c'est du JSON raise def wait_file_processed(file_id: str, timeout: float = 20.0, interval: float = 0.5): deadline = time.time() + timeout last_status = None while time.time() < deadline: resp = requests.get(f"{BASE_URL}/api/v1/files/{file_id}", headers=HEADERS) if resp.status_code == 200: data = resp.json() # Le contenu extrait se trouve généralement dans data["data"]["content"] extracted = (data.get("data") or {}).get("content", "") last_status = extracted if extracted and extracted.strip(): return True time.sleep(interval) print(f" ⚠️ Contenu extrait toujours vide après {timeout}s (dernier check: {last_status!r})") return False def upload(filepath, kb_files, kb_id): filename = filepath.name print(f" 📄 {filename}") try: # 1. Suppression si existant existing_id = find_existing_file(kb_files, filename) if existing_id: remove_file_from_kb(kb_id, existing_id, filename) print(f" ✅ {filename} remove") REPLACEDS.append(1) # 2. Upload file_id = upload_file(filepath) print(f" ⬆️ Uploadé (id: {file_id})") # 2bis. Attendre que OpenWebUI ait fini d'extraire le contenu pour pouvoir l'attacher wait_file_processed(file_id) # 3. Attachement à la KB attach_file_to_kb(kb_id, file_id) print(f" 🔗 Attaché à la KB '{KB_NAME}'") SUCCESSS.append(1) except Exception as e: print(f" ❌ Erreur : {e}") ERRORS.append(1) # --------------------------------------------------------------------------- # Fonction principale d'upload # --------------------------------------------------------------------------- def upload_markdown_to_kb(filepaths, threads=10): print(f"🔍 Connexion à {BASE_URL}...") kb_id = get_kb_id(KB_NAME) print(f"✅ Knowledge base '{KB_NAME}' trouvée (id: {kb_id})") print(f"📂 {len(filepaths)} fichier(s) à traiter\n") kb_files = list_kb_files(kb_id) results = {"ok": 0, "error": 0, "replaced": 0} print(len(kb_files)) parallelism = Parallelism(thread=threads) for filepath in filepaths: parallelism.append(upload, filepath=filepath, kb_files=kb_files, kb_id=kb_id) pieces = parallelism.run() # Résumé print("─" * 40) print(f"✅ Succès : {len(SUCCESSS)}") print(f"🔄 Remplacés : {len(REPLACEDS)}") print(f"❌ Erreurs : {len(ERRORS)}") # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def collect_files(path): files = [] folder = Path(path) if not folder.is_dir(): print(f"❌ Dossier introuvable : {path}") sys.exit(1) pattern = "**/*.md" files.extend(sorted(folder.glob(pattern))) return files if __name__ == "__main__": parser = argparse.ArgumentParser(description="Upload Markdown → KB (OpenWebUI)") parser.add_argument("--folder", help="Dossier contenant des fichiers .md") parser.add_argument("--url", help="url OpenWebUI") parser.add_argument("--apikey", help="key api of OpenWebUI") parser.add_argument("--kb", help="name kb of OpenWebUI") parser.add_argument("--threads", help="number of threads (default 10)", default=10, type=int) args = parser.parse_args() if not args.folder: parser.print_help() sys.exit(1) if args.url: BASE_URL= args.url.rstrip("/") if args.kb: KB_NAME= args.kb if args.apikey: API_KEY = args.apikey HEADERS = {"Authorization": f"Bearer {API_KEY}",} filepaths = collect_files(args.folder) if not filepaths: print("❌ Aucun fichier .md trouvé.") sys.exit(1) upload_markdown_to_kb(filepaths, args.threads) ``` les prérequis dans un fichier **requirements.txt** ``` requests==2.34.2 ```