Unknown user: /* owui-pakai.py */

2025-09-02T01:52:47Z

owui-pakai.py

← Older revision		Revision as of 01:52, 2 September 2025
Line 164:		Line 164:


	== owui-pakai~~.py~~ ==		== owui-pakai ==

	# 0) Set variabel (opsional)		# 0) Set variabel (opsional)

Unknown user: Created page with "== owui_upload_kb.py == #!/usr/bin/env python3 # -- coding: utf-8 -- """ Upload PDF ke Open WebUI Knowledge via API (tanpa GUI). Fitur: - Buat knowledge collection..."

2025-09-02T01:50:45Z

Created page with "== owui_upload_kb.py == #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Upload PDF ke Open WebUI Knowledge via API (tanpa GUI). Fitur: - Buat knowledge collection..."

New page

== owui_upload_kb.py ==

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Upload PDF ke Open WebUI Knowledge via API (tanpa GUI).

Fitur:
- Buat knowledge collection (jika belum ada)
- Upload banyak PDF (folder / list path)
- Tambah setiap file ke collection
- Anti-duplikat sederhana (skip jika server balas "Duplicate content")
- Uji query RAG terhadap collection (opsional)

Dok:
- Upload file: POST /api/v1/files/
- Tambah ke knowledge: POST /api/v1/knowledge/{id}/file/add
- Buat knowledge: POST /api/v1/knowledge/create
"""

import os
import sys
import time
import json
import glob
import argparse
import requests
from typing import List, Optional

def api_headers(api_key: str, accept_json: bool = True):
h = {"Authorization": f"Bearer {api_key}"}
if accept_json:
h["Accept"] = "application/json"
return h

def create_knowledge(base_url: str, api_key: str, name: str, description: str = "") -> str:
url = f"{base_url}/api/v1/knowledge/create"
payload = {"name": name, "description": description}
r = requests.post(url, headers={**api_headers(api_key), "Content-Type": "application/json"}, json=payload, timeout=120)
r.raise_for_status()
data = r.json()
kb_id = data.get("id")
if not kb_id:
raise RuntimeError(f"Gagal membuat knowledge: {data}")
return kb_id

def upload_file(base_url: str, api_key: str, file_path: str) -> Optional[str]:
url = f"{base_url}/api/v1/files/"
with open(file_path, "rb") as f:
files = {"file": (os.path.basename(file_path), f, "application/pdf")}
r = requests.post(url, headers=api_headers(api_key), files=files, timeout=600)
if r.status_code >= 400:
# Banyak kasus gagal karena file khusus, PDF rusak, dsb.
print(f"[WARN] Upload gagal: {file_path} -> {r.status_code} {r.text}")
return None
data = r.json()
return data.get("id") # file_id

def add_file_to_knowledge(base_url: str, api_key: str, knowledge_id: str, file_id: str) -> bool:
url = f"{base_url}/api/v1/knowledge/{knowledge_id}/file/add"
payload = {"file_id": file_id}
r = requests.post(url, headers={**api_headers(api_key), "Content-Type": "application/json"}, json=payload, timeout=120)
if r.status_code == 400 and "Duplicate content" in r.text:
# Open WebUI bisa mengembalikan 400 duplicate bila konten sama sudah ada
print(f"[INFO] Duplikat terdeteksi, skip file_id={file_id}")
return False
r.raise_for_status()
return True

def find_or_create_kb(base_url: str, api_key: str, kb_name: str, kb_description: str = "") -> str:
"""
Jika Anda sudah tahu UUID knowledge, langsung pakai.
Kalau belum, cara termudah adalah buat baru dengan nama yang diinginkan.
"""
print(f"[STEP] Membuat knowledge '{kb_name}'")
return create_knowledge(base_url, api_key, kb_name, kb_description)

def collect_pdf_paths(input_path: str) -> List[str]:
if os.path.isdir(input_path):
# Ambil semua PDF di folder (tanpa rekursif). Ubah ke **/*.pdf jika ingin rekursif.
return sorted(glob.glob(os.path.join(input_path, "*.pdf")))
elif os.path.isfile(input_path):
return [input_path]
else:
raise FileNotFoundError(f"Path tidak ditemukan: {input_path}")

def rag_test_query(base_url: str, api_key: str, model: str, kb_id: str, user_query: str) -> str:
"""
Gunakan chat completions + files:[{type:'collection', id:kb_id}] untuk uji RAG.
"""
url = f"{base_url}/api/chat/completions"
payload = {
"model": model,
"messages": [{"role": "user", "content": user_query}],
"files": [{"type": "collection", "id": kb_id}],
}
r = requests.post(url, headers={**api_headers(api_key), "Content-Type": "application/json"}, json=payload, timeout=600)
r.raise_for_status()
data = r.json()
# Bentuk respons mengikuti OpenAI-compatible schema.
try:
return data["choices"][0]["message"]["content"]
except Exception:
return json.dumps(data, ensure_ascii=False, indent=2)

def main():
p = argparse.ArgumentParser(description="Upload PDF ke Open WebUI Knowledge (tanpa GUI).")
p.add_argument("--base-url", required=True, help="Contoh: http://localhost:3000")
p.add_argument("--api-key", required=True, help="API Key dari Settings > Account")
p.add_argument("--kb-name", required=True, help="Nama knowledge (collection) yang akan dibuat")
p.add_argument("--kb-desc", default="", help="Deskripsi knowledge")
p.add_argument("--input", required=True, help="Path ke file PDF atau folder berisi PDF")
p.add_argument("--model", default="llama3.1", help="Nama model untuk uji RAG (opsional)")
p.add_argument("--test-query", default="", help="Jika diisi, lakukan uji query RAG ke collection")
p.add_argument("--sleep-after-upload", type=int, default=3, help="Delay (detik) antar upload untuk memberi waktu proses embedding")
args = p.parse_args()

base_url = args.base_url.rstrip("/")
api_key = args.api_key

# 1) Buat knowledge
kb_id = find_or_create_kb(base_url, api_key, args.kb_name, args.kb_desc)
print(f"[OK] Knowledge dibuat: {kb_id}")

# 2) Kumpulkan PDF
pdfs = collect_pdf_paths(args.input)
if not pdfs:
print("[WARN] Tidak ada PDF ditemukan.")
sys.exit(0)
print(f"[STEP] Menemukan {len(pdfs)} file PDF")

# 3) Upload + tambahkan ke knowledge
uploaded = 0
for path in pdfs:
print(f"[STEP] Upload: {path}")
file_id = upload_file(base_url, api_key, path)
if not file_id:
continue
print(f"[OK] File terupload, file_id={file_id} -> tambah ke knowledge")
try:
add_file_to_knowledge(base_url, api_key, kb_id, file_id)
uploaded += 1
except requests.HTTPError as e:
print(f"[ERR] Gagal tambah ke knowledge: {e.response.status_code} {e.response.text}")
# beri jeda kecil agar proses embedding tidak numpuk (praktis untuk CPU-only)
time.sleep(args.sleep_after_upload)

print(f"[DONE] Selesai. Total file berhasil diproses: {uploaded}/{len(pdfs)}")

# 4) Uji RAG (opsional)
if args.test_query:
print(f"[TEST] Jalankan uji RAG model={args.model}")
answer = rag_test_query(base_url, api_key, args.model, kb_id, args.test_query)
print("\n=== JAWABAN RAG ===\n")
print(answer)
print("\n===================\n")

if __name__ == "__main__":
main()

== owui-pakai.py ==

# 0) Set variabel (opsional)
export OWUI_URL="http://localhost:3000"
export OWUI_KEY="sk-xxxxxx"

# 1) Jalankan: buat KB + upload semua PDF di folder ./dokumen
python3 owui_upload_kb.py \
--base-url "$OWUI_URL" \
--api-key "$OWUI_KEY" \
--kb-name "Dokumen Kampus" \
--kb-desc "Koleksi PDF Peraturan & Panduan" \
--input "./dokumen" \
--model "llama3.1" \
--test-query "Ringkas poin penting dari semua dokumen tentang akreditasi BAN-PT."

# Atau untuk satu file saja:
python3 owui_upload_kb.py \
--base-url "$OWUI_URL" \
--api-key "$OWUI_KEY" \
--kb-name "RIP ITTS 2025-2050" \
--input "./RIP-ITTS-2025-2050.pdf" \
--test-query "Apa visi utama dokumen ini?"

OpenWebUI: python knowledge PDF CLI API upload - Revision history

Unknown user: /* owui-pakai.py */

Unknown user: Created page with "== owui_upload_kb.py == #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Upload PDF ke Open WebUI Knowledge via API (tanpa GUI). Fitur: - Buat knowledge collection..."

Unknown user: Created page with "== owui_upload_kb.py == #!/usr/bin/env python3 # -- coding: utf-8 -- """ Upload PDF ke Open WebUI Knowledge via API (tanpa GUI). Fitur: - Buat knowledge collection..."