Chat With Any GitHub Repo: Streamlit, LlamaIndex, Groq and Local Embeddings

Getting oriented in an unfamiliar codebase usually costs hours: clicking through files, guessing at the architecture, rereading the same functions. A faster route is a small app where a GitHub URL goes in, questions about the code come out, and answers stream back as they are generated. Below is the full implementation: how the modules connect, what the RAG engine does at query time, a threading technique that lets a blocking library stream smoothly inside Streamlit, and the wording of the prompt that stops the model from fabricating answers.

What the app does

It is a Streamlit app with a simple flow:

1. A GitHub repository URL is pasted in.
2. The app downloads and indexes the repository.
3. Questions are asked in a chat interface.
4. Relevant code and documentation chunks are retrieved and an answer streams back, grounded only in what the repository really contains.

It is the classic retrieval-augmented generation pattern, applied to source code and not to prose documents. The stack is Streamlit for the interface, LlamaIndex for indexing and querying, Groq for the language model, and HuggingFace embeddings that run locally, with gitingest handling the download.

Layout of the project

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

The data path:

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

Step 1: a very small entry point

Everything starts in app.py, kept deliberately tiny. Its only duties are to load environment variables and hand control to the app class:

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

load_dotenv() reads a local .env file and exposes its values as environment variables, which keeps secrets out of the source:

GROQ_API_KEY=your_groq_api_key_here

Start it with:

streamlit run app.py

Keeping this file minimal matters more than it looks. Because it is the only place that actually launches Streamlit, every other module can be unit-tested without Streamlit's runtime.

Step 2: the orchestrator

GitHubRAGApp, in streamlit_app.py, is the coordinator. It owns three collaborators and connects the interface to them:

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

Several details deserve attention:

• st.status(..., expanded=True) shows a live progress log instead of a spinner. The pipeline has three slow stages, download, parse and embed, so the visibility improves the experience considerably.
• Loading is idempotent. If the repository's hash key already sits in st.session_state.repositories, the app jumps straight to already loaded and skips the download and embedding. That cache makes repeated questions and accidental double clicks cheap.
• Each slow step passes through run_async, a small but important utility:

@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 and build are all blocking, synchronous functions covering network, file and CPU-heavy embedding work. run_async moves each onto a worker thread with asyncio.to_thread, so a slow download does not freeze Streamlit's single-threaded rerun loop and st.status keeps updating.

Chat input follows the standard Streamlit pattern, with generation handed off to 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})

Step 3: ingesting the repository

Rather than writing a custom GitHub crawler, the app relies on gitingest, which downloads a repository and flattens it into one Markdown representation that keeps file boundaries, directory structure and content. The logic lives 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()

Two choices stand out. First, content.md is saved to disk as a side effect. It is not the main route the data takes, but it is handy for debugging, since opening it shows exactly what the model sees. Second, load_documents uses a TemporaryDirectory. LlamaIndex's SimpleDirectoryReader wants a directory and not raw text, so a temporary folder is the cleanest way to pass in-memory content without leaving files behind. The folder and its contents vanish when the with block ends.

Step 4: the RAG engine

The core of the project is rag_engine.py, so it is worth going slowly. RAGEngineFactory handles three separate concerns: the language model, the embedding model, and the index and prompt that tie retrieval to generation.

The language model

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

The @st.cache_resource decorator does real work here. Streamlit reruns the whole script from top to bottom on every interaction, so without it every click or keystroke would rebuild the Groq client. cache_resource instructs Streamlit to build the object once and return the same one afterward, which is the correct primitive for API clients and loaded models. st.cache_data is the counterpart for serializable data.

The embedding model

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

The model BAAI/bge-large-en-v1.5 runs locally through HuggingFace, with no network call and no per-token charge. It converts each chunk of the repository, and later each question, into a vector positioned in semantic space so that similar meaning lands close together even when the wording differs. That lets a question like how authentication works here pull up a chunk discussing login, sessions or JWT, though none of those words appear in the question.

The same caching reasoning applies. Loading a model of around 1.3 GB on every rerun would make the app unusable, so it is loaded once and reused.

The index and the 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

Here is what happens when build() runs:

1. Settings.llm and Settings.embed_model are LlamaIndex's global configuration. Every later component, from index construction to the retriever and the query engine, picks them up automatically, so nothing needs to be passed to each call.
2. transformations=[MarkdownNodeParser()] sets the chunking strategy. Cutting the flattened Markdown by character count would happily slice a function in half. MarkdownNodeParser splits along headers and code fences instead, so a chunk tends to be one coherent unit: a file section, a function, an explanation.
3. VectorStoreIndex.from_documents(...) embeds every chunk with the configured model and keeps the vectors in memory, ready for cosine-similarity search against a question's embedding.
4. index.as_query_engine(streaming=True) turns the index into an engine that embeds the question, retrieves the best-matching chunks, packs them into a prompt, and streams the model's reply token by token instead of waiting for the whole answer.
5. query_engine.update_prompts(...) swaps LlamaIndex's default question-answering template for the custom one above. For correctness, this is the single most important piece of the system.

Why the prompt is the real safeguard

With no explicit constraint, a language model will happily produce a fluent, confident answer about code that is nowhere in the retrieved context, since producing plausible continuations is what these models are built to do and verified facts are not. The template closes that gap with a single rule: when the retrieved context does not contain the answer, the reply must be a plain statement that the answer is unknown.

That turns hallucination from the model's default behavior into a rule it has been told to follow, and in practice that is the difference between a tool worth trusting and one that confidently invents things. It is a small piece of text, yet it does more to make the project usable than all the retrieval machinery around it.

Step 5: streaming from a synchronous library

LlamaIndex's response.response_gen is an ordinary synchronous generator, and iterating it blocks the thread until the next piece is ready. Streamlit's chat interface, meanwhile, wants to await chunks asynchronously to stay responsive. AsyncResponseStreamer in response_streamer.py bridges the two:

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)

The detail to remember is that next(response_iterator, end_marker) is itself blocking, since it may wait on the network while Groq produces the next token. It therefore also runs on a worker thread through asyncio.to_thread. Using a dedicated sentinel object in place of a check for None avoids a subtle bug: a legitimate chunk could in theory be an empty string or something falsy, and a sentinel guarantees that the loop stops only when the iterator is truly exhausted.

The pattern is reusable whenever a synchronous AI library, which includes most of them, sits behind an async or streaming interface: push every blocking call to a thread, and yield incrementally whenever the library exposes a generator.

The orchestrator consumes it 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())

The trailing cursor character shown while streaming is a small touch that makes the output feel as if it is being typed and not simply arriving in blocks.

Step 6: session state

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

Three items live in session state. One is the chat history, another is a dictionary of ready query engines indexed by a hash of the URL, and the last is whichever URL is currently chosen. Hashing the URL, instead of using it as a key directly, keeps keys at a fixed, predictable length and avoids escaping problems from slashes, dots and query parameters.

reset_chat() clears only messages and leaves repositories alone, so clearing a conversation does not force the repository to be downloaded and embedded again. The explicit gc.collect() acknowledges that a long chat history together with a large in-memory index can hold meaningful memory, and forcing collection releases it promptly instead of waiting for Python's normal cycle.

One question through every file

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

Each module has exactly one job, which keeps a fairly sophisticated pipeline, with async bridging, caching, streaming and a prompt-constrained RAG loop, readable at under 200 lines across six files.

Known limits

• Memory only. Nothing persists across restarts. A natural improvement is persisting the vector index to disk or to a store such as Chroma or Qdrant, so repositories need not be re-ingested every session.
• One active repository per session. Multiple repositories can sit in the cache, yet chat targets just one at once.
• Public repositories only. gitingest targets public GitHub URLs, and private repositories would need authentication handling.
• No automated tests. Acceptable for a personal tool, though they should be added before anything is called production-grade.

The whole stack, Streamlit with LlamaIndex, Groq and HuggingFace embeddings, costs almost nothing for small repositories, thanks to Groq's free tier and local embeddings. Chatting with a codebase is a satisfying first RAG project because the results can be checked against something already understood: knowing what the code does makes a wrong answer easy to spot.