Rozmawiaj z dowolnym repozytorium GitHub: Streamlit, LlamaIndex, Groq oraz lokalne embeddingi
Stwórz chatbot RAG w oparciu o repozytorium na GitHubie, obejmujący pobieranie danych z gitingest, dzielenie treści na fragmenty w formacie Markdown, wykorzystanie zapisanych modeli, prompt pozwalający przyznać się do braku wiedzy oraz strumieniowanie asynchroniczne w Streamlit.
Źródła: Building a RAG Chatbot That Lets You Chat With Any GitHub Repository (Streamlit + LlamaIndex + Groq)
Osiągnięcie orientacji w nieznanej bazie kodu zwykle zajmuje godziny: przeglądanie plików, domyślanie się architektury, ponowne czytanie tych samych funkcji. Szybszą metodą jest mała aplikacja, do której wprowadza się adres GitHuba, po czym pojawiają się pytania dotyczące kodu, a odpowiedzi są przekazywane w miarę ich generowania. Poniżej znajduje się pełna implementacja: sposób łączenia modułów, to, co robi silnik RAG podczas wyszukiwania, technika wątkowania umożliwiająca płynną pracę biblioteki blokującej wewnątrz Streamlit, oraz sformułowanie promptu zapobiegające tworzeniu przez model fałszywych odpowiedzi.
Czym zajmuje się aplikacja
Jest to aplikacja Streamlit o prostym schemacie działania:
- Wpisuje się adres repozytorium na GitHubie.
- Aplikacja pobiera i indeksuje to repozytorium.
- Pytania zadaje się poprzez interfejs czatu.
To klasyczny wzorzec generowania wzbogaconego o wyszukiwanie, stosowany do kodu źródłowego, a nie do dokumentów prozatorskich. Stack składa się z Streamlit jako interfejsu, LlamaIndex do indeksowania i wyszukiwania, Groq jako modelu językowego oraz embeddingów HuggingFace uruchamianych lokalnie, przy czym gitingest zajmuje się pobieraniem plików.
Struktura projektu
.
├── app.py # Entry point
├── streamlit_app.py # UI orchestrator
├── repository_service.py # Repo download + document loading
├── rag_engine.py # RAG engine: LLM, embeddings, index, prompt
├── response_streamer.py # Sync-to-async streaming bridge
├── app_state.py # Session state management
└── .env # API keys (not committed)
Ścieżka danych:
User enters GitHub URL
│
▼
gitingest downloads & flattens repo → Markdown
│
▼
LlamaIndex splits Markdown into chunks (MarkdownNodeParser)
│
▼
HuggingFace embedding model vectorizes chunks
│
▼
VectorStoreIndex built in memory
│
▼
User asks a question
│
▼
Relevant chunks retrieved
│
▼
Groq LLM answers using only retrieved context
│
▼
Answer streams into the Streamlit chat
Krok 1: bardzo mały punkt wejścia
Wszystko zaczyna się w pliku app.py, który jest celowo utrzymywany w małej formie. Jego jedynymi zadaniami jest załadujenie zmiennych środowiskowych i przekazanie kontroli klasie aplikacji:
from streamlit_app import GitHubRAGApp
from dotenv import load_dotenv
if __name__ == "__main__":
load_dotenv()
GitHubRAGApp().run()
load_dotenv() odczytuje lokalny plik .env i udostępnia jego wartości jako zmienne środowiskowe, dzięki czemu dane poufne nie znajdują się w kodzie źródłowym:
GROQ_API_KEY=your_groq_api_key_here
Zacznij od:
streamlit run app.py
Zachowanie tego pliku w minimalnej postaci ma większe znaczenie, niż się wydaje. Ponieważ to jedyny plik, który faktycznie uruchamia Streamlit, wszystkie pozostałe moduły mogą być testowane jednostkowo bez konieczności używania środowiska działania Streamlit.
Krok 2: koordynator
GitHubRAGApp w pliku streamlit_app.py pełni rolę koordynatora. Zarządza trzema współpracownikami i łączy interfejs z nimi:
class GitHubRAGApp:
def __init__(self) -> None:
self.repository_service = GitHubRepositoryService()
self.engine_factory = RAGEngineFactory()
self.streamer = AsyncResponseStreamer()
def run(self) -> None:
AppState.initialize()
self.render_sidebar()
self.render_header()
self.render_chat_history()
self.handle_chat_input()
def render_sidebar(self) -> None:
with st.sidebar:
st.header("Add your GitHub repository")
github_url = st.text_input(
"GitHub repository URL",
value=st.session_state.github_url,
placeholder="https://github.com/user/repository",
)
st.session_state.github_url = github_url
if st.button("Load Repository", use_container_width=True):
self.load_repository(github_url)
if st.session_state.repositories:
st.success("Repository ready for chat.")
# Loading a repository is the more interesting path. When the user clicks “Load Repository,” the app:
def load_repository(self, github_url: str) -> None:
if not github_url.strip():
st.warning("Enter a GitHub repository URL first.")
return
repository_key = AppState.repository_key(github_url)
if repository_key in st.session_state.repositories:
st.success("Repository already loaded.")
return
status = st.status("Preparing repository...", expanded=True)
try:
status.write("Downloading and extracting repository content...")
content = self.run_async(self.repository_service.ingest_repository, github_url)
status.write("Parsing repository documents...")
documents = self.run_async(self.repository_service.load_documents, content)
status.write("Creating chunks and generating embeddings...")
query_engine = self.run_async(self.engine_factory.build, documents)
st.session_state.repositories[repository_key] = query_engine
status.update(label="Repository is ready for chat.", state="complete", expanded=False)
except Exception as error:
status.update(label="Repository loading failed.", state="error", expanded=True)
st.error(f"Could not process repository: {error}")
Kilka szczegółów zasługuje na uwagę:
st.status(..., expanded=True)pokazuje aktualny log postępów zamiast ikony spinującej. Proces składa się z trzech wolnych etapów: pobierania, analizy i wstawiania, więc możliwość widzenia postępów znacznie poprawia doświadczenie użytkownika.- Proces ładowania jest idempotentny. Jeśli klucz haszowego repozytorium znajduje się już w
st.session_state.repositories, aplikacja przechodzi bezpośrednio do już załadowanych danych, pomijając etapy pobierania i wstawiania. Ten cache sprawia, że powtarzane zapytania oraz przypadkowe podwójne kliknięcia są mało kosztowne. - Każdy wolny krok przechodzi przez
run_async– małą, ale ważną funkcję pomocniczą:
@staticmethod
def run_async(function: Any, *args: Any) -> Any:
async def execute() -> Any:
return await asyncio.to_thread(function, *args)
return asyncio.run(execute())
ingest_repository, load_documents oraz build to funkcje blokujące, synchroniczne, które zajmują się pracą sieciową, obsługą plików oraz intensywnym wykorzystaniem procesora do embedowania danych. Funkcja run_async przenosi każdą z tych funkcji na osobny wątek roboczy za pomocą asyncio.to_thread, dzięki czemu wolne ściąganie nie blokuje jednowątkowego pętla ponownego uruchamiania w Streamlit, a st.status nadal jest aktualizowany.
Wprowadzanie tekstu w czacie odbywa się zgodnie ze standardowym wzorcem Streamlit, przy czym generowanie odpowiedzi jest przekazywane do stream_response:
def handle_chat_input(self) -> None:
prompt = st.chat_input("Ask a question about the repository")
if not prompt:
return
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.markdown(prompt)
with st.chat_message("assistant"):
response_placeholder = st.empty()
response = self.stream_response(prompt, response_placeholder)
st.session_state.messages.append({"role": "assistant", "content": response})
Krok 3: pobieranie repozytorium
Zamiast tworzyć własnego narzędzia do przeszukiwania GitHuba, aplikacja korzysta z gitingest, które pobiera repozytorium i przekształca je w jedną reprezentację w formacie Markdown, zachowując granice plików, strukturę katalogów oraz ich zawartość. Logika znajduje się w pliku repository_service.py:
class GitHubRepositoryService:
"""Download and prepare GitHub repository content."""
CONTENT_FILE = "content.md"
def ingest_repository(self, github_url: str) -> str:
_, _, content = ingest(github_url)
Path(self.CONTENT_FILE).write_text(content, encoding="utf-8")
return content
def load_documents(self, content: str) -> list[Any]:
with tempfile.TemporaryDirectory() as temp_dir:
content_path = Path(temp_dir) / "repository_content.md"
content_path.write_text(content, encoding="utf-8")
reader = SimpleDirectoryReader(input_dir=temp_dir)
return reader.load_data()
Wyróżniają się dwie opcje. Po pierwsze, content.md jest zapisywany na dysku jako efekt uboczny. Nie jest to główna ścieżka, którą przechodzą dane, ale przydaje się do debugowania, ponieważ jego otwarcie pokazuje dokładnie to, co widzi model. Po drugie, load_documents wykorzystuje TemporaryDirectory. SimpleDirectoryReader z LlamaIndex wymaga katalogu, a nie surowego tekstu, więc tymczasowa folder jest najczystszym sposobem na przekazanie treści pamięci operacyjnej bez pozostawiania plików. Folder i jego zawartość znikają, gdy kończy się blok with.
Krok 4: silnik RAG
Jądrzem projektu jest rag_engine.py, więc warto podejść do tego stopniowo. RAGEngineFactory zajmuje się trzema odrębnymi aspektami: modelem językowym, modelem embeddingów oraz indeksem i promptem, które łączą proces wyszukiwania z generowaniem.
Model językowy
@staticmethod
@st.cache_resource
def get_llm() -> Groq:
api_key = os.environ.get("GROQ_API_KEY") or os.environ.get("OPENAI_API_KEY")
if not api_key:
raise ValueError("Set GROQ_API_KEY or OPENAI_API_KEY before loading a repository.")
return Groq(
api_key=api_key,
model="openai/gpt-oss-20b",
)
Tutaj kluczową rolę pełni dekorator @st.cache_resource. Streamlit uruchamia cały skrypt od początku do końca przy każdej interakcji, więc bez niego każdy kliknięcie lub naciśnięcie klawisza powodowałoby ponowną budowę klienta Groq. Funkcja cache_resource poleca Streamlitowi utworzenie obiektu tylko raz i jego późniejsze powrót – to właściwa forma prymitywów dla klientów API i załadowanych modeli. st.cache_data służy natomiast do danych możliwych do zserializowania.
Model embedding
@staticmethod
@st.cache_resource
def get_embedding_model() -> HuggingFaceEmbedding:
return HuggingFaceEmbedding(
model_name="BAAI/bge-large-en-v1.5",
trust_remote_code=True,
)
Model BAAI/bge-large-en-v1.5 działa lokalnie za pośrednictwem HuggingFace, bez żadnych połączeń sieciowych i bez opłat za każdy token. Przekształca każdą część repozytorium, a później każde pytanie, w wektor umieszczony w przestrzeni semantycznej, dzięki czemu podobne znaczenia znajdują się blisko siebie, nawet jeśli sformułowanie jest inne. Dzięki temu pytanie dotyczące sposobu autoryzacji może wyświetlić fragment tekstu omawiający logowanie, sesje lub JWT, mimo że żadne z tych słów nie występuje w pytaniu.
Ta sama zasada dotyczy cache’owania. Ładowanie modelu o wielkości około 1,3 GB przy każdym ponownym uruchomieniu uczyniłoby aplikację niewykorzystywalną, dlatego model jest ładowany tylko raz i następnie ponawialnie wykorzystywany.
Indeks i prompt
PROMPT = PromptTemplate(
"""
Context information is below.
---------------------
{context_str}
---------------------
Answer the query using only the context above.
Be precise and concise. If the answer cannot be found in the context,
respond with: "I don't know."
Query: {query_str}
Answer:
"""
)
def build(self, documents: list[Any]) -> Any:
Settings.llm = self.get_llm()
Settings.embed_model = self.get_embedding_model()
index = VectorStoreIndex.from_documents(
documents=documents,
transformations=[MarkdownNodeParser()],
show_progress=True,
)
query_engine = index.as_query_engine(streaming=True)
query_engine.update_prompts(
{"response_synthesizer:text_qa_template": self.PROMPT}
)
return query_engine
Oto co dzieje się, gdy uruchamiany jest build():
Settings.llmorazSettings.embed_modelto globalna konfiguracja LlamaIndex. Każdy kolejny komponent, od tworzenia indeksu po mechanizm wyszukiwania i silnik zapytań, pobiera je automatycznie, więc nie trzeba ich przekazywać przy każdym wywołaniu.transformations=[MarkdownNodeParser()]określa strategię dzielenia na fragmenty. Podział znormalizowanego tekstu Markdown według liczby znaków mógłby przypadkowo przeciąć funkcję na pół.MarkdownNodeParserzamiast tego dzieli tekst wzdłuż nagłówków i ram kodu, dzięki czemu każdy fragment stanowi spójną jednostkę: sekcję pliku, funkcję lub wyjaśnienie.VectorStoreIndex.from_documents(...)wstawia każdy fragment za pomocą skonfigurowanego modelu i przechowuje wektory w pamięci, gotowe do wyszukiwania pod kątem podobieństwa kosinowego w stosunku do wektora pytania.
index.as_query_engine(streaming=True) przekształca indeks w silnik, który włącza pytanie, wyszukuje najbardziej odpowiednie fragmenty, pakuje je do promptu i przesyła odpowiedź modelu token po tokenie, zamiast czekać na całą odpowiedź.query_engine.update_prompts(...) zastępuje domyślny szablon odpowiadania na pytania LlamaIndex powyższym, dostosowanym szablonem. Ze względu na poprawność to najważniejszy element całego systemu.Dlaczego prompt jest prawdziwą ochroną
Bез żadnych wyraźnych ograniczeń model językowy chętnie dostarczy płynną, pewną siebie odpowiedź na temat kodu, którego w żadnym z pobranych kontekstów nie ma, ponieważ właśnie tworzenie wiarygodnych kontynuacji jest cechą charakterystyczną tych modeli, podczas gdy fakty sprawdzone nie są takim celem. Szablon wypełnia tę lukę jedną zasadą: jeśli pobrany kontekst nie zawiera odpowiedzi, odpowiedź musi być prostym stwierdzeniem, że odpowiedź jest nieznana.
To zamienia halucynacje, będące standardowym zachowaniem modelu, w zasadę, której ma przestrzegać, a w praktyce to właśnie różnica między narzędziem godnym zaufania a takim, które z pewnością siebie wymyśla rzeczy. Jest to krótki fragment tekstu, ale bardziej przyczynia się do użyteczności projektu niż cała infrastruktura służąca do pobierania danych.
Krok 5: strumieniowanie z biblioteki synchronicznej
response.response_gen z LlamaIndex to zwykły generator synchroniczny, a jego iterowanie blokuje wątek do chwili, gdy będzie gotowy następny fragment. Tymczasem interfejs czatowy Streamlit chce asynchronicznie oczekiwać na fragmenty, aby pozostać responsywnym. AsyncResponseStreamer w pliku response_streamer.py łączy te dwa podejścia:
class AsyncResponseStreamer:
"""Convert LlamaIndex synchronous streaming into async streaming."""
async def stream(self, query_engine: Any, prompt: str) -> AsyncIterator[str]:
response = await asyncio.to_thread(query_engine.query, prompt)
if not hasattr(response, "response_gen"):
yield str(response)
return
response_iterator: Iterator[str] = iter(response.response_gen)
end_marker = object()
while True:
chunk = await asyncio.to_thread(next, response_iterator, end_marker)
if chunk is end_marker:
break
if chunk:
yield str(chunk)
Należy pamiętać, że next(response_iterator, end_marker) jest sam w sobie blokujący, ponieważ może czekać na połączeniu sieciowym, podczas gdy Groq generuje kolejny token. Dlatego też jest wykonywany na wątku roboczym za pomocą asyncio.to_thread. Użycie dedykowanego obiektu sygnalizacyjnego zamiast sprawdzania wartości None zapobiega subtelnemu błędowi: prawdziwy fragment danych mógłby teoretycznie być pustą struną lub czymś fałszywym, a obiekt sygnalizacyjny gwarantuje, że pętla zatrzyma się tylko wtedy, gdy iterator jest rzeczywiście wyczerpany.
To rozwiązanie można wykorzystać za każdym razem, gdy synchroniczna biblioteka AI, która obejmuje większość z nich, działa za interfejsem asynchronicznym lub strumieniowym: wszystkie blokujące wywołania należy przekazać na wątek, a wyniki należy dostarczać stopniowo, gdy biblioteka udostępnia generator.
Orchestrator konsumuje je w funkcji stream_response:
async def consume_stream() -> str:
complete_response = ""
try:
async for chunk in self.streamer.stream(query_engine, prompt):
complete_response += chunk
response_placeholder.markdown(complete_response + "▌")
response_placeholder.markdown(complete_response)
return complete_response
except Exception as error:
message = f"An error occurred while generating the response: {error}"
response_placeholder.error(message)
return message
return asyncio.run(consume_stream())
Charakterystyczny znak kursora widoczny podczas transmisji to mały element, który sprawia wrażenie, jakby tekst był wpisywany na żywo, a nie po prostu przychodził w blokach.
Krok 6: stan sesji
class AppState:
"""Manage Streamlit session state."""
@staticmethod
def initialize() -> None:
if "messages" not in st.session_state:
st.session_state.messages = []
if "repositories" not in st.session_state:
st.session_state.repositories = {}
if "github_url" not in st.session_state:
st.session_state.github_url = ""
@staticmethod
def reset_chat() -> None:
st.session_state.messages = []
gc.collect()
@staticmethod
def repository_key(github_url: str) -> str:
return hashlib.sha256(github_url.encode("utf-8")).hexdigest()
Trzy elementy znajdują się w stanie sesji. Jeden to historia rozmowy, drugi to słownik dostępnych silników zapytań indeksowanych za pomocą hasha adresu URL, a ostatni to aktualnie wybrany adres URL. Haszowanie adresu URL zamiast używania go bezpośrednio jako klucza zapewnia stałą, przewidywalną długość kluczy i unika problemów związanych z znakami takimi jak ukośniki, kropki oraz parametry zapytania.
reset_chat() usuwa tylko messages, pozostawiając bez zmian repositories; dzięki temu usunięcie rozmowy nie zmusza do ponownego pobierania i włączania repozytorium. Wyraźne wywołanie gc.collect() uwzględnia fakt, że długa historia rozmów w połączeniu z dużym indeksem w pamięci może zajmować znaczną ilość pamięci, więc przymusowe jej zwolnienie następuje szybko, zamiast czekać na normalny cykl Pythona.
Jedno pytanie w każdym pliku
app.py → loads .env, starts GitHubRAGApp
streamlit_app.py → renders UI, captures the question
app_state.py → looks up the cached query engine by URL hash
rag_engine.py → (already built) holds the index + prompt
response_streamer.py → streams the answer chunk by chunk
streamlit_app.py → writes each chunk into the chat UI
Każdy moduł ma dokładnie jedno zadanie, które utrzymuje dość zaawansowany przepływ danych obejmujący asynchroniczną komunikację, buforowanie, transmisję strumieniową oraz pętlę RAG ograniczoną czasem reakcji, przy czym całość mieści się w mniej niż 200 liniach kodu rozproszonej między sześcioma plikami.
Znane ograniczenia
- Tylko pamięć. Nic nie jest przechowywane między restartami. Naturalnym rozwiązaniem jest zapisywanie indeksu wektorowego na dysku lub w usługach takich jak Chroma czy Qdrant, dzięki czemu nie trzeba ponownie importować repozytoriów przy każdej sesji.
- Jeden aktywny repozytorium na sesję. W buforze mogą znajdować się liczne repozytoria, ale czat obsługuje tylko jeden z nich jednocześnie.
- Tylko publiczne repozytoria.
gitingestobsługuje publiczne adresy URL z GitHuba, natomiast prywatne repozytoria wymagałyby obsługi autoryzacji.
Cały zestaw narzędzi – Streamlit z LlamaIndex, Groq oraz embeddingi od HuggingFace – kosztuje prawie nic w przypadku małych repozytoriów, dzięki darmowej wersji Groq oraz lokalnym embeddingom. Rozmowa z bazą kodu to satysfakcjonujący pierwszy projekt typu RAG, ponieważ wyniki można sprawdzić w odniesieniu do czegoś już znanego: znajomość funkcjonowania kodu ułatwia wykrycie błędnej odpowiedzi.
Literatura pokrewna
- RAG w prostym Pythonie: dzielenie na fragmenty, TF-IDF, podobieństwo kosinowe i formułowanie zapytań — Tworzenie systemów generacji wzbogaconej o wyszukiwanie wyłącznie za pomocą modułu matematycznego: dzielenie treści na fragmenty metodą okna przesuwającego się, wektory TF-IDF, ranking według podobieństwa kosinowego, skuteczne zapytania oraz ich ograniczenia.
- Stworzenie asystenta rozmówczego Wikipedia RAG z użyciem LlamaIndex i ReAct — Indeksowanie stron Wikipedia na żądanie, włączanie fragmentów treści za pomocą małego kodera oraz umożliwienie agencji ReAct wywoływania narzędzi silnika zapytań, dzięki czemu odpowiedzi Chainlit opierają się na odnalezionych dowodach.