可与任何 GitHub 仓库聊天：Streamlit、LlamaIndex、Groq 以及本地嵌入模型

在陌生的代码库中熟悉环境通常需要数小时：逐一查看文件、猜测架构结构、反复阅读相同的函数。更快捷的方法是使用一个小应用，输入 GitHub 链接后即可提出关于代码的问题，系统会在生成答案后立即反馈。以下是完整的实现方案：各模块之间的连接方式、RAG 引擎在查询时的工作原理、一种能让阻塞型库在 Streamlit 中流畅运行的线程技术，以及用于防止模型编造答案的提示词写法。

该应用的功能

这是一款流程简单的 Streamlit 应用：

1. 输入 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

第一步：一个极小的入口点

一切始于刻意保持简洁的 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：协调器

streamlit_app.py 中的 GitHubRAGApp 即为协调器。它管理着三个协作组件，并将界面与它们相连：

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。LlamaIndex的SimpleDirectoryReader需要的是目录而非原始文本，因此使用临时文件夹是在不留下文件的情况下传递内存中内容的最佳方式。当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 GB大小的模型，应用将无法使用，因此该模型只会被加载一次并重复使用。

索引与提示词

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步：从同步库中流式获取数据

LlamaIndex的response.response_gen是一个普通的同步生成器，遍历它时会阻塞线程直到下一部分数据准备好。而Streamlit的聊天界面则希望以异步方式等待数据块，从而保持响应能力。response_streamer.py中的AsyncResponseStreamer起到了连接两者的作用：

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只能处理公开的GitHub地址，而私有存储库则需要额外的身份验证机制。

没有自动化测试。对于个人工具来说可以接受，但若要将其视为生产级产品，则必须添加自动化测试。

得益于 Groq 的免费套餐以及本地嵌入技术，对于小型代码库而言，使用 Streamlit 结合 LlamaIndex、Groq 与 HuggingFace 嵌入模型的整个技术栈几乎不需要任何成本。与代码库进行对话是一个令人满意的初始 RAG 项目，因为其结果可以与已有的认知内容对照验证：既然清楚代码的功能，就能很容易地发现错误答案。