前言
假设要找"处理用户输入的函数"。用 grep 只能搜关键词：user_input、request.body、form.data…但代码里可能用的是sanitize_data()、validate_params()，这些函数名跟"用户输入"没有字面关系，grep 找不到。
RAG 的思路不一样：先把代码转成向量，搜索时按语义匹配。LLM 理解"处理用户输入"是什么意思，能找到语义相关的函数。
系统架构
DeepAudit 的 RAG 有四个组件：
源代码 → Splitter → Embeddings → ChromaDB ↓ Retriever → Agent代码位置在backend/app/services/rag/，总共 3600 多行：
splitter.py：代码分割embeddings.py：向量化indexer.py：索引管理retriever.py：检索
Splitter：按语义分割代码
直接按行分割代码会切断函数、类，语义不完整。DeepAudit 用 Tree-sitter 做 AST 解析，按语义单元分割。
支持十种语言：
# backend/app/services/rag/splitter.py DEFINITION_TYPES = { "python": { "class": ["class_definition"], "function": ["function_definition"], "method": ["function_definition"], "import": ["import_statement", "import_from_statement"], }, "javascript": { "class": ["class_declaration", "class"], "function": ["function_declaration", "function", "arrow_function", "method_definition"], "import": ["import_statement"], }, "typescript": { "class": ["class_declaration", "class"], "function": ["function_declaration", "function", "arrow_function", "method_definition"], "interface": ["interface_declaration"], "import": ["import_statement"], }, "java": { "class": ["class_declaration"], "method": ["method_declaration", "constructor_declaration"], "interface": ["interface_declaration"], "import": ["import_declaration"], }, "go": { "struct": ["type_declaration"], "function": ["function_declaration", "method_declaration"], "interface": ["type_declaration"], "import": ["import_declaration"], }, }分割后每个块的数据结构：
# backend/app/services/rag/splitter.py @dataclass class CodeChunk: """代码块""" id: str content: str file_path: str language: str chunk_type: ChunkType # 位置信息 line_start: int = 0 line_end: int = 0 byte_start: int = 0 byte_end: int = 0 # 语义信息 name: Optional[str] = None parent_name: Optional[str] = None signature: Optional[str] = None docstring: Optional[str] = None # AST 信息 ast_type: Optional[str] = None # 关联信息 imports: List[str] = field(default_factory=list) calls: List[str] = field(default_factory=list) dependencies: List[str] = field(default_factory=list) definitions: List[str] = field(default_factory=list) # 安全相关 security_indicators: List[str] = field(default_factory=list) # 元数据 metadata: Dict[str, Any] = field(default_factory=dict) # Token 估算 estimated_tokens: int = 0注意最后一个字段security_indicators，分割时会自动检测安全相关模式：
# backend/app/services/rag/splitter.py SECURITY_PATTERNS = { "python": [ (r"\bexec\s*\(", "exec"), (r"\beval\s*\(", "eval"), (r"\bcompile\s*\(", "compile"), (r"\bos\.system\s*\(", "os_system"), (r"\bsubprocess\.", "subprocess"), (r"\bcursor\.execute\s*\(", "sql_execute"), (r"\.execute\s*\(.*%", "sql_format"), (r"\bpickle\.loads?\s*\(", "pickle"), (r"\byaml\.load\s*\(", "yaml_load"), (r"\brequests?\.", "http_request"), (r"password\s*=", "password_assign"), (r"secret\s*=", "secret_assign"), (r"api_key\s*=", "api_key_assign"), ], "javascript": [ (r"\beval\s*\(", "eval"), (r"\bFunction\s*\(", "function_constructor"), (r"innerHTML\s*=", "innerHTML"), (r"outerHTML\s*=", "outerHTML"), (r"document\.write\s*\(", "document_write"), (r"\.exec\s*\(", "exec"), (r"\.query\s*\(.*\+", "sql_concat"), (r"password\s*[=:]", "password_assign"), (r"apiKey\s*[=:]", "api_key_assign"), ], # ... }在分割阶段就标记高风险代码，后面检索时可以优先返回这些块。
分割完就是向量化。
Embeddings：向量化
把代码块转成向量。支持七个提供商：
# backend/app/services/rag/embeddings.py class OpenAIEmbedding(EmbeddingProvider): """OpenAI 嵌入服务""" MODELS = { "text-embedding-3-small": 1536, "text-embedding-3-large": 3072, "text-embedding-ada-002": 1536, } async def embed_texts(self, texts: List[str]) -> List[EmbeddingResult]: # ... async with httpx.AsyncClient(timeout=60) as client: response = await client.post(url, headers=headers, json=payload) response.raise_for_status() data = response.json() # ... class OllamaEmbedding(EmbeddingProvider): """Ollama 本地嵌入服务""" MODELS = { "nomic-embed-text": 768, "mxbai-embed-large": 1024, "all-minilm": 384, "snowflake-arctic-embed": 1024, "bge-m3": 1024, } # Azure, Cohere, HuggingFace, Jina, Qwen本地部署可以用 Ollama，不用调 API。
代码库大的时候需要批量处理。
批量嵌入
大量代码块需要批量处理，减少 API 调用次数：
# backend/app/services/rag/embeddings.py async def embed_batch( self, texts: List[str], batch_size: int = 100, show_progress: bool = False, progress_callback: Optional[callable] = None, cancel_check: Optional[callable] = None, ) -> List[List[float]]: """批量嵌入文本""" if not texts: return [] embeddings = [] uncached_indices = [] uncached_texts = [] # 检查缓存 for i, text in enumerate(texts): ... if self.cache_enabled: cache_key = self._cache_key(text) if cache_key in self._cache: embeddings.append(self._cache[cache_key]) continue embeddings.append(None) # 占位 uncached_indices.append(i) uncached_texts.append(text) # 批量处理未缓存的文本 if uncached_texts: for i in range(0, len(uncached_texts), batch_size): ... batch = uncached_texts[i:i + batch_size] batch_indices = uncached_indices[i:i + batch_size] results = await self._provider.embed_texts(batch) for idx, result in zip(batch_indices, results): embeddings[idx] = result.embedding ... return embeddings100 个文本一批，避免单次请求太大。
模型变更检测
换嵌入模型后，索引器通过 vector_store 获取配置并检测变更：
# backend/app/services/rag/indexer.py async def _check_rebuild_needed(self) -> Tuple[bool, str]: """检查是否需要重建索引""" # ... stored_config = self.vector_store.get_embedding_config() # 检查 embedding 提供商 stored_provider = stored_config.get("provider") current_provider = self.embedding_config.get("provider") if stored_provider and current_provider and stored_provider != current_provider: return True, f"Embedding 提供商变更: {stored_provider} -> {current_provider}" # 检查 embedding 模型 stored_model = stored_config.get("model") current_model = self.embedding_config.get("model") if stored_model and current_model and stored_model != current_model: return True, f"Embedding 模型变更: {stored_model} -> {current_model}" # 检查维度 stored_dimension = stored_config.get("dimension") current_dimension = self.embedding_config.get("dimension") if stored_dimension and current_dimension and stored_dimension != current_dimension: return True, f"Embedding 维度变更: {stored_dimension} -> {current_dimension}" return False, ""provider + model + dimension 拼成配置信息，通过对比判断是否需要重建索引。
Indexer：智能索引
索引器做三件事：扫描文件、生成向量、存到 ChromaDB。
ChromaDB 向量存储
# backend/app/services/rag/indexer.py class ChromaVectorStore(VectorStore): """Chroma 向量存储""" async def initialize(self, force_recreate: bool = False): """初始化 Chroma""" import chromadb from chromadb.config import Settings if self.persist_directory: self._client = chromadb.PersistentClient( path=self.persist_directory, settings=Settings(anonymized_telemetry=False), ) else: self._client = chromadb.Client(...) # 构建 collection 元数据 collection_metadata = { "hnsw:space": "cosine", # 余弦相似度 "index_version": INDEX_VERSION, "embedding_provider": self.embedding_config.get("provider", "openai"), "embedding_model": self.embedding_config.get("model", "text-embedding-3-small"), "embedding_dimension": self.embedding_config.get("dimension", 1536), } self._collection = self._client.get_or_create_collection( name=self.collection_name, metadata=collection_metadata, ) async def query( self, query_embedding: List[float], n_results: int = 10, where: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: result = await asyncio.to_thread( self._collection.query, query_embeddings=[query_embedding], n_results=n_results, where=where, include=["documents", "metadatas", "distances"], ) return { "ids": result["ids"][0] if result["ids"] else [], "documents": result["documents"][0] if result["documents"] else [], "metadatas": result["metadatas"][0] if result["metadatas"] else [], "distances": result["distances"][0] if result["distances"] else [], }用 HNSW 算法做近似最近邻检索，余弦相似度作为距离度量。
索引进度跟踪
# backend/app/services/rag/indexer.py @dataclass class IndexingProgress: """索引进度""" total_files: int = 0 processed_files: int = 0 total_chunks: int = 0 indexed_chunks: int = 0 current_file: str = "" errors: List[str] = None # 增量更新统计 added_files: int = 0 updated_files: int = 0 deleted_files: int = 0 skipped_files: int = 0 update_mode: str = "full" # 状态消息（用于前端显示） status_message: str = "" @property def progress_percentage(self) -> float: if self.total_files == 0: return 0.0 return (self.processed_files / self.total_files) * 100每处理一个文件就 yield 一次进度，前端可以实时展示。
索引模式
有三种模式：
# backend/app/services/rag/indexer.py class IndexUpdateMode(Enum): """索引更新模式""" FULL = "full" # 全量重建：删除旧索引，完全重新索引 INCREMENTAL = "incremental" # 增量更新：只更新变化的文件 SMART = "smart" # 智能模式：根据情况自动选择SMART 模式会计算文件哈希，对比变更：
# backend/app/services/rag/indexer.py async def _incremental_index(self, directory: str, ...): """增量索引""" # 获取已索引文件的 hash indexed_file_hashes = await self.vector_store.get_file_hashes() indexed_files = set(indexed_file_hashes.keys()) # 收集当前文件 current_files = self._collect_files(directory, exclude_patterns, include_patterns) current_file_map = {} for file_path in current_files: relative_path = os.path.relpath(file_path, directory) current_file_map[relative_path] = file_path current_file_set = set(current_file_map.keys()) # 计算差异 files_to_add = current_file_set - indexed_files files_to_delete = indexed_files - current_file_set files_to_check = current_file_set & indexed_files # 检查需要更新的文件（hash 变化） files_to_update = set() for relative_path in files_to_check: content = await asyncio.to_thread(self._read_file_sync, file_path) current_hash = hashlib.md5(content.encode()).hexdigest() if current_hash != indexed_file_hashes.get(relative_path): files_to_update.add(relative_path) # ...只处理变化的文件，大项目不用每次全量重建。
Retriever：检索
检索器提供三种方式：
1. 语义检索（rag_query）
# backend/app/services/rag/retriever.py async def retrieve( self, query: str, top_k: int = 10, filter_file_path: Optional[str] = None, filter_language: Optional[str] = None, filter_chunk_type: Optional[str] = None, min_score: float = 0.0, ) -> List[RetrievalResult]: """语义检索""" await self.initialize() # 生成查询嵌入 query_embedding = await self.embedding_service.embed(query) # 构建过滤条件 where = {} if filter_file_path: where["file_path"] = filter_file_path if filter_language: where["language"] = filter_language if filter_chunk_type: where["chunk_type"] = filter_chunk_type # 查询向量存储 raw_results = await self.vector_store.query( query_embedding=query_embedding, n_results=top_k * 2, # 多查一些，后面过滤 where=where if where else None, ) # 转换结果 results = [] for id_, doc, meta, dist in zip(...): score = 1 - dist # 将距离转换为相似度分数 if score < min_score: continue result = RetrievalResult(...) results.append(result) results.sort(key=lambda x: x.score, reverse=True) return results[:top_k]就是标准的向量检索。
2. 安全相关检索（security_search）
# backend/app/services/rag/retriever.py async def retrieve_security_related( self, vulnerability_type: Optional[str] = None, top_k: int = 20, ) -> List[RetrievalResult]: """检索与安全相关的代码""" # 根据漏洞类型构建查询 security_queries = { "sql_injection": "SQL query execute database user input", "xss": "HTML render user input innerHTML template", "command_injection": "system exec command shell subprocess", "path_traversal": "file path read open user input", "ssrf": "HTTP request URL user input fetch", "deserialization": "deserialize pickle yaml load object", "auth_bypass": "authentication login password token session", "hardcoded_secret": "password secret key token credential", } if vulnerability_type and vulnerability_type in security_queries: query = security_queries[vulnerability_type] else: query = "security vulnerability dangerous function user input" return await self.retrieve(query, top_k=top_k)先检索，再按安全评分重排。有安全标记的代码块排在前面。
3. 函数上下文检索（function_context）
# backend/app/services/rag/retriever.py async def retrieve_function_context( self, function_name: str, file_path: Optional[str] = None, include_callers: bool = True, include_callees: bool = True, top_k: int = 10, ) -> Dict[str, List[RetrievalResult]]: """检索函数上下文""" context = { "definition": [], "callers": [], "callees": [], } # 查找函数定义 definition_query = f"function definition {function_name}" definitions = await self.retrieve( definition_query, top_k=5, filter_file_path=file_path, ) # 过滤出真正的定义 for result in definitions: if result.name == function_name or function_name in (result.content or ""): context["definition"].append(result) if include_callers: # 查找调用此函数的代码 caller_query = f"calls {function_name} invoke {function_name}" callers = await self.retrieve(caller_query, top_k=top_k) for result in callers: # 检查是否真的调用了这个函数 if re.search(rf'\b{re.escape(function_name)}\s*\(', result.content): if result not in context["definition"]: context["callers"].append(result) if include_callees and context["definition"]: # 从函数定义中提取调用的其他函数 for definition in context["definition"]: calls = re.findall(r'\b(\w+)\s*\(', definition.content) unique_calls = list(set(calls))[:5] # 限制数量 for call in unique_calls: if call == function_name: continue callees = await self.retrieve(f"function {call} definition", top_k=2) context["callees"].extend(callees) return context追踪函数调用链，分析数据流用得上。
Agent 怎么用 RAG
审计任务初始化时，会创建索引器和检索器：
# backend/app/api/v1/endpoints/agent_tasks.py # 创建索引器 indexer = CodeIndexer( collection_name=f"project_{project_id}", embedding_service=embedding_service, ) # 索引项目代码 async for progress in indexer.smart_index_directory( directory=project_root, exclude_patterns=exclude_patterns or [], include_patterns=target_files, update_mode=IndexUpdateMode.SMART, embedding_progress_callback=on_embedding_progress, cancel_check=cancel_check, ): await emit(f"索引进度: {progress.processed_files}/{progress.total_files}") # 创建检索器 retriever = CodeRetriever(collection_name, embedding_service)RAG 工具的分配有讲究。Recon 只有rag_query，Analysis 有全部三个：
# backend/app/services/agent/tools/__init__.py # Recon Agent - 只有通用搜索 recon_tools = { "rag_query": RAGQueryTool(retriever), } # Analysis Agent - 全套 RAG 工具 analysis_tools = { "rag_query": RAGQueryTool(retriever), "security_search": SecurityCodeSearchTool(retriever), "function_context": FunctionContextTool(retriever), }Verification Agent 没有 RAG 工具，它的任务是执行验证，不是搜索代码。
如果 retriever 初始化失败（比如向量数据库没配置），这些工具就不会注册，Agent 会回退到search_code做关键词匹配。
LLM 什么时候选 RAG
System Prompt 明确引导：
🔍 代码搜索工具对比 | 工具 | 特点 | 适用场景 | |------|------|---------| | rag_query | 🔥 语义搜索 | **首选！** 查找"处理用户输入的函数" | | security_search | 🔥 安全专用 | **首选！** 查找"SQL注入相关代码" | | search_code | ❌ 关键词匹配 | **不推荐**，仅查找精确字符串 |几个典型场景：
场景 1：初始代码探索
Thought: 需要找到项目中处理用户认证的代码 Action: rag_query Action Input: {"query": "用户登录认证逻辑", "top_k": 10}返回结果：
找到 10 个相关代码片段: --- 结果 1 (相似度: 0.89) --- 文件: app/views.py 行号: 45-60 名称: login_handler 安全指标: user_input, database_query 代码: def login_handler(request): username = request.POST.get('username') ...场景 2：漏洞定向搜索
Semgrep 报了 SQL 注入警告，需要找更多相关代码：
Thought: 需要查找可能存在 SQL 注入的代码 Action: security_search Action Input: {"vulnerability_type": "sql_injection", "top_k": 20}返回结果会按安全评分重排：
找到 5 个可能与 sql_injection 相关的代码: --- 可疑代码 1 --- 文件: app/models.py:45 ⚠️ 安全指标: database_query, string_concat 代码: cursor.execute("SELECT * FROM users WHERE id = " + user_id)有安全标记的代码块排在前面。
场景 3：数据流追踪
找到可疑函数后，追踪谁调用了它：
Thought: execute_query 可能有 SQL 注入，需要看谁调用了它 Action: function_context Action Input: {"function_name": "execute_query", "include_callers": true}返回结果包含完整的调用链：
函数 'execute_query' 的上下文分析: ### 函数定义: 文件: app/db.py:23 def execute_query(sql, params=None): cursor = conn.cursor() cursor.execute(sql, params) return cursor.fetchall() ### 调用此函数的代码 (3 处): - app/views.py:45 - app/api.py:78 - app/models.py:123 ### 此函数调用的其他函数: - cursor (app/db.py) - fetchall (builtin)方便判断用户输入能不能流到这个函数。
Agent 使用时就像调其他工具一样：
Thought: 需要找数据库查询相关的代码 Action: rag_query Action Input: {"query": "database query execute SQL", "top_k": 15} Thought: 追踪 execute_query 函数的调用链 Action: function_context Action Input: {"function_name": "execute_query"}一些设计亮点
几个设计细节：
- 分割阶段做安全标记：不在检索时才判断，而是在分割阶段就用正则检测打标记，后续检索可以直接过滤或加权。
- 嵌入模型变更检测：换了模型自动重建索引，用配置签名（provider + model + dimension 的哈希）检测变更，不用手动清理。
- 增量更新：大项目几十万行代码全量索引太慢，用文件哈希检测变更，只处理新增、修改、删除的文件。
- 分层检索：不只一种检索方式，通用搜索、安全搜索、函数上下文，不同场景用不同方法。
局限
也有一些局限：
- 没有重排序。Retriever 直接返回向量检索结果，没用 Cross-Encoder 做精排
- 代码跨文件引用分析不够。function_context 只追踪一层调用
- 没有缓存层。相同查询会重复计算
这个 RAG 实现功能比较完整。
小结
- Tree-sitter 按语义分割，保证代码块完整
- 分割时标记安全相关代码
- 支持多种嵌入模型，包括本地部署
- 智能增量索引，检测模型变更
- 三种检索方式适配不同场景
学AI大模型的正确顺序，千万不要搞错了
🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！
有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！
就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋
📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇
学习路线:
✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经
以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！
我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～
)
)
)
)
