企业官网建设流程全解析

自己设计家装的免费设计软件,seo好的外贸网站,网站设计一般用什么软件,网页设计应用论文描述前言 假设要找处理用户输入的函数。用 grep 只能搜关键词#xff1a;user_input、request.body、form.data…但代码里可能用的是 sanitize_data()、validate_params()#xff0c;这些函数名跟用户输入没有字面关系#xff0c;grep 找不到。 RAG 的…前言假设要找处理用户输入的函数。用 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_factorylist) calls: List[str] field(default_factorylist) dependencies: List[str] field(default_factorylist) definitions: List[str] field(default_factorylist) # 安全相关 security_indicators: List[str] field(default_factorylist) # 元数据 metadata: Dict[str, Any] field(default_factorydict) # 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), (rpassword\s*, password_assign), (rsecret\s*, secret_assign), (rapi_key\s*, api_key_assign), ], javascript: [ (r\beval\s*\(, eval), (r\bFunction\s*\(, function_constructor), (rinnerHTML\s*, innerHTML), (routerHTML\s*, outerHTML), (rdocument\.write\s*\(, document_write), (r\.exec\s*\(, exec), (r\.query\s*\(.*\, sql_concat), (rpassword\s*[:], password_assign), (rapiKey\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(timeout60) as client: response await client.post(url, headersheaders, jsonpayload) 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, fEmbedding 提供商变更: {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, fEmbedding 模型变更: {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, fEmbedding 维度变更: {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( pathself.persist_directory, settingsSettings(anonymized_telemetryFalse), ) 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( nameself.collection_name, metadatacollection_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_resultsn_results, wherewhere, 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_embeddingquery_embedding, n_resultstop_k * 2, # 多查一些后面过滤 wherewhere 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(keylambda x: x.score, reverseTrue) 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_ktop_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 ffunction definition {function_name} definitions await self.retrieve( definition_query, top_k5, filter_file_pathfile_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 fcalls {function_name} invoke {function_name} callers await self.retrieve(caller_query, top_ktop_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(ffunction {call} definition, top_k2) context[callees].extend(callees) return context追踪函数调用链分析数据流用得上。Agent 怎么用 RAG审计任务初始化时会创建索引器和检索器# backend/app/api/v1/endpoints/agent_tasks.py # 创建索引器 indexer CodeIndexer( collection_namefproject_{project_id}, embedding_serviceembedding_service, ) # 索引项目代码 async for progress in indexer.smart_index_directory( directoryproject_root, exclude_patternsexclude_patterns or [], include_patternstarget_files, update_modeIndexUpdateMode.SMART, embedding_progress_callbackon_embedding_progress, cancel_checkcancel_check, ): await emit(f索引进度: {progress.processed_files}/{progress.total_files}) # 创建检索器 retriever CodeRetriever(collection_name, embedding_service)RAG 工具的分配有讲究。Recon 只有rag_queryAnalysis 有全部三个# 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 什么时候选 RAGSystem 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, paramsNone): 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时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】