Discutez avec n’importe quel dépôt GitHub : Streamlit, LlamaIndex, Groq et embeddings locaux

Se repérer dans une base de code inconnue prend généralement des heures : parcourir les fichiers, deviner l’architecture, relire les mêmes fonctions. Une solution plus rapide consiste à utiliser une petite application qui prend en entrée une URL GitHub, génère des questions sur le code, et affiche les réponses au fur et à mesure qu’elles sont produites. Voici la mise en œuvre complète : comment les modules se connectent, ce que fait le moteur RAG au moment de la requête, une technique d’encadrement qui permet à une bibliothèque bloquante de fonctionner correctement au sein de Streamlit, ainsi que la formulation du prompt qui empêche le modèle de créer des réponses fictives.

Fonctionnalités de l’application

C’est une application Streamlit avec un flux simple :

1. Une URL de répertoire GitHub est collée.
2. L’application télécharge et indexe le répertoire.
3. Les questions sont posées via une interface de chat.

Les extraits de code et de documentation pertinents sont récupérés, ce qui permet d’obtenir une réponse basée exclusivement sur le contenu réel du dépôt.

C’est le modèle classique de génération augmentée par la récupération d’informations, appliqué au code source et non aux documents en prose. L’architecture utilise Streamlit pour l’interface, LlamaIndex pour l’indexation et les requêtes, Groq comme modèle de langage, ainsi que des embeddings HuggingFace exécutés localement, avec gitingest chargé du téléchargement.

Structure du projet

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

Le chemin des données :

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

Étape 1 : un point d’entrée très petit

Tout commence dans app.py, conçu délibérément pour être très compact. Ses seules fonctions sont de charger les variables d’environnement et de transférer le contrôle à la classe de l’application :

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

load_dotenv() lit un fichier local .env et expose ses valeurs en tant que variables d’environnement, ce qui permet de protéger les secrets du code source :

GROQ_API_KEY=your_groq_api_key_here

Commencez par :

streamlit run app.py

Il est plus important que cela n’y paraît de garder ce fichier minimal. Comme c’est le seul endroit qui lance réellement Streamlit, tous les autres modules peuvent être testés unitairement sans l’environnement de exécution de Streamlit.

Étape 2 : l’orchestrateur

GitHubRAGApp, présent dans streamlit_app.py, est le coordinateur. Il gère trois collaborateurs et relie l’interface à eux :

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

Plusieurs détails méritent attention :

• st.status(..., expanded=True) affiche un journal de progression en temps réel plutôt qu’un indicateur en rotation. Le pipeline comporte trois étapes lentes : le téléchargement, l’analyse et l’incorporation, donc cette visibilité améliore considérablement l’expérience utilisateur.
• Le chargement est idempotent. Si la clé de hachage du répertoire se trouve déjà dans st.session_state.repositories, l’application passe directement aux données déjà chargées et omet le téléchargement ainsi que l’incorporation. Ce cache rend les demandes répétées et les clics doubles accidentels peu coûteux en ressources.
• Chaque étape lente passe par run_async, une petite mais importante utilité :

@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 et build sont toutes des fonctions synchrones bloquantes qui s’occupent des tâches d’incorporation gourmandes en ressources réseau, en fichiers et en CPU. run_async transfère chacune d’elles sur un thread de travail à l’aide de asyncio.to_thread, de sorte qu’un téléchargement lent ne bloque pas la boucle de réexécution monolithique de Streamlit et que st.status continue d’être mis à jour.

Les entrées de chat suivent le schéma standard de Streamlit, la génération étant confiée à 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})

Étape 3 : ingestion du répertoire

Au lieu d’écrire un parcourreur GitHub personnalisé, l’application s’appuie sur gitingest, qui télécharge un répertoire et le transforme en une représentation Markdown conservant les limites des fichiers, la structure des dossiers ainsi que leur contenu. La logique se trouve dans 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()

Deux options se détachent. Premièrement, content.md est enregistré sur le disque en tant qu’effet secondaire. Ce n’est pas la voie principale empruntée par les données, mais il est utile pour le débogage, car son ouverture montre exactement ce que voit le modèle. Deuxièmement, load_documents utilise un TemporaryDirectory. SimpleDirectoryReader de LlamaIndex nécessite un répertoire et non du texte brut, donc un dossier temporaire constitue le moyen le plus propre pour transmettre du contenu en mémoire sans laisser de fichiers. Le dossier et son contenu disparaissent lorsque le bloc with se termine.

Étape 4 : le moteur RAG

Le cœur du projet est rag_engine.py, il convient donc d’y prêter une attention minutieuse. RAGEngineFactory gère trois aspects distincts : le modèle de langage, le modèle d’incorporation, ainsi que l’index et la requête qui relient la récupération aux génération.

Le modèle de langage

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

Le décorateur @st.cache_resource effectue ici le travail réel. Streamlit exécute à nouveau tout le script du début à la fin à chaque interaction, de sorte que sans lui, chaque clic ou frappe au clavier récréerait le client Groq. cache_resource indique à Streamlit de créer l’objet une seule fois puis de renvoyer le même objet par la suite, ce qui correspond au type approprié pour les clients API et les modèles chargés. st.cache_data est l’équivalent pour les données serialisables.

Le modèle d’incorporation

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

Le modèle BAAI/bge-large-en-v1.5 s’exécute localement via HuggingFace, sans appel réseau et sans frais par token. Il convertit chaque partie du répertoire, puis chaque question, en un vecteur situé dans l’espace sémantique, de sorte que des significations similaires se trouvent proches les unes des autres même en cas de différences de formulation. Cela permet à une question comme « Comment fonctionne l’authentification ici ? » de faire apparaître un extrait abordant la connexion, les sessions ou JWT, même si aucun de ces mots n’apparaît dans la question.

Le même principe de mise en cache s’applique. Charger un modèle d’environ 1,3 GB à chaque exécution rendrait l’application inutilisable ; il est donc chargé une seule fois et réutilisé.

L’index et le 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

Ceci montre ce qui se passe lorsque build() est exécuté :

1. Settings.llm et Settings.embed_model constituent la configuration globale de LlamaIndex. Chaque composant ultérieur, depuis la construction de l’index jusqu’au récupérateur et au moteur de requêtes, les utilise automatiquement, il n’est donc pas nécessaire de les transmettre à chaque appel.
2. transformations=[MarkdownNodeParser()] définit la stratégie de segmentation. Diviser le Markdown normalisé en fonction du nombre de caractères risquerait de couper une fonction en deux. MarkdownNodeParser effectue plutôt la segmentation le long des en-têtes et des blocs de code, de sorte que chaque segment constitue une unité cohérente : une section de fichier, une fonction ou une explication.
3. VectorStoreIndex.from_documents(...) incorpore chaque segment avec le modèle configuré et conserve les vecteurs en mémoire, prêts pour une recherche par similarité cosinus par rapport à l’embedding d’une question.

index.as_query_engine(streaming=True) transforme l’index en un moteur qui intègre la question, récupère les fragments les plus pertinents, les assemble en un prompt, et transmet les tokens de réponse du modèle un par un au lieu d’attendre la réponse complète.

query_engine.update_prompts(...) remplace le modèle par défaut de LlamaIndex pour la réponse à des questions par celui personnalisé ci-dessus. Pour assurer la précision, c’est l’élément le plus important du système.

Pourquoi le prompt est la véritable protection

Lorsqu’aucune contrainte explicite n’est imposée, un modèle de langage produira volontiers une réponse fluide et assurée concernant du code qui n’apparaît nulle part dans le contexte récupéré, car c’est précisément produire des continuations plausibles que ces modèles sont conçus pour faire, contrairement aux faits vérifiés. Le modèle de template comble cette lacune grâce à une règle simple : lorsque le contexte récupéré ne contient pas la réponse, la réponse doit être une déclaration directe indiquant que celle-ci est inconnue.

Cela transforme l’hallucination, qui était le comportement par défaut du modèle, en une règle qu’il doit suivre ; en pratique, c’est là la différence entre un outil digne de confiance et un outil qui invente des choses avec assurance. Il s’agit d’un petit fragment de texte, pourtant il contribue davantage à rendre le projet utilisable que toute la machinerie de récupération qui l’entoure.

Étape 5 : streaming depuis une bibliothèque synchrone

response.response_gen de LlamaIndex est un générateur synchrone ordinaire, et son itération bloque le thread jusqu’à ce que la prochaine partie soit prête. L’interface de chat de Streamlit, quant à elle, souhaite attendre les données par tranches de manière asynchrone pour rester réactive. AsyncResponseStreamer dans response_streamer.py crée un pont entre les deux :

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)

Il convient de se rappeler que next(response_iterator, end_marker) est lui-même bloquant, car il peut attendre sur le réseau pendant que Groq génère le token suivant. Il est donc également exécuté sur un thread de travail via asyncio.to_thread. L’utilisation d’un objet sentinelle dédié à la place de une vérification pour None permet d’éviter un bug subtil : un chunk légitime pourrait en théorie être une chaîne vide ou quelque chose de falsy, et un objet sentinelle garantit que la boucle s’arrête uniquement lorsque l’itérateur est véritablement épuisé.

Cette approche est reutilisable chaque fois qu’une bibliothèque d’IA synchrone, qui en comprend la plupart, est utilisée derrière une interface asynchrone ou de streaming : il suffit d’envoyer chaque appel bloquant sur un thread et de produire progressivement des résultats dès que la bibliothèque expose un générateur.

L’orchestrateur l’utilise dans 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())

Le caractère de curseur affiché pendant la diffusion est un petit détail qui donne l’impression que le contenu est tapé au fur et à mesure plutôt que d’arriver par blocs.

Étape 6 : état de session

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

Trois éléments sont stockés dans l’état de session. Le premier est l’historique des conversations, le second est un dictionnaire des moteurs de requête prêts à l’emploi, indexé par le hash de l’URL, et le dernier est l’URL actuellement sélectionnée. Le fait d’hasher l’URL, au lieu de l’utiliser directement comme clé, permet aux clés d’avoir une longueur fixe et prévisible, tout en évitant les problèmes liés aux barres obliques, aux points et aux paramètres de requête.

reset_chat() ne supprime que les éléments messages et laisse intacts les repositories ; ainsi, la suppression d’une conversation n’oblige pas à télécharger et à intégrer à nouveau le répertoire correspondant. L’appel explicite de gc.collect() prend en compte le fait qu’un historique de conversations long accompagné d’un index important en mémoire peut occuper beaucoup de mémoire, et force sa libération immédiate au lieu d’attendre le cycle normal de Python.

Une seule question par fichier

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

Chaque module a exactement une seule fonction, ce qui permet de maintenir un pipeline assez sophistiqué, incluant des mécanismes d’asynchronisation, de mise en cache, de streaming ainsi qu’un cycle RAG contraint par des délais courts, tout en restant lisible avec moins de 200 lignes réparties sur six fichiers.

Limites connues

• Seule la mémoire est utilisée. Rien n’est conservé entre les redémarrages. Une amélioration naturelle consisterait à enregistrer l’index vectoriel sur disque ou dans un stockage comme Chroma ou Qdrant, afin de ne pas avoir à retraiter les dépôts à chaque session.
• Un seul dépôt actif par session. Plusieurs dépôts peuvent être stockés en cache, mais le chat ne cible qu’un seul d’entre eux à la fois.
• Seuls les dépôts publics sont pris en charge. gitingest ne fonctionne que avec des URLs GitHub publiques ; les dépôts privés nécessiteraient un mécanisme d’authentification.

Aucun test automatisé.Cela est acceptable pour un outil personnel, mais ils devraient être ajoutés avant que quoi que ce soit ne soit qualifié de niveau production.

Toute la stack, Streamlit avec LlamaIndex, Groq et des embeddings de HuggingFace, coûte presque rien pour de petits dépôts, grâce au plan gratuit de Groq et aux embeddings locaux. Discuter avec un codebase constitue un premier projet RAG satisfaisant, car les résultats peuvent être vérifiés par rapport à quelque chose déjà compris : savoir ce que fait le code permet de repérer facilement une réponse erronée.