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