Чат с любым репозиторием 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 — это все блокирующие, синхронные функции, обрабатывающие задачи, связанные с сетью, файлами и ресурсоемкой обработкой данных. 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 позволяет избежать скрытой ошибки: законный фрагмент данных теоретически может быть пустой строкой или чем-то ложным, а сентинел гарантирует, что цикл прекратится только тогда, когда итератор действительно исчерпан.

Этот подход можно использовать в любом случае, когда синхронная библиотека ИИ, которых большинство, работает через асинхронный или потоковый интерфейс: все блокирующие вызовы отправляются в отдельный поток, а результаты генерируются постепенно каждый раз, когда библиотека предоставляет генератор.

Оркестратор обрабатывает данные в функции 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, поскольку результаты можно проверить на соответствие уже известной информации: понимание функций кода позволяет легко обнаружить неверные ответы.