Chatten mit jedem GitHub-Repo: Streamlit, LlamaIndex, Groq und lokale Embeddings

Das Orientieren in einer unbekannten Codebasis kostet in der Regel Stunden: Das Durchklicken von Dateien, das Raten bezüglich der Architektur, das Wiederlesen derselben Funktionen. Ein schnellerer Weg ist eine kleine Anwendung, in die eine GitHub-URL eingegeben wird, aus der Fragen zum Code gestellt werden und zu denen Antworten in Echtzeit generiert und zurückgegeben werden. Unten finden Sie die vollständige Implementierung: wie die Module miteinander verbunden sind, was der RAG-Engine bei Abfragen zur Verfügung steht, eine Thread-Technik, die es einer blockierenden Bibliothek ermöglicht, reibungslos innerhalb von Streamlit zu laufen, sowie die Formulierung der Prompt, die verhindert, dass das Modell falsche Antworten erzeugt.

Was die Anwendung tut

Es handelt sich um eine Streamlit-Anwendung mit einem einfachen Ablauf:

1. Eine GitHub-Repository-URL wird eingegeben.
2. Die Anwendung lädt das Repository herunter und erstellt einen Index davon.
3. Fragen werden über eine Chat-Schnittstelle gestellt.

Relevante Code- und Dokumentationsabschnitte werden abgerufen, wobei die Antwort ausschließlich auf dem basiert, was das Repository tatsächlich enthält.

Es handelt sich um das klassische Retrieval-augmented Generation-Muster, das auf Quellcode und nicht auf prosaischen Dokumenten angewandt wird. Für die Benutzeroberfläche kommt Streamlit zum Einsatz, für Indexierung und Abfragen LlamaIndex, für das Sprachmodell Groq sowie HuggingFace-Embeddings, die lokal ausgeführt werden, wobei gitingest für das Herunterladen zuständig ist.

Aufbau des Projekts

.
├── 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)

Der Datenpfad:

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

Schritt 1: ein sehr kleiner Eingangspunkt

Alles beginnt in app.py, das absichtlich sehr klein gehalten wird. Seine einzigen Aufgaben sind es, Umgebungsvariablen zu laden und die Steuerung an die App-Klasse weiterzugeben:

from streamlit_app import GitHubRAGApp
from dotenv import load_dotenv
if __name__ == "__main__":
    load_dotenv()
    GitHubRAGApp().run()

load_dotenv() liest eine lokale .env-Datei und stellt ihre Werte als Umgebungsvariablen zur Verfügung, wodurch Geheimnisse aus dem Quellcode entfernt werden:

GROQ_API_KEY=your_groq_api_key_here

Fangen Sie damit an:

streamlit run app.py

Es ist wichtiger, diese Datei möglichst klein zu halten, als es auf den ersten Blick scheint. Da sie der einzige Ort ist, an dem Streamlit tatsächlich gestartet wird, können alle anderen Module ohne den Laufzeitumfeld von Streamlit unitgetestet werden.

Schritt 2: Der Orchesterer

GitHubRAGApp in streamlit_app.py ist der Koordinator. Er verantwortet drei Hilfskomponenten und verbindet die Benutzeroberfläche mit ihnen:

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}")

Mehrere Details verdienen Aufmerksamkeit:

• st.status(..., expanded=True) zeigt ein aktuelles Fortschrittsprotokoll anstelle eines Spinners. Der Pipeline umfasst drei langsame Phasen – Download, Parsing und Einbetten – wodurch die Sichtbarkeit das Erlebnis erheblich verbessert.
• Das Laden ist idempotent. Wenn der Hash-Schlüssel des Repositoriums bereits in st.session_state.repositories vorhanden ist, springt die Anwendung direkt zu den bereits geladenen Daten und überspringt den Download sowie das Einbetten. Dieser Cache macht wiederholte Anfragen sowie versehentliche Doppelklicks unproblematisch.
• Jeder langsame Schritt wird über run_async abgewickelt, ein kleines, aber wichtiges Hilfsprogramm:

@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 und build sind alle blockierende, synchrone Funktionen, die Netzwerk-, Datei- sowie rechenintensive Einbettungsarbeiten umfassen. run_async verschiebt jede dieser Funktionen mithilfe von asyncio.to_thread auf einen Worker-Thread, sodass ein langsames Herunterladen den eindimensionalen Wiederholungszyklus von Streamlit nicht blockiert und st.status weiterhin aktualisiert wird.

Die Eingaben im Chat folgen dem Standardmuster von Streamlit, wobei die Generierung an stream_response übergeben wird:

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})

Schritt 3: Einbeziehen des Repositoriums

Anstatt einen eigenen GitHub-Crawler zu schreiben, nutzt die Anwendung gitingest, welches ein Repositorium herunterlädt und es in eine einzige Markdown-Darstellung umwandelt, die Dateigrenzen, Verzeichnisstruktur und Inhalt beibehält. Die Logik befindet sich in 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()

Zwei Optionen kommen besonders in Frage. Erstens wird content.md als Nebeneffekt auf der Festplatte gespeichert. Es ist nicht der Hauptweg, den die Daten nehmen, aber es ist nützlich zum Debuggen, da das Öffnen des Files genau zeigt, was das Modell sieht. Zweitens verwendet load_documents einen TemporaryDirectory. SimpleDirectoryReader von LlamaIndex benötigt einen Ordner und nicht rohen Text, weshalb ein temporärer Ordner die sauberste Methode ist, In-Memory-Inhalte ohne hinterlassene Dateien weiterzugeben. Der Ordner und sein Inhalt verschwinden, sobald der with-Block beendet ist.

Schritt 4: Der RAG-Engine

Der Kern des Projekts ist rag_engine.py, daher lohnt es sich, dabei vorsichtig vorzugehen. RAGEngineFactory kümmert sich um drei separate Aspekte: das Sprachmodell, das Embedding-Modell sowie den Index und die Anfrage, die die Suche mit der Generierung verbinden.

Das Sprachmodell

@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",
    )

Der Dekorator @st.cache_resource übernimmt hier die eigentliche Arbeit. Streamlit führt bei jeder Interaktion den gesamten Script-Code von oben nach unten erneut aus, wodurch ohne ihn jeder Klick oder Tastendruck den Groq-Client neu erstellen würde. cache_resource weist Streamlit an, das Objekt einmal zu erstellen und danach immer dasselbe zurückzugeben – was für API-Cliente sowie geladene Modelle die richtige Vorgehensweise ist. st.cache_data dient hingegen für serialisierbare Daten.

Das Embedding-Modell

@staticmethod
@st.cache_resource
def get_embedding_model() -> HuggingFaceEmbedding:
    return HuggingFaceEmbedding(
        model_name="BAAI/bge-large-en-v1.5",
        trust_remote_code=True,
    )

Das Modell BAAI/bge-large-en-v1.5 wird über HuggingFace lokal ausgeführt, ohne Netzwerkanrufe und ohne Gebühren pro Token. Es wandelt jeden Teil des Repositoriums sowie später jede Frage in einen Vektor um, der im semantischen Raum positioniert wird, sodass ähnliche Bedeutungen auch bei unterschiedlicher Formulierung nahe beieinander liegen. Dadurch kann eine Frage wie „Wie funktioniert die Authentifizierung hier?“ einen Abschnitt zum Thema Anmelden, Sessions oder JWT anzeigen, obwohl keines dieser Wörter in der Frage vorkommt.

Dieselbe Logik gilt für das Caching. Das Laden eines etwa 1,3 GB großen Modells bei jedem Neustart würde die Anwendung unbrauchbar machen, weshalb es nur einmal geladen und danach wiederverwendet wird.

Der Index und die Anfrage

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

So verhält es sich, wenn build() ausgeführt wird:

1. Settings.llm und Settings.embed_model stellen die globale Konfiguration von LlamaIndex dar. Jedes nachfolgende Komponente, von der Indexerstellung über den Retriever bis zum Abfragesystem, verwendet diese automatisch, sodass nichts bei jeder Aufrufung übergeben werden muss.
2. transformations=[MarkdownNodeParser()] legt die Strategie zur Aufteilung in Blöcke fest. Eine Aufteilung des flachen Markdown-Textes nach Zeichenzahl könnte beispielsweise eine Funktion in zwei Teile teilen. MarkdownNodeParser teilt hingegen entlang von Überschriften und Codeblöcken auf, wodurch ein Block in der Regel eine zusammenhängende Einheit darstellt: ein Dateisegment, eine Funktion oder eine Erklärung.
3. VectorStoreIndex.from_documents(...) embeddet jeden Block mit dem konfigurierten Modell und speichert die Vektoren im Speicher, damit sie für eine Suche nach kosinusähnlicher Ähnlichkeit mit dem Embedding einer Frage verwendet werden können.

index.as_query_engine(streaming=True) verwandelt den Index in einen Motor, der die Frage einbettet, die am besten passenden Abschnitte abruft, sie in einen Prompt zusammenfasst und die Antwort des Modells tokenweise streamt, anstatt auf die gesamte Antwort zu warten.

query_engine.update_prompts(...) ersetzt das Standard-Frage-Antwort-Muster von LlamaIndex durch das oben beschriebene benutzerdefinierte Muster. Für die Korrektheit ist dies der wichtigste Bestandteil des Systems.

Warum der Prompt der eigentliche Schutzmechanismus ist

Solange es keine explizite Einschränkung gibt, wird ein Sprachmodell gerne eine flüssige, selbstsichere Antwort zu Code liefern, der im abgerufenen Kontext überhaupt nicht vorkommt – schließlich sind solche plausiblen Fortsetzungen genau das, wofür diese Modelle konzipiert sind, während überprüfte Fakten dies nicht tun. Das Muster schließt diese Lücke mit einer einzigen Regel: Wenn der abgerufene Kontext die Antwort nicht enthält, muss die Antwort eine einfache Aussage sein, dass die Antwort unbekannt ist.

Dadurch wird Halluzination von dem üblichen Verhalten des Modells zu einer Regel, der es folgen soll, und in der Praxis ist das der Unterschied zwischen einem vertrauenswürdigen Tool und einem, das selbstbewusst Dinge erfindet. Es handelt sich dabei um einen kurzen Textabschnitt, doch er trägt mehr dazu bei, das Projekt nutzbar zu machen, als all die Mechanismen zur Informationsabrufung zusammen.

Schritt 5: Streaming aus einer synchronen Bibliothek

LlamaIndex’s response.response_gen ist ein gewöhnlicher synchroner Generator, und dessen Iteration blockiert den Thread, bis der nächste Datensatz bereit ist. Die Chat-Oberfläche von Streamlit hingegen möchte, um weiterhin reaktiv zu bleiben, die Datenblöcke asynchron abwarten. AsyncResponseStreamer in response_streamer.py verbindet diese beiden Ansätze miteinander:

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)

Man sollte sich merken, dass next(response_iterator, end_marker) selbst blockierend ist, da es im Netzwerk warten muss, während Groq das nächste Token erzeugt. Deshalb wird es auch über asyncio.to_thread auf einem Worker-Thread ausgeführt. Die Verwendung eines speziellen Sentinel-Objekts anstelle einer Überprüfung auf None vermeidet einen subtilen Fehler: Ein legitimer Datensatz könnte theoretisch eine leere Zeichenkette oder etwas Falsches sein, und ein Sentinel stellt sicher, dass der Loop nur dann stoppt, wenn der Iterator tatsächlich erschöpft ist.

Dieses Muster lässt sich immer dann wiederverwenden, wenn eine synchrone KI-Bibliothek – zu der die meisten gehören – hinter einer asynchronen oder strömenden Schnittstelle steht: Jeder blockierende Aufruf wird auf einen Thread verschoben, und es werden schrittweise Werte bereitgestellt, sobald die Bibliothek einen Generator bereitstellt.

Der Orchestrator verwendet dies in 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())

Das am Ende angezeigte Cursor-Symbol während des Streamens ist eine kleine Geste, die das Erscheinungsbild so gestaltet, als würde der Inhalt eingegeben und nicht einfach in Blöcken übertragen.

Schritt 6: Sitzungszustand

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()

Drei Elemente befinden sich im Sitzungszustand. Eines ist die Chat-Geschichte, ein weiteres ist ein Wörterbuch mit verfügbaren Abfragemaschinen, das nach dem Hash der URL indiziert ist, und das letzte ist die derzeit ausgewählte URL. Durch das Hashen der URL anstelle einer direkten Verwendung als Schlüssel bleiben die Schlüssel von konstanter, vorhersehbarer Länge und es werden Probleme durch Schrägstriche, Punkte sowie Abfragemethoden vermieden.

reset_chat() löscht nur messages und lässt repositories unberührt, sodass das Löschen einer Konversation nicht dazu führt, dass der Repository erneut heruntergeladen und eingebettet wird. Die explizite Aufrufung von gc.collect() berücksichtigt, dass eine lange Chat-Geschichte zusammen mit einem großen In-Memory-Index erheblichen Speicherbedarf verursachen kann, und führt eine schnelle Freigabe des Speichers durch, anstatt auf den normalen Zyklus von Python zu warten.

Eine Frage in jeder Datei

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

Jedes Modul hat genau eine Aufgabe, wodurch ein recht komplexer Ablauf mit asynchroner Verbindungsaufrechterhaltung, Caching, Streaming sowie einem durch Zeitlimits begrenzten RAG-Loop entsteht, der in insgesamt sechs Dateien unter 200 Zeilen umfasst.

Bekannte Einschränkungen

• Nur Speicher. Nichts bleibt nach Neustarten erhalten. Eine natürliche Verbesserung besteht darin, den Vektorindex auf der Festplatte oder in einem Speicher wie Chroma oder Qdrant zu speichern, sodass die Repositorien nicht bei jeder Sitzung erneut eingebunden werden müssen.
• Ein aktives Repository pro Sitzung. Mehrere Repositorien können im Cache gespeichert sein, doch der Chat arbeitet jeweils nur mit einem davon gleichzeitig.
• Nur öffentliche Repositorien. gitingest verwendet öffentliche GitHub-URLs; private Repositorien würden eine Authentifizierungsverarbeitung erfordern.

Keine automatisierten Tests. Für ein persönliches Tool akzeptabel, doch sie sollten hinzugefügt werden, bevor etwas als produktionsreif bezeichnet wird.

Dank der kostenlosen Stufe von Groq sowie lokaler Embeddings kostet die gesamte Technologie – Streamlit mit LlamaIndex, Groq und HuggingFace Embeddings – für kleine Repositorien fast nichts. Das Chatten mit einer Codebasis ist ein zufriedenstellendes erstes RAG-Projekt, da die Ergebnisse mit bereits Bekanntem verglichen werden können: Wenn man weiß, was der Code tut, fällt ein falsches Antwort leicht auf.