一、要做成什么样子的工具目标很简单使用地址https://ai.multikb.com在浏览器里输入一个问题比如「用户注册流程在哪些文件里实现的」「支付模块的主要类和调用关系」「这个项目如何本地启动和部署」网页会在整个代码仓库里检索相关代码片段/文档RAG把这些片段交给本地 Qwen3.5用中文给你一段结构化回答并展示引用到的代码段架构示意浏览器团队成员│ Streamlit 页面▼IncrementalCodeRAG我们写的 RAG 类│├─ ChromaDB存向量└─ Ollama本地 Embedding Qwen3.5 问答特点全本地、离线可用只要能访问部署机多人可同时访问支持全量索引 增量索引专门为「问代码仓库」设计的 UI二、环境准备2.1 目录结构新建一个工程目录例如streamlit_code_ragstreamlit_code_rag/├── main.py # Streamlit Web 应用├── code_rag_incremental.py # 代码仓库 RAG 核心类含增量索引├── requirements.txt # 依赖└── README.md # 简单说明可选2.2 Python 依赖requirements.txtstreamlit chromadb ollama python-dotenv tqdm glob2创建虚拟环境并安装cd streamlit_code_rag python -m venv venv # Windows: # venv\Scripts\activate # macOS / Linux: source venv/bin/activate pip install -r requirements.txt2.3 安装 Ollama 模型# 安装 OllamamacOS/Linux curl -fsSL https://ollama.com/install.sh | sh # Windows 从官网下载安装包https://ollama.com/download # 拉取向量模型Embedding ollama pull nomic-embed-text # 拉取问答模型Qwen3.5 指令模型 ollama pull qwen3.5:7b-instruct-q4_0确保 Ollama 服务在运行一般安装后自动常驻ollama list能看到刚拉下来的模型即可。三、核心后端支持增量更新的代码 RAG 类创建code_rag_incremental.py这是整个工具的「大脑」负责遍历仓库、构建向量索引、按问题检索代码块并调用 Qwen3.5 回答。下面代码已适配 Streamlit用到了st.info / st.error / st.progress做简单 UI 反馈# code_rag_incremental.py import os import json import hashlib import datetime import glob from typing import List, Tuple, Dict, Optional import chromadb import ollama from tqdm import tqdm import streamlit as st class IncrementalCodeRAG: 支持增量更新的代码知识库 RAG - full_ingest(): 全量构建索引初次或重建用 - incremental_ingest(): 增量更新只处理变更文件 - ask(): 检索 调 Qwen3.5 回答 def __init__(self, repo_path: str, db_path: str ./code_rag_db, state_file: str ./index_state.json, embedding_model: str nomic-embed-text, chat_model: str qwen3.5:7b-instruct-q4_0, patterns: Optional[List[str]] None): self.repo_path os.path.abspath(repo_path) self.db_path db_path self.state_file state_file self.embedding_model embedding_model self.chat_model chat_model self.patterns patterns or [*.py, *.md, *.js, *.ts, *.java, *.go, *.html, *.css] # 初始化向量库 self.client chromadb.PersistentClient(pathdb_path) self.collection self.client.get_or_create_collection( namecode_collection, metadata{hnsw:space: cosine} ) # 加载/初始化状态 self._load_state() # 状态管理 def _load_state(self): if os.path.exists(self.state_file): try: with open(self.state_file, r, encodingutf-8) as f: self.state json.load(f) except Exception as e: st.warning(f加载状态文件失败: {e}) self.state { last_build_time: datetime.datetime.now().isoformat(), file_hashes: {} } else: self.state { last_build_time: datetime.datetime.now().isoformat(), file_hashes: {} } self._save_state() def _save_state(self): with open(self.state_file, w, encodingutf-8) as f: json.dump(self.state, f, ensure_asciiFalse, indent2) def _file_sha256(self, path: str) - str: try: h hashlib.sha256() with open(path, rb) as f: for chunk in iter(lambda: f.read(8192), b): h.update(chunk) return h.hexdigest() except Exception: return # IO 切分 def _read_file(self, file_path: str) - str: try: with open(file_path, r, encodingutf-8) as f: return f.read() except Exception as e: st.error(f✗ 读取失败: {file_path} - {e}) return def _split_chunks(self, text: str, max_len: int 800) - List[str]: 简单按字符长度切分后面可替换成“按函数/类切分”版本 lines text.split(\n) chunks, cur, cur_len [], [], 0 for line in lines: l len(line) if cur_len l max_len and cur: chunks.append(\n.join(cur)) cur, cur_len [], 0 cur.append(line) cur_len l if cur: chunks.append(\n.join(cur)) return chunks def _iter_files(self) - List[str]: all_files [] for pat in self.patterns: all_files.extend( glob.glob(os.path.join(self.repo_path, **, pat), recursiveTrue) ) return all_files # 索引状态 def get_index_stats(self) - Dict: try: result self.client.get_or_create_collection(namecode_collection) ids result.get().get(ids, []) return { total_chunks: len(ids), total_files: len(self.state.get(file_hashes, {})), last_updated: self.state.get(last_build_time, N/A) } except Exception: return { total_chunks: 0, total_files: 0, last_updated: 未索引 } # 全量索引 def full_ingest(self): st.info(f 开始全量索引{self.repo_path}) # 清空旧数据 existing self.collection.get() if existing.get(ids): self.collection.delete(idsexisting[ids]) files self._iter_files() st.success(f✓ 发现 {len(files)} 个文件开始切分 向量化) all_ids, all_docs, all_metas [], [], [] file_hashes: Dict[str, str] {} progress_bar st.progress(0.0) for idx, file_path in enumerate(files): rel os.path.relpath(file_path, self.repo_path) content self._read_file(file_path) if not content.strip(): continue chunks self._split_chunks(content) file_hashes[rel] self._file_sha256(file_path) for chunk_idx, chunk in enumerate(chunks): doc_id f{rel}::{chunk_idx} all_ids.append(doc_id) all_docs.append(chunk) all_metas.append({file_path: rel, chunk_index: chunk_idx}) progress_bar.progress((idx 1) / max(len(files), 1)) if all_docs: st.info(⏱ 正在向量化并存储...) for i in range(0, len(all_docs), 64): batch_docs all_docs[i:i 64] resp ollama.embeddings(modelself.embedding_model, promptbatch_docs) embeddings resp[embeddings] if embeddings in resp else resp self.collection.add( idsall_ids[i:i 64], documentsbatch_docs, metadatasall_metas[i:i 64], embeddingsembeddings, ) self.state[last_build_time] datetime.datetime.now().isoformat() self.state[file_hashes] file_hashes self._save_state() st.success(✅ 全量索引完成) return self.get_index_stats() # 增量索引 def _scan_changes(self): files self._iter_files() old_hashes: Dict[str, str] self.state.get(file_hashes, {}) current_hashes: Dict[str, str] {} changed_abs: List[str] [] deleted_rel: List[str] [] # 现有文件 for file_path in files: rel os.path.relpath(file_path, self.repo_path) h self._file_sha256(file_path) current_hashes[rel] h if (rel not in old_hashes) or (old_hashes[rel] ! h): changed_abs.append(file_path) # 已删除文件 for rel in list(old_hashes.keys()): abs_path os.path.join(self.repo_path, rel) if not os.path.exists(abs_path): deleted_rel.append(rel) return changed_abs, deleted_rel, current_hashes def incremental_ingest(self): st.info( 开始增量索引...) changed_abs, deleted_rel, current_hashes self._scan_changes() st.success(f✓ 变更文件{len(changed_abs)} 个已删除文件{len(deleted_rel)} 个) # 删除已删除文件对应的向量 if deleted_rel: to_delete_ids [] existing self.collection.get() for _id, meta in zip(existing[ids], existing[metadatas]): if meta.get(file_path) in deleted_rel: to_delete_ids.append(_id) if to_delete_ids: self.collection.delete(idsto_delete_ids) st.info(f✓ 已删除 {len(to_delete_ids)} 个向量块) # 重新索引变更文件 all_ids, all_docs, all_metas [], [], [] for file_path in changed_abs: rel os.path.relpath(file_path, self.repo_path) content self._read_file(file_path) if not content.strip(): continue chunks self._split_chunks(content) # 删除旧块 existing self.collection.get() del_ids [ _id for _id, meta in zip(existing[ids], existing[metadatas]) if meta.get(file_path) rel ] if del_ids: self.collection.delete(idsdel_ids) for idx, chunk in enumerate(chunks): doc_id f{rel}::{idx} all_ids.append(doc_id) all_docs.append(chunk) all_metas.append({file_path: rel, chunk_index: idx}) if all_docs: st.info(f⏱ 正在向量化 {len(all_docs)} 个变更的代码块...) for i in range(0, len(all_docs), 64): batch_docs all_docs[i:i 64] resp ollama.embeddings(modelself.embedding_model, promptbatch_docs) embeddings resp[embeddings] if embeddings in resp else resp self.collection.add( idsall_ids[i:i 64], documentsbatch_docs, metadatasall_metas[i:i 64], embeddingsembeddings, ) st.success(f✓ 已更新 {len(all_docs)} 个向量块) self.state[last_build_time] datetime.datetime.now().isoformat() for rel, h in current_hashes.items(): self.state[file_hashes][rel] h for rel in deleted_rel: self.state[file_hashes].pop(rel, None) self._save_state() st.success(✅ 增量索引完成) return self.get_index_stats() # 检索 问答 def query(self, question: str, top_k: int 5): try: q_emb ollama.embeddings(modelself.embedding_model, promptquestion) # 不同版本返回结构略不同这里兼容两种 if embedding in q_emb: q_vec q_emb[embedding] else: q_vec q_emb[embeddings][0] res self.collection.query( query_embeddings[q_vec], n_resultstop_k, ) docs res[documents][0] metas res[metadatas][0] scores res[distances][0] return [ {text: d, meta: m, score: s} for d, m, s in zip(docs, metas, scores) ] except Exception as e: st.error(f检索失败: {e}) return [] def ask(self, question: str, top_k: int 5, max_context_chars: int 3000) - str: if not question.strip(): return 请提出一个具体的问题。 candidates self.query(question, top_ktop_k) if not candidates: return 未能找到相关代码片段。请检查索引是否已正确建立。 context_parts [] for item in candidates: fp item[meta][file_path] idx item[meta][chunk_index] sc item[score] context_parts.append( f--- 文件: {fp} 片段#{idx} (相似度: {sc:.3f}) ---\n{item[text]} ) context \n\n.join(context_parts) if len(context) max_context_chars: context context[:max_context_chars] \n...\n[上下文已截断] prompt f 你是一个资深软件工程师现在有一个代码仓库的部分片段如下 {context} 用户问题是 {question} 请你基于这些代码片段给出尽量准确的回答 1. 用中文说明相关实现大致逻辑 2. 如果涉及具体函数/类请指明所在文件及大致位置 3. 如有不确定的地方要说明不确定不要瞎编 4. 可以适当给出重构或优化建议。 try: st.info(⏱ 正在生成回答...) resp ollama.chat( modelself.chat_model, messages[{role: user, content: prompt}], temperature0.2, ) return resp[message][content].strip() except Exception as e: st.error(f调用大模型失败: {e}) return f无法生成回答: {e}四、前端Streamlit 网页主程序创建main.py提供 3 个 Tab「状态」看索引状态 一键全量/增量索引「问答」输入问题 → 返回回答 参考代码「配置」调整向量库路径、状态文件重置等# main.py import streamlit as st import os from code_rag_incremental import IncrementalCodeRAG st.set_page_config( page_title代码仓库问答助手, page_icon, layoutwide, initial_sidebar_stateexpanded ) st.title( 代码仓库问答助手) st.caption(用 Streamlit Ollama RAG 搭建团队内部小工具) # Session 状态 if rag not in st.session_state: st.session_state.rag None if repo_path not in st.session_state: st.session_state.repo_path if db_path not in st.session_state: st.session_state.db_path ./code_rag_db if state_file not in st.session_state: st.session_state.state_file ./index_state.json if tab not in st.session_state: st.session_state.tab status # 侧边栏 with st.sidebar: st.header(️ 配置) repo_path st.text_input( 代码仓库路径, valuest.session_state.repo_path or ./your_repo, placeholder请输入代码仓库路径 ) st.session_state.repo_path repo_path embedding_model st.selectbox( Embedding 模型, [nomic-embed-text], index0 ) chat_model st.selectbox( 问答模型, [qwen3.5:7b-instruct-q4_0, qwen2.5:7b-instruct, llama3:8b, deepseek-coder:6.7b], index0 ) file_types st.multiselect( 索引文件类型, [.py, .js, .ts, .java, .go, .md, .html, .css], default[.py, .md, .js, .ts] ) patterns [f*{ft} for ft in file_types] st.divider() c1, c2, c3 st.columns(3) with c1: if st.button( 状态): st.session_state.tab status with c2: if st.button( 问答): st.session_state.tab chat with c3: if st.button(⚙️ 配置): st.session_state.tab config st.divider() st.caption(Powered by Ollama Qwen3.5) st.caption(2026) # 初始化 RAG if st.session_state.repo_path and st.session_state.repo_path ! ./your_repo: need_reinit ( st.session_state.rag is None or st.session_state.rag.repo_path ! os.path.abspath(st.session_state.repo_path) ) if need_reinit: try: st.session_state.rag IncrementalCodeRAG( repo_pathst.session_state.repo_path, db_pathst.session_state.db_path, state_filest.session_state.state_file, embedding_modelembedding_model, chat_modelchat_model, patternspatterns ) st.success(f✅ 已初始化仓库: {os.path.basename(st.session_state.repo_path)}) except Exception as e: st.error(f初始化失败: {e}) def show_status_tab(): st.subheader( 索引状态) if st.session_state.rag is None: st.warning(请在左侧配置有效的代码仓库路径) return stats st.session_state.rag.get_index_stats() st.info(f最后更新时间{stats[last_updated]}) st.info(f已索引文件数{stats[total_files]}) st.info(f已索引代码块数{stats[total_chunks]}) col1, col2 st.columns(2) with col1: if st.button( 增量索引): st.session_state.rag.incremental_ingest() st.rerun() with col2: if st.button( 全量索引): st.session_state.rag.full_ingest() st.rerun() def show_chat_tab(): st.subheader( 问答界面) if st.session_state.rag is None: st.warning(请先配置仓库路径并完成索引) return stats st.session_state.rag.get_index_stats() if stats[total_files] 0: st.warning(当前仓库尚未索引请先在“状态”页进行索引) return question st.text_area( 你的问题, placeholder例如用户注册逻辑在哪些文件中主要步骤是什么, height150 ) if st.button( 提问): if not question.strip(): st.warning(请输入问题) return with st.spinner(正在分析代码仓库...): answer st.session_state.rag.ask(question) st.subheader( 回答) st.markdown(answer) with st.expander( 参考代码片段): cands st.session_state.rag.query(question, top_k3) for i, item in enumerate(cands): st.caption(f文件 {i1}: {item[meta][file_path]} (相似度: {item[score]:.3f})) st.code(item[text], languageplaintext) def show_config_tab(): st.subheader(⚙️ 高级配置) db_path st.text_input( 向量数据库路径, valuest.session_state.db_path, placeholder./code_rag_db ) st.session_state.db_path db_path state_file st.text_input( 状态文件路径, valuest.session_state.state_file, placeholder./index_state.json ) st.session_state.state_file state_file if st.button( 删除状态文件并重置): if os.path.exists(state_file): os.remove(state_file) st.info(状态文件已删除下次操作会重新初始化索引状态) else: st.info(状态文件不存在无需删除) # 路由 if st.session_state.tab status: show_status_tab() elif st.session_state.tab chat: show_chat_tab() elif st.session_state.tab config: show_config_tab() st.divider() st.caption( 示例问题) st.caption(- 这个项目的整体架构是什么) st.caption(- 用户认证和鉴权逻辑在哪些文件) st.caption(- 支付模块的主要类和调用关系) st.caption(请确保 Ollama 已启动并提前拉取所需模型。)五、启动和使用5.1 启动应用在工程目录下运行streamlit run main.py浏览器会打开http://localhost:8501如果是部署在服务器上团队成员可通过http://服务器IP:8501访问。5.2 第一次使用步骤左侧填入代码仓库路径比如text复制/home/yourname/projects/awesome_project确认文件类型列表先用默认的.py, .md, .js, .ts即可切到 状态」Tab 点击「 全量索引」索引完成后切到「问答」Tab开始提问5.3 推荐提问示例「用户注册的流程在哪些文件涉及哪些接口」「这个项目一共有几个微服务分别负责什么」「支付模块中异常处理逻辑有哪些可以优化的点」「请帮我总结下 README 和 docs 里的部署步骤。」六、团队内部落地建议加到 Git 提交流程可选​在仓库的.git/hooks/pre-commit脚本中调用incremental_ingest()保证每次提交前知识库自动更新。部署在一台内网服务器Linux 服务器上跑streamlit run main.py --server.headless true --server.port 8501通过内网域名映射如http://code-rag.internal:8501权限 范围控制只对内部仓库使用避免将敏感代码对外开放如有需要可在外层加简单的登录Nginx Basic Auth 或公司统一登录