Použití MinIO k vytvoření aplikace pro chat s rozšířenou generací načítání

Často se říká, že ve věku umělé inteligence jsou data vaším příkopem. Za tímto účelem vyžaduje vytvoření aplikace RAG na produkční úrovni vhodnou datovou infrastrukturu pro ukládání, verzi, zpracování, vyhodnocování a dotazování na části dat, které tvoří váš proprietární korpus. Vzhledem k tomu, že MinIO používá k AI přístup na prvním místě na datech, naším výchozím doporučením pro počáteční infrastrukturu pro projekt tohoto typu je nastavení Modern Data Lake (MinIO) a vektorové databáze. Zatímco ostatní pomocné nástroje může být nutné zapojit během cesty, tyto dvě jednotky infrastruktury jsou základní. Budou sloužit jako těžiště pro téměř všechny úkoly, se kterými se následně setkáte při uvedení vaší aplikace RAG do výroby. Ale jste v rébusu. O těchto pojmech LLM a RAG jste již slyšeli, ale kromě toho jste se kvůli neznámému moc neodvážili. Nebylo by ale hezké, kdyby existovala aplikace „Hello World“ nebo standardní aplikace, která vám pomůže začít? Neboj, byl jsem na stejné lodi. V tomto blogu si tedy ukážeme, jak používat MinIO k vytvoření chatovací aplikace založené na Retrieval Augmented Generation (RAG) pomocí komoditního hardwaru. Použijte MinIO k uložení všech dokumentů, zpracovaných kusů a vložení pomocí vektorové databáze. Použijte funkci oznámení o bucketu MinIO ke spuštění událostí při přidávání nebo odebírání dokumentů do bucketu Webhook, který spotřebovává událost a zpracovává dokumenty pomocí Langchain a ukládá metadata a rozdělené dokumenty do segmentu metadat Spouštět události upozornění MinIO bucket pro nově přidané nebo odstraněné blokované dokumenty Webhook, který spotřebovává události a generuje vložení a ukládá je do vektorové databáze (LanceDB), která je uložena v MinIO Použité klíčové nástroje - Object Store pro zachování všech dat MinIO - Serverless open-source vektorová databáze, která přetrvává data v úložišti objektů LanceDB – Chcete-li lokálně spustit LLM a model vkládání (kompatibilní s OpenAI API) Ollama - Rozhraní, jehož prostřednictvím lze komunikovat s aplikací RAG Gradio – Server pro webhooky, který přijímá oznámení o bucketu z MinIO a zpřístupňuje aplikaci Gradio FastAPI - extrahovat užitečný text z našich dokumentů a rozdělit je pro vkládání LangChain & Unstructured Použité modely – Phi-3-128K (3,8B parametrů) LLM – Nomic Embed Text v1.5 ( / 768 Dim, 8K kontext) Embeddings Matryoshka Embeddings Spusťte MinIO Server si můžete stáhnout binární soubor, pokud jej ještě nemáte Zde # Run MinIO detached !minio server ~/dev/data --console-address :9090 & Spustit Ollama Server + stáhnout LLM & Embedding Model Stáhněte si Ollama odtud # Start the Server !ollama serve # Download Phi-3 LLM !ollama pull phi3:3.8b-mini-128k-instruct-q8_0 # Download Nomic Embed Text v1.5 !ollama pull nomic-embed-text:v1.5 # List All the Models !ollama ls Vytvořte základní aplikaci Gradio pomocí FastAPI k testování modelu LLM_MODEL = "phi3:3.8b-mini-128k-instruct-q8_0" EMBEDDING_MODEL = "nomic-embed-text:v1.5" LLM_ENDPOINT = "http://localhost:11434/api/chat" CHAT_API_PATH = "/chat" def llm_chat(user_question, history): history = history or [] user_message = f"**You**: {user_question}" llm_resp = requests.post(LLM_ENDPOINT, json={"model": LLM_MODEL, "keep_alive": "48h", # Keep the model in-memory for 48 hours "messages": [ {"role": "user", "content": user_question } ]}, stream=True) bot_response = "**AI:** " for resp in llm_resp.iter_lines(): json_data = json.loads(resp) bot_response += json_data["message"]["content"] yield bot_response import json import gradio as gr import requests from fastapi import FastAPI, Request, BackgroundTasks from pydantic import BaseModel import uvicorn import nest_asyncio app = FastAPI() with gr.Blocks(gr.themes.Soft()) as demo: gr.Markdown("## RAG with MinIO") ch_interface = gr.ChatInterface(llm_chat, undo_btn=None, clear_btn="Clear") ch_interface.chatbot.show_label = False ch_interface.chatbot.height = 600 demo.queue() if __name__ == "__main__": nest_asyncio.apply() app = gr.mount_gradio_app(app, demo, path=CHAT_API_PATH) uvicorn.run(app, host="0.0.0.0", port=8808) Testovací model vkládání import numpy as np EMBEDDING_ENDPOINT = "http://localhost:11434/api/embeddings" EMBEDDINGS_DIM = 768 def get_embedding(text): resp = requests.post(EMBEDDING_ENDPOINT, json={"model": EMBEDDING_MODEL, "prompt": text}) return np.array(resp.json()["embedding"][:EMBEDDINGS_DIM], dtype=np.float16) ## Test with sample text get_embedding("What is MinIO?") Přehled potrubí příjmu Vytvářejte MiniIO buckety Použijte příkaz mc nebo to proveďte z uživatelského rozhraní custom-corpus - Pro uložení všech dokumentů sklad – pro uložení všech metadat, bloků a vektorových vložení !mc alias set 'myminio' 'http://localhost:9000' 'minioadmin' 'minioadmin' !mc mb myminio/custom-corpus !mc mb myminio/warehouse Vytvořte webhook, který spotřebovává oznámení o segmentu z vlastního korpusového segmentu import json import gradio as gr import requests from fastapi import FastAPI, Request from pydantic import BaseModel import uvicorn import nest_asyncio app = FastAPI() @app.post("/api/v1/document/notification") async def receive_webhook(request: Request): json_data = await request.json() print(json.dumps(json_data, indent=2)) with gr.Blocks(gr.themes.Soft()) as demo: gr.Markdown("## RAG with MinIO") ch_interface = gr.ChatInterface(llm_chat, undo_btn=None, clear_btn="Clear") ch_interface.chatbot.show_label = False demo.queue() if __name__ == "__main__": nest_asyncio.apply() app = gr.mount_gradio_app(app, demo, path=CHAT_API_PATH) uvicorn.run(app, host="0.0.0.0", port=8808) ## Test with sample text get_embedding("What is MinIO?") Vytvořte upozornění na události MinIO a propojte je s vlastním korpusovým bucketem Vytvořit událost Webhooku V konzole přejděte na Události-> Přidat cíl události -> Webhook Vyplňte pole následujícími hodnotami a stiskněte Uložit - doc-webhook Identifikátor – Koncový bod http://localhost:8808/api/v1/document/notification Po zobrazení výzvy klikněte na Restartovat MinIO v horní části ( : K tomu můžete také použít mc) Poznámka Propojte událost webhooku s událostmi vlastního korpusu V konzoli přejděte na Buckets (Administrator) -> custom-corpus -> Events Vyplňte pole následujícími hodnotami a stiskněte Uložit – Z rozevírací nabídky vyberte webhook dokumentu ARN - Zaškrtněte PUT a DELETE Vyberte Události ( : K tomu můžete také použít mc) Poznámka Máme za sebou první nastavení webhooku Nyní otestujte přidáním a odebráním objektu Extrahujte data z dokumentů a bloku Použijeme Langchain a Nestrukturované ke čtení objektu z MinIO a rozdělených dokumentů na více částí from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.document_loaders import S3FileLoader MINIO_ENDPOINT = "http://localhost:9000" MINIO_ACCESS_KEY = "minioadmin" MINIO_SECRET_KEY = "minioadmin" # Split Text from a given document using chunk_size number of characters text_splitter = RecursiveCharacterTextSplitter(chunk_size=1024, chunk_overlap=64, length_function=len) def split_doc_by_chunks(bucket_name, object_key): loader = S3FileLoader(bucket_name, object_key, endpoint_url=MINIO_ENDPOINT, aws_access_key_id=MINIO_ACCESS_KEY, aws_secret_access_key=MINIO_SECRET_KEY) docs = loader.load() doc_splits = text_splitter.split_documents(docs) return doc_splits # test the chunking split_doc_by_chunks("custom-corpus", "The-Enterprise-Object-Store-Feature-Set.pdf") Přidejte do Webhooku logiku Chunking Přidejte logiku bloků do webhooku a uložte metadata a bloky do segmentu skladu import urllib.parse import s3fs METADATA_PREFIX = "metadata" # Using s3fs to save and delete objects from MinIO s3 = s3fs.S3FileSystem() # Split the documents and save the metadata to warehouse bucket def create_object_task(json_data): for record in json_data["Records"]: bucket_name = record["s3"]["bucket"]["name"] object_key = urllib.parse.unquote(record["s3"]["object"]["key"]) print(record["s3"]["bucket"]["name"], record["s3"]["object"]["key"]) doc_splits = split_doc_by_chunks(bucket_name, object_key) for i, chunk in enumerate(doc_splits): source = f"warehouse/{METADATA_PREFIX}/{bucket_name}/{object_key}/chunk_{i:05d}.json" with s3.open(source, "w") as f: f.write(chunk.json()) return "Task completed!" def delete_object_task(json_data): for record in json_data["Records"]: bucket_name = record["s3"]["bucket"]["name"] object_key = urllib.parse.unquote(record["s3"]["object"]["key"]) s3.delete(f"warehouse/{METADATA_PREFIX}/{bucket_name}/{object_key}", recursive=True) return "Task completed!" Aktualizujte server FastAPI pomocí nové logiky import json import gradio as gr import requests from fastapi import FastAPI, Request, BackgroundTasks from pydantic import BaseModel import uvicorn import nest_asyncio app = FastAPI() @app.post("/api/v1/document/notification") async def receive_webhook(request: Request, background_tasks: BackgroundTasks): json_data = await request.json() if json_data["EventName"] == "s3:ObjectCreated:Put": print("New object created!") background_tasks.add_task(create_object_task, json_data) if json_data["EventName"] == "s3:ObjectRemoved:Delete": print("Object deleted!") background_tasks.add_task(delete_object_task, json_data) return {"status": "success"} with gr.Blocks(gr.themes.Soft()) as demo: gr.Markdown("## RAG with MinIO") ch_interface = gr.ChatInterface(llm_chat, undo_btn=None, clear_btn="Clear") ch_interface.chatbot.show_label = False demo.queue() if __name__ == "__main__": nest_asyncio.apply() app = gr.mount_gradio_app(app, demo, path=CHAT_API_PATH) uvicorn.run(app, host="0.0.0.0", port=8808) Přidejte nový webhook ke zpracování metadat/dílů dokumentu Nyní, když funguje první webhook, dalším krokem je získat všechny části s metadaty, vygenerovat vložení a uložit je do vektorové databáze import json import gradio as gr import requests from fastapi import FastAPI, Request, BackgroundTasks from pydantic import BaseModel import uvicorn import nest_asyncio app = FastAPI() @app.post("/api/v1/metadata/notification") async def receive_metadata_webhook(request: Request, background_tasks: BackgroundTasks): json_data = await request.json() print(json.dumps(json_data, indent=2)) @app.post("/api/v1/document/notification") async def receive_webhook(request: Request, background_tasks: BackgroundTasks): json_data = await request.json() if json_data["EventName"] == "s3:ObjectCreated:Put": print("New object created!") background_tasks.add_task(create_object_task, json_data) if json_data["EventName"] == "s3:ObjectRemoved:Delete": print("Object deleted!") background_tasks.add_task(delete_object_task, json_data) return {"status": "success"} with gr.Blocks(gr.themes.Soft()) as demo: gr.Markdown("## RAG with MinIO") ch_interface = gr.ChatInterface(llm_chat, undo_btn=None, clear_btn="Clear") ch_interface.chatbot.show_label = False demo.queue() if __name__ == "__main__": nest_asyncio.apply() app = gr.mount_gradio_app(app, demo, path=CHAT_API_PATH) uvicorn.run(app, host="0.0.0.0", port=8808) Vytvořte oznámení o událostech MinIO a propojte je s kbelíkem skladu Vytvořit událost Webhooku V konzole přejděte na Události-> Přidat cíl události -> Webhook Vyplňte pole následujícími hodnotami a stiskněte Uložit – metadata-webhook Identifikátor – Koncový bod http://localhost:8808/api/v1/metadata/notification Po zobrazení výzvy klikněte na Restartovat MinIO v horní části ( : K tomu můžete také použít mc) Poznámka Propojte událost webhooku s událostmi vlastního korpusu V konzoli přejděte do Buckets (Administrátor) -> sklad -> Události Vyplňte pole následujícími hodnotami a stiskněte Uložit – Z rozbalovací nabídky vyberte webhook metadat ARN - metadata/ Prefix – .json Přípona - Zaškrtněte PUT a DELETE Vyberte Události ( : K tomu můžete také použít mc) Poznámka Máme za sebou první nastavení webhooku Nyní otestujte přidáním a odebráním objektu ve vlastním korpusu a zjistěte, zda se tento webhook spustí Vytvořte vektorovou databázi LanceDB v MinIO Nyní, když máme funkční základní webhook, pojďme nastavit vektorovou databázi lanceDB v kbelíku skladu MinIO, do kterého uložíme všechna vložení a další pole metadat. import os import lancedb # Set these environment variables for the lanceDB to connect to MinIO os.environ["AWS_DEFAULT_REGION"] = "us-east-1" os.environ["AWS_ACCESS_KEY_ID"] = MINIO_ACCESS_KEY os.environ["AWS_SECRET_ACCESS_KEY"] = MINIO_SECRET_KEY os.environ["AWS_ENDPOINT"] = MINIO_ENDPOINT os.environ["ALLOW_HTTP"] = "True" db = lancedb.connect("s3://warehouse/v-db/") # list existing tables db.table_names() # Create a new table with pydantic schema from lancedb.pydantic import LanceModel, Vector import pyarrow as pa DOCS_TABLE = "docs" EMBEDDINGS_DIM = 768 table = None class DocsModel(LanceModel): parent_source: str # Actual object/document source source: str # Chunk/Metadata source text: str # Chunked text vector: Vector(EMBEDDINGS_DIM, pa.float16()) # Vector to be stored def get_or_create_table(): global table if table is None and DOCS_TABLE not in list(db.table_names()): return db.create_table(DOCS_TABLE, schema=DocsModel) if table is None: table = db.open_table(DOCS_TABLE) return table # Check if that worked get_or_create_table() # list existing tables db.table_names() Přidejte Ukládání/odebírání dat z lanceDB do metadata-webhook import multiprocessing EMBEDDING_DOCUMENT_PREFIX = "search_document" # Add queue that keeps the processed meteadata in memory add_data_queue = multiprocessing.Queue() delete_data_queue = multiprocessing.Queue() def create_metadata_task(json_data): for record in json_data["Records"]: bucket_name = record["s3"]["bucket"]["name"] object_key = urllib.parse.unquote(record["s3"]["object"]["key"]) print(bucket_name, object_key) with s3.open(f"{bucket_name}/{object_key}", "r") as f: data = f.read() chunk_json = json.loads(data) embeddings = get_embedding(f"{EMBEDDING_DOCUMENT_PREFIX}: {chunk_json['page_content']}") add_data_queue.put({ "text": chunk_json["page_content"], "parent_source": chunk_json.get("metadata", "").get("source", ""), "source": f"{bucket_name}/{object_key}", "vector": embeddings }) return "Metadata Create Task Completed!" def delete_metadata_task(json_data): for record in json_data["Records"]: bucket_name = record["s3"]["bucket"]["name"] object_key = urllib.parse.unquote(record["s3"]["object"]["key"]) delete_data_queue.put(f"{bucket_name}/{object_key}") return "Metadata Delete Task completed!" Přidejte plánovač, který zpracovává data z front from apscheduler.schedulers.background import BackgroundScheduler import pandas as pd def add_vector_job(): data = [] table = get_or_create_table() while not add_data_queue.empty(): item = add_data_queue.get() data.append(item) if len(data) > 0: df = pd.DataFrame(data) table.add(df) table.compact_files() print(len(table.to_pandas())) def delete_vector_job(): table = get_or_create_table() source_data = [] while not delete_data_queue.empty(): item = delete_data_queue.get() source_data.append(item) if len(source_data) > 0: filter_data = ", ".join([f'"{d}"' for d in source_data]) table.delete(f'source IN ({filter_data})') table.compact_files() table.cleanup_old_versions() print(len(table.to_pandas())) scheduler = BackgroundScheduler() scheduler.add_job(add_vector_job, 'interval', seconds=10) scheduler.add_job(delete_vector_job, 'interval', seconds=10) Aktualizujte FastAPI pomocí změn Vector Embedding Changes import json import gradio as gr import requests from fastapi import FastAPI, Request, BackgroundTasks from pydantic import BaseModel import uvicorn import nest_asyncio app = FastAPI() @app.on_event("startup") async def startup_event(): get_or_create_table() if not scheduler.running: scheduler.start() @app.on_event("shutdown") async def shutdown_event(): scheduler.shutdown() @app.post("/api/v1/metadata/notification") async def receive_metadata_webhook(request: Request, background_tasks: BackgroundTasks): json_data = await request.json() if json_data["EventName"] == "s3:ObjectCreated:Put": print("New Metadata created!") background_tasks.add_task(create_metadata_task, json_data) if json_data["EventName"] == "s3:ObjectRemoved:Delete": print("Metadata deleted!") background_tasks.add_task(delete_metadata_task, json_data) return {"status": "success"} @app.post("/api/v1/document/notification") async def receive_webhook(request: Request, background_tasks: BackgroundTasks): json_data = await request.json() if json_data["EventName"] == "s3:ObjectCreated:Put": print("New object created!") background_tasks.add_task(create_object_task, json_data) if json_data["EventName"] == "s3:ObjectRemoved:Delete": print("Object deleted!") background_tasks.add_task(delete_object_task, json_data) return {"status": "success"} with gr.Blocks(gr.themes.Soft()) as demo: gr.Markdown("## RAG with MinIO") ch_interface = gr.ChatInterface(llm_chat, undo_btn=None, clear_btn="Clear") ch_interface.chatbot.show_label = False ch_interface.chatbot.height = 600 demo.queue() if __name__ == "__main__": nest_asyncio.apply() app = gr.mount_gradio_app(app, demo, path=CHAT_API_PATH) uvicorn.run(app, host="0.0.0.0", port=8808) Nyní, když máme funkční potrubí Ingestion, pojďme integrovat finální potrubí RAG. Přidejte možnost vektorového vyhledávání Nyní, když máme dokument přijatý do lanceDB, přidejte možnost vyhledávání EMBEDDING_QUERY_PREFIX = "search_query" def search(query, limit=5): query_embedding = get_embedding(f"{EMBEDDING_QUERY_PREFIX}: {query}") res = get_or_create_table().search(query_embedding).metric("cosine").limit(limit) return res # Lets test to see if it works res = search("What is MinIO Enterprise Object Store Lite?") res.to_list() Vyzvěte LLM, aby použila příslušné dokumenty RAG_PROMPT = """ DOCUMENT: {documents} QUESTION: {user_question} INSTRUCTIONS: Answer in detail the user's QUESTION using the DOCUMENT text above. Keep your answer ground in the facts of the DOCUMENT. Do not use sentence like "The document states" citing the document. If the DOCUMENT doesn't contain the facts to answer the QUESTION only Respond with "Sorry! I Don't know" """ context_df = [] def llm_chat(user_question, history): history = history or [] global context_df # Search for relevant document chunks res = search(user_question) documents = " ".join([d["text"].strip() for d in res.to_list()]) # Pass the chunks to LLM for grounded response llm_resp = requests.post(LLM_ENDPOINT, json={"model": LLM_MODEL, "messages": [ {"role": "user", "content": RAG_PROMPT.format(user_question=user_question, documents=documents) } ], "options": { # "temperature": 0, "top_p": 0.90, }}, stream=True) bot_response = "**AI:** " for resp in llm_resp.iter_lines(): json_data = json.loads(resp) bot_response += json_data["message"]["content"] yield bot_response context_df = res.to_pandas() context_df = context_df.drop(columns=['source', 'vector']) def clear_events(): global context_df context_df = [] return context_df Aktualizujte FastAPI Chat Endpoint pro použití RAG import json import gradio as gr import requests from fastapi import FastAPI, Request, BackgroundTasks from pydantic import BaseModel import uvicorn import nest_asyncio app = FastAPI() @app.on_event("startup") async def startup_event(): get_or_create_table() if not scheduler.running: scheduler.start() @app.on_event("shutdown") async def shutdown_event(): scheduler.shutdown() @app.post("/api/v1/metadata/notification") async def receive_metadata_webhook(request: Request, background_tasks: BackgroundTasks): json_data = await request.json() if json_data["EventName"] == "s3:ObjectCreated:Put": print("New Metadata created!") background_tasks.add_task(create_metadata_task, json_data) if json_data["EventName"] == "s3:ObjectRemoved:Delete": print("Metadata deleted!") background_tasks.add_task(delete_metadata_task, json_data) return {"status": "success"} @app.post("/api/v1/document/notification") async def receive_webhook(request: Request, background_tasks: BackgroundTasks): json_data = await request.json() if json_data["EventName"] == "s3:ObjectCreated:Put": print("New object created!") background_tasks.add_task(create_object_task, json_data) if json_data["EventName"] == "s3:ObjectRemoved:Delete": print("Object deleted!") background_tasks.add_task(delete_object_task, json_data) return {"status": "success"} with gr.Blocks(gr.themes.Soft()) as demo: gr.Markdown("## RAG with MinIO") ch_interface = gr.ChatInterface(llm_chat, undo_btn=None, clear_btn="Clear") ch_interface.chatbot.show_label = False ch_interface.chatbot.height = 600 gr.Markdown("### Context Supplied") context_dataframe = gr.DataFrame(headers=["parent_source", "text", "_distance"], wrap=True) ch_interface.clear_btn.click(clear_events, [], context_dataframe) @gr.on(ch_interface.output_components, inputs=[ch_interface.chatbot], outputs=[context_dataframe]) def update_chat_context_df(text): global context_df if context_df is not None: return context_df return "" demo.queue() if __name__ == "__main__": nest_asyncio.apply() app = gr.mount_gradio_app(app, demo, path=CHAT_API_PATH) uvicorn.run(app, host="0.0.0.0", port=8808) Podařilo se vám projít a implementovat chat založený na RAG s MinIO jako backendem datového jezera? V blízké budoucnosti uspořádáme webový seminář na stejné téma, kde vám poskytneme živé demo, když vytváříme tuto chatovací aplikaci založenou na RAG. RAGs-R-Us Jako vývojář zaměřený na integraci AI ve společnosti MinIO neustále zkoumám, jak lze naše nástroje hladce integrovat do moderních architektur AI, aby se zvýšila efektivita a škálovatelnost. V tomto článku jsme vám ukázali, jak integrovat MinIO s Retrieval-Augmented Generation (RAG) a vytvořit chatovací aplikaci. Toto je jen špička ledovce, která vám poskytne podporu ve vaší snaze vytvořit více unikátních použitých pouzder pro RAG a MinIO. Nyní k tomu máte stavební kameny. Pojďme na to! Pokud máte nějaké dotazy ohledně integrace MinIO RAG, určitě se na nás obraťte ! Slack