Чат з будь-яким репозиторієм GitHub: Streamlit, LlamaIndex, Groq та локальні ембеддинги

Орієнтація у незнайомому кодовому базисі зазвичай займає години: перегляд файлів, припущення щодо архітектури, повторне читання одних і тих самих функцій. Швидшим способом є невеликий додаток, куди вводиться URL з GitHub, потім ставляться запитання щодо коду, а відповіді надходять у момент їх генерації. Нижче наведено повну реалізацію: як підключаються модулі, що робить двигун RAG під час запиту, техніка потоків, яка дозволяє блокуючій бібліотеці плавно працювати всередині Streamlit, та формулювання запиту, яке заважає моделі вигадувати відповіді.

Що робить додаток

Це додаток Streamlit із простим алгоритмом роботи:

1. Вводиться URL репозиторію з GitHub.
2. Додаток завантажує та індексує репозиторій.
3. Запитання ставляться через інтерфейс чату.

Завантажуються відповідні фрагменти коду та документації, і назад надсилається відповідь, яка ґрунтується виключно на тому, що дійсно міститься у репозиторії.

Це класичний патерн генерації з допомогою пошуку, застосований до вихідного коду, а не до прозових документів. Для інтерфейсу використовується Streamlit, для індексації та пошуку — LlamaIndex, для мовної моделі — Groq, а ембеддинги від HuggingFace працюють локально; завантаження здійснюється за допомогою gitingest.

Структура проекту

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

Шлях до даних:

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

Крок 1: дуже малий точка входу

Усе починається з файлу app.py, який навмисно залишається дуже малим. Його єдині функції — завантажувати змінні середовища та передавати керування класу додатку:

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

load_dotenv() читає локальний файл .env та робить його значення доступними як змінні середовища, що дозволяє залишити конфіденційну інформацію поза кодом:

GROQ_API_KEY=your_groq_api_key_here

Почніть з:

streamlit run app.py

Зберігання цього файлу у мінімальному вигляді має більше значення, ніж здається. Оскільки саме він є єдиним місцем, де фактично запускається Streamlit, кожен інший модуль можна протестувати як окремий елемент без використання середовища виконання Streamlit.

Крок 2: організатор

GitHubRAGApp у файлі streamlit_app.py є координатором. Він керує трьома співробітниками та під’єднує інтерфейс до них:

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

Кілька деталей заслуговують уваги:

• st.status(..., expanded=True) відображає живий журнал прогресу замість індикатора завантаження. У процесі є три повільні етапи – завантаження, обробка та вбудовування, тож така наявність інформації значно покращує користувацький досвід.
• Завантаження є ідемпотентним. Якщо хеш-ключ репозиторію вже знаходиться у st.session_state.repositories, додаток одразу переходить до вже завантажених даних, пропускаючи етапи завантаження та вбудовування. Цей кеш робить повторні запити та випадкові подвійні кліки ефективними.
• Кожен повільний крок проходить через run_async – невелику, але важливу утиліту:

@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 та build — це всі блокуючі, синхронні функції, які виконують роботу з мережею, файлами та інтенсивними обчисленнями на CPU. run_async переміщує кожну з них на окрему робочу нитку за допомогою asyncio.to_thread, тож повільне завантаження не зупиняє однонитковий цикл перезапуску Streamlit, а st.status продовжує оновлюватися.

Вхідні дані для чату відповідають стандартній схемі Streamlit, причому генерація тексту передається до 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})

Крок 3: завантаження репозиторію

Замість створення власного сканера GitHub додаток використовує gitingest, який завантажує репозиторій та перетворює його на один формат Markdown, що зберігає межі файлів, структуру каталогів та їхній вміст. Логіка реалізована у файлі 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()

Існують два основні варіанти. По-перше, content.md зберігається на диску як побічний ефект. Це не основний шлях передачі даних, але це зручно для виправлення помилок, оскільки його відкриття показує саме те, що бачить модель. По-друге, load_documents використовує TemporaryDirectory. SimpleDirectoryReader від LlamaIndex потребує каталогу, а не сирого тексту, тому тимчасова папка є найкращим способом передачі вмісту у пам’яті без залишення файлів. Папка та її вміст зникають, коли закінчується блок with.

Крок 4: двигун RAG

Ядром проекту є rag_engine.py, тому варто розглядати його детально. RAGEngineFactory керує трьома окремими аспектами: мовною моделлю, моделлю ембеддингів та індексом та запитом, які поєднують процес пошуку з генерацією.

Мовна модель

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

Декоратор @st.cache_resource виконує справжню роботу тут. Streamlit перезапускає весь скрипт зверху вниз після кожної взаємодії, тож без нього кожен клік чи натискання клавіши призводило б до перескладання клієнта Groq. cache_resource наказує Streamlit створити об’єкт один раз та повернути той самий пізніше, що є правильним форматом для API-клієнтів та завантажених моделей. st.cache_data є аналогом для серіалізованих даних.

Модель ембеддингу

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

Модель BAAI/bge-large-en-v1.5 працює локально через HuggingFace, без звернень до мережі та без оплати за кожен токен. Вона перетворює кожен фрагмент репозиторія, а потім кожне запитання, на вектор у семантичному просторі так, щоб подібні за значенням елементи знаходилися поруч навіть у разі різниці в формулюванні. Це дозволяє запитанню на кшталт «Як тут працює автентифікація?» відображати фрагмент, який обговорює входження, сеанси чи JWT, хоча жодне з цих слів не зустрічається в самому запитанні.

Той самий принцип кешування застосовується й тут. Завантаження моделі об’ємом близько 1,3 ГБ при кожному запуску зробило б додаток непридатним для використання, тому її завантажують один раз та використовують знову.

Індекс та запит

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

Ось що відбувається під час виконання build():

1. Settings.llm та Settings.embed_model є глобальною конфігурацією LlamaIndex. Кожен наступний компонент, від створення індексу до механізму пошуку та двигуна запитів, автоматично бере їх, тому нічого не потрібно передавати з кожним викликом.
2. transformations=[MarkdownNodeParser()] визначає стратегію розділення на частини. Якби флатній Markdown-текст розділяли за кількістю символів, це могло б розрізати функцію навпіл. MarkdownNodeParser замість цього розділяє текст за заголовками та рамками коду, тож кожна частинка є цілісною одиницею: розділом файлу, функцією чи поясненням.
3. VectorStoreIndex.from_documents(...) інкорпорує кожну частинку за допомогою налаштованої моделі та зберігає вектори в пам’яті для пошуку за косинусовою схожістю з вектором запиту.

index.as_query_engine(streaming=True) перетворює індекс на двигун, який вбудовує запит, знаходить найкраще підходящі фрагменти, об’єднує їх у запрошення до моделі та передає відповідь моделі по токену, замість того щоб чекати на повну відповідь.

query_engine.update_prompts(...) замінює стандартний шаблон формулювання запитів та відповідей LlamaIndex на вищезгаданий користувацький. Для правильної роботи це найважливіший елемент системи.

Чому саме запрошення є справжнім захистом

Без жодних прямих обмежень мовна модель з радістю надасть плавну, впевнену відповідь щодо коду, якого зовсім немає у отриманому контексті, адже саме створення правдоподібних продовжень є основною функцією таких моделей, на відміну від перевірених фактів. Шаблон закриває цю прогалину одним правилом: якщо отриманий контекст не містить відповіді, відповідь має бути простим твердженням про те, що відповідь невідома.

Це перетворює галюцинації з типової поведінки моделі на правило, якого їй наказано дотримуватися, і на практиці саме це визначає різницю між інструментом, якому можна довіряти, та тим, що впевнено вигадує речі. Це невеликий фрагмент тексту, проте він більше сприяє придатності проекту, ніж усе обладнання для пошуку інформації навколо нього.

Крок 5: стрімування з синхронної бібліотеки

response.response_gen у LlamaIndex — це звичайний синхронний генератор, і його ітерація блокує потік до моменту, поки не буде готовий наступний елемент. У той же час інтерфейс чату Streamlit хоче асинхронно очікувати частини даних, щоб залишатися реактивним. AsyncResponseStreamer у файлі response_streamer.py поєднує ці два підходи:

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)

Важливо пам’ятати, що next(response_iterator, end_marker) є сам по собі блокуючим, оскільки він може чекати на мережі, поки Groq генерує наступний токен. Тому він також виконується на робочій нитці за допомогою asyncio.to_thread. Використання спеціального об’єкта-сентинела замість перевірки на None усуває приховану помилку: легітимний фрагмент у теорії може бути порожнім рядком або чимось хибним, а сентинел гарантує, що цикл зупиниться лише тоді, коли ітератор справді вичерпаний.

Цю схему можна використовувати щоразу, коли синхронна бібліотека AI, яка включає більшість з них, працює через асинхронний або потоковий інтерфейс: кожен блокуючий виклик потрібно передавати на окрему нитку, а результати потрібно отримувати поступово, коли бібліотека надає генератор.

Оркестратор використовує його у функції 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())

Знак курсора, який відображається під час стрімування, — це невеликий елемент, який створює враження, ніби текст вводиться в реальному часі, а не надходить у блоках.

Крок 6: стан сеансу

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

Три елементи зберігаються у стані сесії. Один з них — історія чату, інший — словник готових двигунів запитів, індексований за допомогою хешу URL, а останній — це URL, який наразі обраний. Хешування URL замість прямого використання його як ключа забезпечує фіксовану, передбачувану довжину ключів та уникає проблем з ескейпуванням символів слешів, крапок та параметрів запиту.

reset_chat() очищує лише messages, залишаючи repositories недоторканими, тож очищення розмови не змушує завантажувати та знову вбудовувати репозиторій. Пряме викликання gc.collect() враховує те, що довга історія чату разом із великим індексом у пам’яті можуть займати значну кількість пам’яті, тож примусове збирання вільної пам’яті дозволяє її швидко звільнити, замість того щоб чекати на звичайний цикл Python.

Одне запитання у кожному файлі

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

Кожен модуль виконує лише одну функцію, що забезпечує досить складну систему обробки даних із асинхронним поєднанням процесів, кешуванням, стрімінгом та циклом RAG із обмеженнями щодо швидкості виконання; вся структура займає менше 200 рядків у шести файлах.

Відомі обмеження

• Лише пам’ять. Жодні дані не зберігаються після перезапуску. Логічним способом покращення є зберігання векторного індексу на диску чи у сервісах типу Chroma або Qdrant, щоб репозиторії не доводилося заново завантажувати щоразу під час сеансу.
• Один активний репозиторій на сеанс. Кілька репозиторіїв можуть знаходитися в кеші, проте чат працює лише з одним з них за раз.
• Лише публічні репозиторії. gitingest працює з публічними URL-адресами GitHub, а для приватних репозиторіїв знадобиться обробка автентифікації.

Немає автоматизованих тестів. Це прийнятно для особистого інструменту, хоча їх слід додати ще до того, як інструмент буде вважатися продакшн-готовим.

Усі компоненти — Streamlit разом із LlamaIndex, Groq та ембеддингами від HuggingFace — коштують майже нічого для невеликих репозиторіїв завдяки безкоштовному тарифу Groq та використанню локальних ембеддингів. Робота з кодовою базою є цікавим початковим проектом типу RAG, адже результати можна перевірити за чимось, що вже зрозуміле: знання того, що робить код, дозволяє легко виявити неправильну відповідь.