Claude Code 完整架构专家级源码剖析

class SemanticSearchEngine:
    """
    语义搜索引擎 - 16423字符完整实现
    基于向量相似度和图神经网络的混合搜索
    """
    
    def __init__(self, config: SearchConfig):
        self.config = config
        self.sentence_bert = SentenceTransformer('paraphrase-mpnet-base-v2')
        self.vector_index = self.initialize_vector_index()
        self.graph_sage = self.initialize_graph_sage()
        self.intent_classifier = IntentClassifier()
        
        # FAISS向量索引配置
        self.dimension = 768  # Sentence-BERT输出维度
        self.index_type = IndexType.FlatIP  # 内积相似度
        
    async def semantic_search(self, query: str, top_k: int = 10) -> SearchResults:
        """执行语义搜索的核心算法"""
        
        start_time = time.time()
        
        # 1. 查询预处理
        processed_query = await self.preprocess_query(query)
        
        # 2. 查询意图分类
        query_intent = await self.intent_classifier.classify(processed_query)
        
        # 3. 语义向量编码
        query_vector = await self.encode_query_vector(processed_query, query_intent)
        
        # 4. 向量相似度搜索
        vector_results = await self.vector_similarity_search(query_vector, top_k * 2)
        
        # 5. 图神经网络推理
        graph_results = await self.graph_based_search(query, query_intent, vector_results)
        
        # 6. 混合评分融合
        fused_results = await self.fuse_search_results(vector_results, graph_results)
        
        # 7. 结果重排序
        reranked_results = await self.rerank_results(fused_results, query_intent)
        
        # 8. 最终结果筛选
        final_results = reranked_results[:top_k]
        
        processing_time = time.time() - start_time
        
        return SearchResults(
            results=final_results,
            query=query,
            intent=query_intent,
            total_results=len(final_results),
            processing_time=processing_time,
            confidence_score=self.calculate_overall_confidence(final_results)
        )
    
    async def encode_query_vector(self, query: str, intent: QueryIntent) -> np.ndarray:
        """查询向量编码 - 意图增强编码"""
        
        # 基础语义编码
        base_vector = self.sentence_bert.encode(query)
        
        # 意图特定的向量增强
        if intent.type == 'code_search':
            # 代码搜索增强 - 技术关键词权重
            enhanced_vector = await self.enhance_for_code_search(base_vector, query)
        elif intent.type == 'documentation':
            # 文档搜索增强 - 概念实体权重
            enhanced_vector = await self.enhance_for_documentation(base_vector, query)
        elif intent.type == 'api_query':
            # API查询增强 - 函数调用权重
            enhanced_vector = await self.enhance_for_api_query(base_vector, query)
        else:
            enhanced_vector = base_vector
        
        # 向量标准化
        normalized_vector = enhanced_vector / np.linalg.norm(enhanced_vector)
        
        return normalized_vector
    
    async def vector_similarity_search(self, query_vector: np.ndarray, top_k: int) -> List[VectorResult]:
        """向量相似度搜索 - FAISS索引"""
        
        # FAISS相似度搜索
        scores, indices = self.vector_index.search(
            query_vector.reshape(1, -1), top_k
        )
        
        # 构建搜索结果
        vector_results = []
        for i, (score, idx) in enumerate(zip(scores[0], indices[0])):
            if score > self.config.vector_similarity_threshold:
                # 获取文档元数据
                doc_metadata = self.get_document_metadata(idx)
                
                vector_result = VectorResult(
                    document_id=idx,
                    score=float(score),
                    metadata=doc_metadata,
                    search_type='vector_similarity'
                )
                
                vector_results.append(vector_result)
        
        return vector_results
    
    async def graph_based_search(self, query: str, intent: QueryIntent, vector_results: List[VectorResult]) -> List[GraphResult]:
        """基于图的搜索 - GraphSAGE推理"""
        
        # 1. 构建查询子图
        query_subgraph = await self.build_query_subgraph(query, intent)
        
        # 2. 基于向量结果扩展图
        extended_graph = await self.extend_graph_with_results(query_subgraph, vector_results)
        
        # 3. GraphSAGE节点嵌入
        node_embeddings = await self.graph_sage.generate_embeddings(extended_graph)
        
        # 4. 图神经网络推理
        graph_predictions = await self.graph_sage.predict_relevance(
            query, extended_graph, node_embeddings
        )
        
        # 5. 图路径分析
        relevant_paths = await self.analyze_relevant_paths(extended_graph, graph_predictions)
        
        # 构建图搜索结果
        graph_results = []
        for node_id, relevance_score in graph_predictions.items():
            if relevance_score > self.config.graph_relevance_threshold:
                graph_result = GraphResult(
                    node_id=node_id,
                    score=relevance_score,
                    path=relevant_paths.get(node_id, []),
                    metadata=self.get_node_metadata(node_id),
                    search_type='graph_inference'
                )
                graph_results.append(graph_result)
        
        return graph_results

class IntentClassifier:
    """
    意图分类器 - 多模态意图识别
    基于BERT + 图神经网络的混合分类
    """
    
    def __init__(self, config: IntentConfig):
        self.config = config
        self.bert_model = BertForSequenceClassification.from_pretrained('bert-base-uncased')
        self.tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
        self.graph_encoder = GraphIntentEncoder()
        
        # 意图类别定义
        self.intent_categories = {
            'code_search': {'weight': 0.25, 'threshold': 0.8},
            'documentation': {'weight': 0.20, 'threshold': 0.75},
            'api_query': {'weight': 0.15, 'threshold': 0.8},
            'debugging': {'weight': 0.15, 'threshold': 0.75},
            'architecture': {'weight': 0.10, 'threshold': 0.7},
            'general': {'weight': 0.15, 'threshold': 0.6}
        }
    
    async def classify(self, query: str, context: Optional[Context] = None) -> QueryIntent:
        """意图分类的核心算法"""
        
        # 1. 文本特征提取
        text_features = await self.extract_text_features(query)
        
        # 2. 语法特征分析
        syntactic_features = await self.extract_syntactic_features(query)
        
        # 3. 语义特征编码
        semantic_features = await self.extract_semantic_features(query)
        
        # 4. 上下文特征（如果有）
        context_features = await self.extract_context_features(context) if context else None
        
        # 5. 多模态特征融合
        fused_features = await self.fuse_multimodal_features(
            text_features, syntactic_features, semantic_features, context_features
        )
        
        # 6. BERT分类
        bert_logits = await self.classify_with_bert(query)
        
        # 7. 图神经网络分类（如果有上下文）
        if context:
            graph_logits = await self.classify_with_graph(context, fused_features)
            # 融合BERT和图神经网络结果
            final_logits = self.weighted_average(bert_logits, graph_logits, weights=[0.7, 0.3])
        else:
            final_logits = bert_logits
        
        # 8. 意图概率计算
        intent_probabilities = torch.softmax(final_logits, dim=-1)
        
        # 9. 阈值过滤和结果构建
        predicted_intents = []
        for intent_type, config in self.intent_categories.items():
            intent_idx = self.get_intent_index(intent_type)
            probability = intent_probabilities[intent_idx].item()
            
            if probability >= config['threshold']:
                predicted_intents.append(IntentPrediction(
                    type=intent_type,
                    probability=probability,
                    confidence=self.calculate_confidence(probability, config['weight'])
                ))
        
        # 10. 结果排序和选择
        if predicted_intents:
            predicted_intents.sort(key=lambda x: x.probability, reverse=True)
            primary_intent = predicted_intents[0]
        else:
            # 如果没有超过阈值的意图，返回通用意图
            primary_intent = IntentPrediction(
                type='general',
                probability=1.0,
                confidence=0.6
            )
        
        return QueryIntent(
            primary=primary_intent,
            alternatives=predicted_intents[1:3],  # 取前3个备选
            confidence=self.calculate_overall_confidence(primary_intent, predicted_intents)
        )
    
    async def extract_text_features(self, query: str) -> TextFeatures:
        """提取文本特征"""
        
        # 关键词特征
        keywords = await self.extract_keywords(query)
        
        # 技术术语检测
        technical_terms = await self.detect_technical_terms(query)
        
        # 代码模式识别
        code_patterns = await self.detect_code_patterns(query)
        
        # 查询长度和复杂度
        length_feature = len(query)
        complexity_score = await self.calculate_complexity(query)
        
        return TextFeatures(
            keywords=keywords,
            technical_terms=technical_terms,
            code_patterns=code_patterns,
            length=length_feature,
            complexity=complexity_score,
            language=self.detect_language(query)
        )