Post jest kontynuacją 2-giej części serii.
Nota wstępu:
Z racji że zaczynamy wchodzić coraz głębiej, podejście czysto funkcyjne przestaje zdawać egzamin. Zamiast tego, wdrożyłem rozwiązanie oparte na programowaniu obiektowym. Jeśli chodzi o użyte funkcje, nic praktycznie się nie zmienia. Jedynie odpowiednie metody zostały rozdzielone oraz pogrupowane per klasy. Same ich wywołania przez to są dużo bardziej przyjemniejsze.
Dostęp do pełnego kodu przepisanego na obiektowe podejście znajdziesz pod tym linkiem.
Zacznijmy od zdefiniowania naszej funkcji której celem jest:
- Utworzenie zmiennej przechowującej całą historię naszej konwersacji,
- Pobranie za pomocą metody collection.query(), dokumentów znajdujących się, (wektorowo) najbliżej naszego zapytania,
- Przesłaniem naszego zapytania do funkcji search(),
- Zwróceniem odpowiedzi wygenerowanej przez LLM (framework gradio już sam zajmie się odpowiednim jej wyrenderowaniem),
def chat_fn(
message: str,
history: Optional[List[Tuple[str, str]]],
search_provider: Literal["openai", "local"],
n_results: int,
) -> Tuple[List[Tuple[str, str]], List[Tuple[str, str]]]:
history = history or []
results = collection.query(
query_embeddings=embed_fn([message]), n_results=int(n_results)
)
if search_provider == "openai":
search_model = "gpt-4o-mini"
else:
search_model = "llama3.1:8b"
answer = search(message, results, search_provider, search_model, args.verbose)
history.append((message, answer))
return history, history
*Z racji że w tym przypadku chcemy dodać (sterowaną z gui), obsługę różnych providerów. Musimy sami wyliczyć z jakiego modelu chcemy skorzystać.
Możesz również dołączyć ostatnie wiadomości do przesłanego promptu (aby np. pamiętać z o czym jest rozmowa i jakie fakty zostały ustalone), korzystając z tablicy znajdującej się pod zmienną history.
if history:
previous_messages = "\n".join(
[
f" User: {q.strip()}\n Assistant: {a.strip()}"
for q, a in history[-2:]
]
)
full_message = f"{message}\n\n History:\n{previous_messages}"
else:
full_message = f"User: {message}"
answer = search(full_message, results, search_provider, search_model, args.verbose)
Teraz zajmiemy się stworzeniem obiektu gradio, który utworzy za nas stronę na której przeprowadzimy konwersację z naszym nowym botem. W tej sytuacji rozszerzamy interfejs użytkownika o przyciski, suwak oraz teksty.
import gradio as gr
with gr.Blocks() as my_rag:
gr.Markdown("## Chat RAG + LLM")
chatbot = gr.Chatbot(label="KamDev.pl")
msg = gr.Textbox(label="Enter message", placeholder="Ask question...")
send_btn = gr.Button("Send")
state = gr.State([])
with gr.Row():
search_provider = gr.Dropdown(
choices=["openai", "local"], value="local", label="Search Provider"
)
n_results = gr.Slider(
minimum=1,
maximum=10,
step=1,
value=3,
label="Set a limit on documents returned by ChromaDB.",
)
Następnie pod przycisk "Send", podpinamy naszą funkcję z początku tutoriala oraz przekazujemy jej wybrane przez nas parametry (które ustawiliśmy przez dropdown oraz slider).
send_btn.click(
fn=chat_fn,
inputs=[msg, state, search_provider, n_results],
outputs=[chatbot, state],
)
Na samym końcu pozostało nam tylko wywołać metodę renderującą nasze GUI.
my_rag.launch()
Wystarczy teraz uruchomić adres http://127.0.0.1:7860 w przeglądarce.
Ważna uwaga:
Dotychczas korzystaliśmy z modeli tworzących embedingi oraz wnioskujących, od tych samych dostawców. Nie musimy jednak się ograniczać. Możemy tworzyć lokalnie embedingi, a następnie wnioskować już zupełnie innymi modelami np z chmury. Ważne aby funkcja przeszukująca bazę jak i tworzaca w niej wektory, korzystała już z tego samego modelu.
Po pobraniu danych z ChromaDB, operujemy już czystym tekstem, który możemy już wedle uznania innymi modelami przetwarzać.
Pełny kod GUI
from typing import List, Literal, Optional, Tuple
import gradio as gr
from chromadb import PersistentClient
from lib.embeding import embeder_loader
from lib.parser import parse_args
from lib.search import search
def chat_fn(
message: str,
history: Optional[List[Tuple[str, str]]],
search_provider: Literal["openai", "local"],
n_results: int,
) -> Tuple[List[Tuple[str, str]], List[Tuple[str, str]]]:
"""
Handles a single chat interaction.
Args:
message (str): The user's message.
history (Optional[List[Tuple[str, str]]]): Conversation history as list of (user, assistant) message pairs.
search_provider (Literal["openai", "local"]): The search backend to use.
n_results (int): Number of top documents to retrieve from ChromaDB.
Returns:
Tuple[List[Tuple[str, str]], List[Tuple[str, str]]]:
- First element: messages to display in the chat UI (usually full or truncated history).
- Second element: updated history to store in state.
"""
history = history or []
results = collection.query(
query_embeddings=embed_fn([message]), n_results=int(n_results)
)
if search_provider == "openai":
search_model = "gpt-4o-mini"
else:
search_model = "llama3.1:8b"
if history:
previous_messages = "\n".join(
[
f" User: {q.strip()}\n Assistant: {a.strip()}"
for q, a in history[-2:]
]
)
full_message = f"{message}\n\n History:\n{previous_messages}"
else:
full_message = f"User: {message}"
answer = search(full_message, results, search_provider, search_model, args.verbose)
history.append((message, answer))
return history, history
with gr.Blocks() as my_rag:
gr.Markdown("## Chat RAG + LLM")
chatbot = gr.Chatbot(label="KamDev.pl")
msg = gr.Textbox(label="Enter message", placeholder="Ask question...")
send_btn = gr.Button("Send")
state = gr.State([])
with gr.Row():
search_provider = gr.Dropdown(
choices=["openai", "local"], value="local", label="Search Provider"
)
n_results = gr.Slider(
minimum=1,
maximum=10,
step=1,
value=3,
label="Set a limit on documents returned by ChromaDB.",
)
send_btn.click(
fn=chat_fn,
inputs=[msg, state, search_provider, n_results],
outputs=[chatbot, state],
)
if __name__ == "__main__":
args = parse_args()
collection = PersistentClient(path=args.db_location).get_or_create_collection(
"documents"
)
embed_fn = embeder_loader(args)
my_rag.launch()
W 4-tej części, sprawimy że nasz RAG będzie nie tylko operował na danych które mu dostarczymy, ale i również na tych które sam zdobędzie scrapując strony www.
Komentarze