AI 创作日记 | DeepSeek 构建知识图谱实战，从零搭建新零售商品关系网络的5个关键步骤

class DataIntegrator:
    def __init__(self):
        # 多协议数据源连接
        self.sources = {
            'transaction': KafkaConsumer('pos_logs', 
                auto_offset_reset='latest',
                consumer_timeout_ms=1000),
            'inventory': RESTClient('http://wms/api',
                retries=3,
                timeout=5),
            'user_behavior': WebSocket('/user-events',
                ping_interval=30)
        }

    def _normalize(self, raw_data):
        """多模态数据标准化（关键路径）"""
        # 货币单位统一（支持12种货币转换）
        normalized = {}
        if raw_data.get('currency'):
            normalized['amount'] = self._convert_currency(
                raw_data['amount'],
                raw_data['currency']
            )
        
        # 时间维度统一（处理6种时间格式）
        ts = self._parse_timestamp(raw_data['timestamp'])
        normalized['timestamp'] = ts.isoformat()
        
        # 地理坐标转换（WGS84 -> GCJ02）
        if 'location' in raw_data:
            normalized['coord'] = coordinate_transform(
                raw_data['location'],
                src='WGS84',
                dst='GCJ02'
            )
        
        return {**raw_data, **normalized}

    def generate_triples(self):
        """实时三元组生成流水线"""
        while True:
            # 多源数据并行获取
            batch = []
            for source in self.sources.values():
                try:
                    msgs = source.poll(timeout=100)
                    batch.extend([self._process(msg) for msg in msgs])
                except Exception as e:
                    logger.error(f"数据源{source}异常: {str(e)}")
            
            # 时空去重（5维去重策略）
            deduped = self._deduplicate(batch)
            
            # 生成三元组（支持动态schema）
            for data in deduped:
                yield (
                    data['subject'],
                    data['predicate'],
                    data['object']
                )

def _deduplicate(self, batch):
    """五维去重策略"""
    dedup_cache = {}
    for item in batch:
        # 构建复合键：时间+空间+主体+事件类型+数据源
        key = (
            item['timestamp'][:10],  # 日期级
            round(item['coord'][0],2),  # 0.01度精度
            item['subject'],
            item['event_type'],
            item['source']
        )
        if key not in dedup_cache:
            dedup_cache[key] = item
    return list(dedup_cache.values())

class DynamicNER:
    def __init__(self):
        # 初始化基础模型
        self.nlp = spacy.load("zh_core_web_lg")
        
        # 动态实体存储
        self.entity_cache = LRUCache(max_size=10000)
        self.entity_db = RedisConnection()
        
        # 增量训练配置
        self.retrain_interval = 3600  # 每小时重训练
        self.last_retrain = time.time()
        
        # 上下文分析模型
        self.context_analyzer = BertForSequenceClassification.from_pretrained(
            "bert-base-chinese"
        )
    
    def detect_entities(self, text):
        """动态实体识别主流程"""
        # 基础识别
        doc = self.nlp(text)
        entities = []
        
        # 动态识别
        for ent in doc.ents:
            if self._is_unknown_entity(ent):
                # 上下文语义分析
                context_score = self._analyze_context(text, ent)
                
                # 动态实体缓存
                if context_score > 0.7:
                    self._cache_entity(ent)
                    entities.append({
                        'text': ent.text,
                        'label': 'DYNAMIC',
                        'confidence': context_score
                    })
            else:
                entities.append(ent)
        
        # 增量训练触发
        if time.time() - self.last_retrain > self.retrain_interval:
            self._incremental_train()
        
        return entities
    
    def _is_unknown_entity(self, entity):
        """判断是否为未知实体"""
        return entity.label_ not in ['ORG', 'PER', 'LOC']  # 排除基础实体类型
    
    def _analyze_context(self, text, entity):
        """上下文语义分析"""
        inputs = self.context_tokenizer(
            text, 
            return_tensors="pt",
            padding=True,
            truncation=True
        )
        outputs = self.context_analyzer(**inputs)
        return torch.sigmoid(outputs.logits).item()
    
    def _cache_entity(self, entity):
        """缓存动态实体"""
        key = f"entity:{hash(entity.text)}"
        self.entity_cache.set(key, entity.text)
        self.entity_db.sadd("dynamic_entities", entity.text)
    
    def _incremental_train(self):
        """增量训练模型"""
        new_data = self._prepare_training_data()
        self.nlp.update(
            new_data,
            drop=0.2,  # 防止灾难性遗忘
            sgd=self.optimizer,
            losses=self.losses
        )
        self.last_retrain = time.time()

def _analyze_context(self, text, entity):
    """改进的上下文分析算法"""
    # 构建上下文窗口
    start = max(0, entity.start_char - 20)
    end = min(len(text), entity.end_char + 20)
    context_window = text[start:end]
    
    # 提取语义特征
    features = [
        self._get_semantic_role(entity.text, context_window),
        self._get_syntactic_pattern(entity.text, context_window),
        self._get_cooccurrence_freq(entity.text)
    ]
    
    # 集成学习预测
    return self.ensemble_model.predict(features)

def _get_semantic_role(self, entity, context):
    """语义角色分析"""
    roles = {
        'agent': 0.3,
        'patient': 0.5,
        'location': 0.8
    }
    return roles.get(parse_semantic_role(entity, context), 0)

def _get_cooccurrence_freq(self, entity):
    """共现频率分析"""
    total = self.entity_db.get(f"count:{entity}")
    return math.log(total + 1) if total else 0

class IncrementalTrainer:
    def __init__(self, model):
        self.model = model
        self.memory_buffer = deque(maxlen=1000)  # 记忆缓冲区
        self.optimizer = AdamW(model.parameters(), lr=5e-5)
    
    def update(self, new_data):
        # 弹性权重巩固（防止灾难性遗忘）
        fisher_matrix = self._calculate_fisher()
        
        # 混合训练数据
        training_data = self.memory_buffer + new_data
        
        # 自定义损失函数
        loss = self._ewc_loss(fisher_matrix)
        loss += F.cross_entropy(outputs, labels)
        
        # 参数更新
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()
        
        # 更新记忆缓冲区
        self.memory_buffer.extend(new_data)
    
    def _calculate_fisher(self):
        # 计算Fisher信息矩阵
        pass

class ContextAwareRE:
    def __init__(self):
        # 初始化预训练模型
        self.tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese")
        self.model = AutoModelForSequenceClassification.from_pretrained(
            "bert-base-chinese",
            num_labels=10  # 关系类型数
        )
        
        # 上下文缓存
        self.context_cache = LRUCache(maxsize=1000)
        
        # 上下文分析器
        self.context_analyzer = ContextGraph()
    
    def extract_relations(self, text, entities):
        """上下文感知关系抽取主流程"""
        # 生成候选关系对
        pairs = self._generate_pairs(entities)
        
        relations = []
        for head, tail in pairs:
            # 获取上下文特征
            local_ctx = self._get_local_context(text, head, tail)
            global_ctx = self._get_global_context(head, tail)
            
            # 多粒度特征融合
            combined = self._fuse_features(local_ctx, global_ctx)
            
            # 关系分类
            logits = self.model(**combined).logits
            pred = torch.argmax(logits, dim=-1)
            
            if pred != 'NO_RELATION':
                relations.append({
                    'head': head,
                    'tail': tail,
                    'relation': pred,
                    'confidence': torch.max(F.softmax(logits, dim=-1)).item()
                })
        
        return relations
    
    def _get_local_context(self, text, head, tail):
        """提取局部上下文"""
        # 构建上下文窗口
        start = max(0, head['start'] - 20)
        end = min(len(text), tail['end'] + 20)
        window = text[start:end]
        
        # 编码局部上下文
        inputs = self.tokenizer(
            window,
            return_tensors="pt",
            padding='max_length',
            truncation=True,
            max_length=128
        )
        return inputs
    
    def _get_global_context(self, head, tail):
        """提取全局上下文"""
        # 从知识图谱获取关联实体
        graph_ctx = self.context_analyzer.query(
            f"""
            MATCH (h)-[r*..2]-(t)
            WHERE h.id = '{head['id']}' AND t.id = '{tail['id']}'
            RETURN r
            """
        )
        
        # 从缓存获取历史交互
        cache_key = f"{head['id']}-{tail['id']}"
        history_ctx = self.context_cache.get(cache_key, [])
        
        return {
            'graph': graph_ctx,
            'history': history_ctx
        }
    
    def _fuse_features(self, local, global_ctx):
        """特征融合模块"""
        # 局部特征
        local_features = self.model.bert(**local).last_hidden_state[:,0,:]
        
        # 全局图特征
        graph_emb = self._encode_graph(global_ctx['graph'])
        
        # 历史特征
        history_emb = self._encode_history(global_ctx['history'])
        
        # 动态权重融合
        combined = self.attention_fusion(
            local_features, 
            graph_emb, 
            history_emb
        )
        return combined

def _encode_graph(self, graph_data):
    """图结构编码"""
    # 使用GCN编码子图
    node_embeddings = self.gcn(graph_data)
    return torch.mean(node_embeddings, dim=0)

def _encode_history(self, history):
    """时序上下文编码"""
    # 使用LSTM编码历史交互
    embeddings = [self._get_embedding(h) for h in history]
    lstm_out, _ = self.lstm(torch.stack(embeddings))
    return lstm_out[-1]

def attention_fusion(self, *features):
    """基于注意力的特征融合"""
    query = torch.cat(features, dim=-1)
    keys = self.fusion_linear(query)
    weights = F.softmax(keys, dim=-1)
    return torch.sum(weights * query, dim=0)

class AdaptiveCache:
    def __init__(self, max_size=1000):
        self.cache = OrderedDict()
        self.max_size = max_size
        self.access_counter = defaultdict(int)
    
    def get(self, key):
        if key in self.cache:
            self.access_counter[key] += 1
            self.cache.move_to_end(key)
            return self.cache[key]
        return None
    
    def set(self, key, value):
        if len(self.cache) >= self.max_size:
            # 淘汰策略：访问频率+最近使用
            candidates = list(self.cache.keys())[:self.max_size//10]
            evict_key = min(candidates, 
                key=lambda k: self.access_counter[k])
            del self.cache[evict_key]
            del self.access_counter[evict_key]
        
        self.cache[key] = value
        self.access_counter[key] = 0

class SpatioTemporalWeighter:
    def __init__(self):
        # 基础权重配置
        self.base_weights = {
            'purchase': 0.85,    # 购买行为
            'view': 0.35,        # 浏览行为  
            'combo': 0.7,        # 组合购买
            'search': 0.25,      # 搜索行为
            'add_to_cart': 0.5   # 加购行为
        }
        
        # 时间衰减参数
        self.decay_rates = {
            'short_term': 0.15,  # 短期衰减
            'mid_term': 0.08,    # 中期衰减
            'long_term': 0.03    # 长期衰减
        }
        
        # 空间增强参数
        self.spatial_factors = {
            'same_warehouse': 1.25,    # 同仓
            'same_region': 1.1,        # 同区域
            'cross_region': 0.9        # 跨区域
        }
    
    def calculate_weight(self, relation_type, context):
        """时空动态权重计算"""
        # 基础权重校验
        if relation_type not in self.base_weights:
            raise ValueError(f"未知关系类型: {relation_type}")
        
        # 时间衰减计算
        time_decay = self._calculate_time_decay(context['timestamp'])
        
        # 空间增强计算
        spatial_boost = self._calculate_spatial_boost(
            context['product1'], 
            context['product2']
        )
        
        # 综合权重计算
        base = self.base_weights[relation_type]
        return base * time_decay * spatial_boost
    
    def _calculate_time_decay(self, event_time):
        """动态时间衰减因子"""
        days_diff = (datetime.now() - event_time).days
        if days_diff <= 7:
            rate = self.decay_rates['short_term']
        elif days_diff <= 30:
            rate = self.decay_rates['mid_term']
        else:
            rate = self.decay_rates['long_term']
        
        return math.exp(-rate * days_diff)
    
    def _calculate_spatial_boost(self, p1, p2):
        """多级空间增强"""
        warehouse_match = same_warehouse(p1, p2)
        region_match = same_region(p1, p2)
        
        if warehouse_match:
            return self.spatial_factors['same_warehouse']
        elif region_match:
            return self.spatial_factors['same_region']
        else:
            return self.spatial_factors['cross_region']

def _calculate_time_decay(self, event_time):
    """改进的时间衰减函数"""
    hours_diff = (datetime.now() - event_time).total_seconds() / 3600
    
    # 分段衰减策略
    if hours_diff < 24:  # 24小时内
        return 1.0 - 0.15 * (hours_diff / 24)
    elif hours_diff < 168:  # 7天内
        return 0.85 * math.exp(-0.05 * (hours_diff - 24)/24)
    else:  # 超过7天
        return 0.6 * math.exp(-0.02 * (hours_diff - 168)/24)

class SpatialRelationGraph:
    def __init__(self):
        self.graph = nx.Graph()
        self._build_initial_graph()
    
    def _build_initial_graph(self):
        # 加载仓库拓扑数据
        warehouses = Warehouse.objects.all()
        for wh in warehouses:
            self.graph.add_node(wh.id, type='warehouse', region=wh.region)
        
        # 构建运输路线
        routes = TransportRoute.objects.all()
        for route in routes:
            self.graph.add_edge(
                route.source.id,
                route.dest.id,
                weight=route.transit_time
            )
    
    def spatial_proximity(self, p1, p2):
        """计算空间亲密度"""
        try:
            path = nx.shortest_path(
                self.graph, 
                source=p1.warehouse.id,
                target=p2.warehouse.id,
                weight='weight'
            )
            return 1 / (len(path) ** 0.5)
        except nx.NetworkXNoPath:
            return 0

class EnhancedRealTimeInferencer:
    def __init__(self, kg):
        # 知识图谱连接池
        self.kg_pool = KnowledgeGraphPool(kg, max_connections=10)
        
        # 多级缓存系统
        self.cache = TieredCache(
            fast_cache=LRUCache(10000),  # 内存缓存
            slow_cache=RedisCache()      # Redis缓存
        )
        
        # 上下文处理器
        self.context_processor = ContextAnalyzer()
    
    def recommend(self, user_id):
        """增强版实时推荐"""
        # 获取增强上下文
        context = self._get_enhanced_context(user_id)
        
        # 缓存检查
        cache_key = f"rec:{user_id}:{hash(str(context))}"
        if cached := self.cache.get(cache_key):
            return cached
        
        # 动态构建查询
        query = self._build_query(user_id, context)
        
        # 执行图谱查询
        with self.kg_pool.get_connection() as conn:
            results = conn.execute_query(query)
        
        # 后处理结果
        processed = self._post_process(results, context)
        
        # 缓存结果
        self.cache.set(cache_key, processed, ttl=60)
        
        return processed

    def _build_query(self, user_id, context):
        """动态查询构建"""
        base_query = """
        MATCH (u:User {{id: '{user_id}'}})-[r1:{rels}]->(p1:Product)
        WHERE {context_filters}
        WITH u, p1, r1 ORDER BY r1.timestamp DESC LIMIT 50
        MATCH (p1)-[r2:{relation_types}]->(p2:Product)
        WHERE datetime() < r2.valid_period
          AND {spatial_condition}
        RETURN p2.sku, sum(r1.weight * r2.weight * {context_weights}) as score
        ORDER BY score DESC LIMIT {limit}
        """
        
        return base_query.format(
            user_id=user_id,
            rels=self._get_relevant_relations(context),
            context_filters=self._build_filters(context),
            relation_types="|".join(['combo', 'substitute', 'cross_sell']),
            spatial_condition=self._get_spatial_condition(context),
            context_weights=self._calculate_context_weights(context),
            limit=5 + int(context.get('premium', False)*3)  # 付费用户增加推荐数量
        )

def _get_relevant_relations(self, context):
    """基于上下文的关联关系选择"""
    time_based = {
        'morning': ['view', 'search'],
        'evening': ['purchase', 'add_to_cart']
    }.get(context['time_period'], ['view', 'purchase'])
    
    location_based = ['store_view'] if context['in_store'] else []
    
    return list(set(time_based + location_based))

def _get_spatial_condition(self, context):
    """空间过滤条件生成"""
    if context.get('store_id'):
        return f"p2.available_stores CONTAINS '{context['store_id']}'"
    return "1=1"  # 无空间限制

class TieredCache:
    def __init__(self, fast_cache, slow_cache):
        self.fast = fast_cache
        self.slow = slow_cache
    
    def get(self, key):
        # 先查快速缓存
        if value := self.fast.get(key):
            return value
        # 再查慢速缓存
        if value := self.slow.get(key):
            self.fast.set(key, value)
            return value
        return None
    
    def set(self, key, value, ttl):
        # 双写策略
        self.fast.set(key, value)
        self.slow.set(key, value, ex=ttl)

from py2neo import Graph

# 连接到Neo4j数据库
graph = Graph("bolt://localhost:7687", auth=("neo4j", "password"))

# 定义用户感兴趣的商品
product_name = 'Apple iPhone 14 Pro Max'

# 查询相关商品
query = f"MATCH (p:product {{name: '{product_name}'}})-[:belongs_to]->(c:category)<-[:belongs_to]-(related:product) WHERE related.name <> '{product_name}' RETURN related.name"
result = graph.run(query)

# 输出推荐商品
for record in result:
    print(record[0])

from py2neo import Graph

# 连接到Neo4j数据库
graph = Graph("bolt://localhost:7687", auth=("neo4j", "password"))

# 定义用户查询
query = "Apple products"

# 执行查询
result = graph.run(f"MATCH (p:product)-[:belongs_to]->(b:brand {{name: 'Apple'}}) RETURN p.name")

# 输出查询结果
for record in result:
    print(record[0])