Chatea con cualquier repositorio de GitHub: Streamlit, LlamaIndex, Groq y embeddings locales
Construir un chatbot RAG sobre un repositorio de GitHub, que incluya la ingesta de gitingest, el fragmentado en Markdown, modelos en caché, un prompt que reconozca la ignorancia y transmisión asíncrona en Streamlit.
Fuentes: Building a RAG Chatbot That Lets You Chat With Any GitHub Repository (Streamlit + LlamaIndex + Groq)
Encontrarse a gusto en un código base desconocido suele llevar horas: hacer clic en los archivos, adivinar la arquitectura, volver a leer las mismas funciones. Una forma más rápida es utilizar una pequeña aplicación en la que se ingresa una URL de GitHub, se formulan preguntas sobre el código y las respuestas aparecen a medida que se generan. A continuación se muestra la implementación completa: cómo se conectan los módulos, qué hace el motor RAG al recibir una consulta, una técnica de hilos que permite que una biblioteca que bloquea el flujo funcione sin problemas dentro de Streamlit, y la redacción del prompt que evita que el modelo invente respuestas.
Qué hace la aplicación
Se trata de una aplicación de Streamlit con un flujo sencillo:
- Se pega la URL de un repositorio de GitHub.
- La aplicación descarga e indexa el repositorio.
- Las preguntas se formulan a través de una interfaz de chat.
Se trata del patrón clásico de generación mejorada por recuperación de información, aplicado al código fuente y no a documentos en prosa. La tecnología utilizada es Streamlit para la interfaz, LlamaIndex para el indexado y las búsquedas, Groq como modelo de lenguaje, además de embeddings de HuggingFace que se ejecutan localmente, mientras que gitingest se encarga de la descarga.
Disposición del proyecto
.
├── 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)
La ruta de los datos:
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
Paso 1: un punto de entrada muy pequeño
Todo comienza en app.py, mantenido deliberadamente pequeño. Sus únicas funciones son cargar las variables de entorno y pasar el control a la clase de la aplicación:
from streamlit_app import GitHubRAGApp
from dotenv import load_dotenv
if __name__ == "__main__":
load_dotenv()
GitHubRAGApp().run()
load_dotenv() lee un archivo local .env y expone sus valores como variables de entorno, lo que mantiene los secretos fuera del código fuente:
GROQ_API_KEY=your_groq_api_key_here
Comience con:
streamlit run app.py
Mantener este archivo mínimo es más importante de lo que parece. Dado que es el único lugar donde realmente se inicia Streamlit, todos los demás módulos pueden probarse de forma unitaria sin el entorno en tiempo de ejecución de Streamlit.
Paso 2: el orquestador
GitHubRAGApp, en streamlit_app.py, es el coordinador. Él gestiona a tres colaboradores y conecta la interfaz con ellos:
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}")
Varios detalles merecen atención:
st.status(..., expanded=True)muestra un registro de progreso en tiempo real en lugar de un spinner. El proceso consta de tres etapas lentas: descarga, análisis e incrustación, por lo que la visibilidad mejora considerablemente la experiencia.- La carga es idempotente. Si la clave de hash del repositorio ya se encuentra en
st.session_state.repositories, la aplicación pasa directamente a los datos ya cargados y omite la descarga e incrustación. Ese caché hace que las consultas repetidas y los clics dobles accidentales sean poco costosos. - Cada paso lento pasa por
run_async, una utilidad pequeña pero importante:
@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 y build son funciones síncronas y bloqueantes que se encargan de tareas intensivas en red, archivos y procesamiento por CPU. run_async traslada cada una de ellas a un hilo de trabajo mediante asyncio.to_thread, de modo que una descarga lenta no bloquee el bucle de reejecución single-threaded de Streamlit y st.status siga actualizándose.
La entrada del chat sigue el patrón estándar de Streamlit, siendo la generación de respuestas gestionada por 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})
Paso 3: ingestión del repositorio
En lugar de escribir un rastreador personalizado para GitHub, la aplicación se basa en gitingest, el cual descarga un repositorio y lo convierte en una representación en Markdown que mantiene los límites de los archivos, la estructura de directorios y su contenido. La lógica se encuentra en 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()
Hay dos opciones destacables. Primero, content.md se guarda en el disco como efecto secundario. No es la vía principal por la que pasan los datos, pero resulta útil para depurar, ya que al abrirlo se muestra exactamente lo que ve el modelo. Segundo, load_documents utiliza un TemporaryDirectory. SimpleDirectoryReader de LlamaIndex necesita un directorio y no texto en bruto, por lo que una carpeta temporal es la forma más adecuada de pasar contenido en memoria sin dejar archivos. La carpeta y su contenido desaparecen cuando finaliza el bloque with.
Paso 4: el motor RAG
El núcleo del proyecto es rag_engine.py, por lo que vale la pena analizarlo con calma. RAGEngineFactory se encarga de tres aspectos separados: el modelo de lenguaje, el modelo de embedding y el índice junto con la instrucción que vinculan la recuperación de información con la generación.
El modelo de lenguaje
@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",
)
El decorador @st.cache_resource es el que realiza el trabajo real aquí. Streamlit vuelve a ejecutar todo el script de arriba hacia abajo en cada interacción, por lo que sin él cada clic o tecla presionada reconstruiría el cliente Groq. cache_resource indica a Streamlit que construya el objeto una sola vez y que devuelva el mismo posteriormente, lo cual es la forma adecuada para clientes de API y modelos cargados. st.cache_data es el equivalente para datos serializables.
El modelo de incrustación
@staticmethod
@st.cache_resource
def get_embedding_model() -> HuggingFaceEmbedding:
return HuggingFaceEmbedding(
model_name="BAAI/bge-large-en-v1.5",
trust_remote_code=True,
)
El modelo BAAI/bge-large-en-v1.5 se ejecuta localmente a través de HuggingFace, sin necesidad de conexiones de red ni costos por token. Convierte cada fragmento del repositorio, y posteriormente cada pregunta, en un vector ubicado en el espacio semántico, de modo que los significados similares queden cerca unos de otros incluso cuando la redacción difiere. Eso permite que una pregunta como “¿cómo funciona la autenticación aquí?” muestre un fragmento que hable de inicio de sesión, sesiones o JWT, aunque ninguna de esas palabras aparezca en la pregunta.
Se aplica el mismo razonamiento de caché. Cargar un modelo de aproximadamente 1,3 GB en cada ejecución volvería la aplicación inutilizable, por lo que se carga una sola vez y se reutiliza.
El índice y la 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
Esto es lo que sucede cuando se ejecuta build():
Settings.llmySettings.embed_modelson la configuración global de LlamaIndex. Cada componente posterior, desde la construcción del índice hasta el recuperador y el motor de consultas, los utiliza automáticamente, por lo que no es necesario pasarlos en cada llamada.transformations=[MarkdownNodeParser()]establece la estrategia de particionamiento. Dividir el Markdown plano según el conteo de caracteres podría cortar una función por la mitad. En cambio,MarkdownNodeParserdivide el texto siguiendo los encabezados y los bloques de código, de modo que cada fragmento suele ser una unidad coherente: una sección de archivo, una función o una explicación.VectorStoreIndex.from_documents(...)incrusta cada fragmento con el modelo configurado y mantiene los vectores en memoria, listos para búsquedas de similitud coseno frente al vector de una pregunta.
index.as_query_engine(streaming=True) convierte el índice en un motor que integra la pregunta, recupera los fragmentos que mejor se ajustan, los agrupa en un prompt y transmite los tokens de la respuesta del modelo uno por uno en lugar de esperar a obtener toda la respuesta.query_engine.update_prompts(...) reemplaza el plantilla predeterminado de LlamaIndex para responder preguntas por el personalizado mencionado anteriormente. Para garantizar la precisión, este es el elemento más importante del sistema.Por qué el prompt es la verdadera protección
Sin ninguna restricción explícita, un modelo de lenguaje producirá con facilidad una respuesta fluida y segura sobre código que no aparece en el contexto recuperado, ya que su función es generar continuaciones plausibles, no hechos verificados. La plantilla cierra esa brecha con una sola regla: cuando el contexto recuperado no contiene la respuesta, la respuesta debe ser una declaración simple de que dicha respuesta es desconocida.
Eso convierte la alucinación, que antes era el comportamiento por defecto del modelo, en una regla que se le ha indicado seguir; en la práctica, esa es la diferencia entre una herramienta digna de confianza y otra que inventa cosas con total seguridad. Se trata de un pequeño fragmento de texto, pero contribuye más a que el proyecto sea utilizable que todo el mecanismo de recuperación que lo rodea.
Paso 5: transmisión en tiempo real desde una biblioteca síncrona
response.response_gen de LlamaIndex es un generador síncrono ordinario, y al iterarlo se bloquea el hilo hasta que esté listo el siguiente fragmento. Mientras tanto, la interfaz de chat de Streamlit desea esperar los fragmentos de forma asíncrona para mantenerse receptiva. AsyncResponseStreamer en response_streamer.py conecta ambos enfoques:
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)
Lo importante de recordar es que next(response_iterator, end_marker) es en sí mismo una llamada bloqueante, ya que puede esperar en la red mientras Groq genera el siguiente token. Por lo tanto, también se ejecuta en un hilo de trabajo a través de asyncio.to_thread. Utilizar un objeto centinela dedicado en lugar de verificar None evita un error sutil: en teoría, un bloque legítimo podría ser una cadena vacía o algo falso, y el centinela garantiza que el bucle se detenga solo cuando el iterador esté realmente agotado.
Este patrón es reutilizable siempre que una biblioteca de IA síncrona, que incluye a la mayoría de ellas, esté detrás de una interfaz asíncrona o de streaming: se envían todas las llamadas bloqueantes a un hilo y se avanzan de forma incremental cada vez que la biblioteca expone un generador.
El orquestador lo consume en 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())
El carácter de cursor que aparece al transmitir es un detalle sutil que hace que la salida parezca estar siendo escrita en tiempo real y no simplemente llegando por bloques.
Paso 6: estado de la sesión
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()
Tres elementos se encuentran en el estado de sesión. Uno es el historial de chat, otro es un diccionario de motores de consulta listos indexados mediante el hash de la URL, y el último es la URL que se haya seleccionado actualmente. Al hacer un hash de la URL, en lugar de usarla directamente como clave, se mantiene una longitud fija y predecible para las claves, además de evitar problemas derivados de barras invertidas, puntos y parámetros de consulta.
reset_chat() solo elimina los messages y deja intactos los repositories; por lo tanto, borrar una conversación no obliga a descargar e incrustar nuevamente el repositorio. La llamada explícita a gc.collect() reconoce que un historial de chat largo junto con un índice grande en memoria pueden ocupar una cantidad significativa de memoria, y al forzar su recolección se libera rápidamente en lugar de esperar al ciclo normal de Python.
Una sola pregunta en cada archivo
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
Cada módulo tiene exactamente una función, lo que permite mantener un pipeline bastante sofisticado con conexión asíncrona, caché, transmisión en flujo y un bucle RAG restringido por tiempo de respuesta, todo ello con menos de 200 líneas distribuidas en seis archivos.
Límites conocidos
- Solo memoria. Nada se mantiene entre reinicios. Una mejora natural sería guardar el índice vectorial en disco o en un almacenamiento como Chroma o Qdrant, de modo que no sea necesario volver a procesar los repositorios en cada sesión.
- Un solo repositorio activo por sesión. Se pueden tener varios repositorios en el caché, pero el chat solo trabaja con uno a la vez.
- Solo repositorios públicos.
gitingestfunciona con URLs públicas de GitHub; los repositorios privados requerirían manejo de autenticación.
Toda la infraestructura, Streamlit con LlamaIndex, Groq y embeddings de HuggingFace, cuesta casi nada para repositorios pequeños, gracias a la versión gratuita de Groq y a los embeddings locales. Conversar con un código es un primer proyecto RAG satisfactorio, ya que los resultados se pueden verificar contra algo ya comprendido: saber qué hace el código facilita detectar respuestas incorrectas.
Lecturas relacionadas
- RAG en Python sencillo: fragmentación, TF-IDF, similitud cosénica y formulación de prompts — Construye generación mejorada por recuperación de información únicamente con el módulo matemático: fragmentación mediante ventana deslizante, vectores TF-IDF, clasificación por similitud cosénica, un prompt basado en hechos reales y sus limitaciones.
- Construye un asistente de chat RAG para Wikipedia con LlamaIndex y ReAct — Indexa páginas de Wikipedia según sea necesario, incrusta los fragmentos con un pequeño codificador y permite que un agente ReAct invoque una herramienta de motor de consultas para que las respuestas de Chainlit estén basadas en evidencia recuperada.