冲突检测与去重

68class ConflictType(str, Enum):69    """Conflict type enumeration."""7071    VALUE_CONFLICT = "value_conflict"72    TYPE_CONFLICT = "type_conflict"73    RELATIONSHIP_CONFLICT = "relationship_conflict"74    TEMPORAL_CONFLICT = "temporal_conflict"75    LOGICAL_CONFLICT = "logical_conflict"767778@dataclass79class Conflict:80    """Conflict information."""8182    conflict_id: str83    conflict_type: ConflictType84    entity_id: Optional[str] = None85    property_name: Optional[str] = None86    relationship_id: Optional[str] = None87    conflicting_values: List[Any] = field(default_factory=list)88    sources: List[Dict[str, Any]] = field(default_factory=list)89    confidence: float = 1.090    severity: str = "medium"  # low, medium, high, critical91    recommended_action: Optional[str] = None92    metadata: Dict[str, Any] = field(default_factory=dict)

237            for j, (entity_id, entity_list) in enumerate(entity_groups.items()):238                if len(entity_list) < 2:239                    continue  # Need at least 2 sources to have conflict240241                values = []242                sources = []243244                for entity in entity_list:245                    if property_name in entity:246                        value = entity[property_name]247                        values.append(value)248249                        # Track source if available250                        if self.track_provenance:251                            source_ref = SourceReference(252                                document=entity.get("source", "unknown"),253                                page=entity.get("page"),254                                section=entity.get("section"),255                                confidence=entity.get("confidence", 1.0),256                                metadata=entity.get("metadata", {}),257                            )258                            self.source_tracker.track_property_source(259                                entity_id, property_name, value, source_ref260                            )261                            sources.append(262                                {263                                    "document": source_ref.document,264                                    "page": source_ref.page,265                                    "confidence": source_ref.confidence,266                                    "metadata": source_ref.metadata,267                                }268                            )269270                # Check for value conflicts271                unique_values = list(set(str(v) for v in values if v is not None))272273                if len(unique_values) > 1:274                    conflict = Conflict(275                        conflict_id=f"{entity_id}_{property_name}_conflict",276                        conflict_type=ConflictType.VALUE_CONFLICT,277                        entity_id=entity_id,278                        property_name=property_name,279                        conflicting_values=values,280                        sources=sources,281                        confidence=self._calculate_conflict_confidence(values, sources),282                        severity=self._calculate_severity(property_name, values),283                        recommended_action=self._recommend_action(284                            property_name, values285                        ),286                    )287                    conflicts.append(conflict)288                    self.detected_conflicts[conflict.conflict_id] = conflict

558    def _calculate_conflict_confidence(559        self, values: List[Any], sources: List[Dict[str, Any]]560    ) -> float:561        """Calculate confidence score for conflict."""562        if not sources:563            return 0.5564565        # Average confidence of sources566        avg_confidence = sum(s.get("confidence", 0.5) for s in sources) / len(sources)567568        # Higher confidence if values are very different569        value_diversity = (570            len(set(str(v) for v in values)) / len(values) if values else 0571        )572573        return min(1.0, avg_confidence * (1 + value_diversity))574575    def _calculate_severity(self, property_name: str, values: List[Any]) -> str:576        """Calculate conflict severity."""577        # Critical fields578        critical_fields = ["id", "name", "type", "founded_year", "revenue"]579        if property_name.lower() in critical_fields:580            return "critical"581582        # High severity for numeric conflicts with large differences583        try:584            numeric_values = [float(v) for v in values if v is not None]585            if numeric_values:586                value_range = max(numeric_values) - min(numeric_values)587                if value_range > 1000:  # Large difference588                    return "high"589        except (ValueError, TypeError):590            pass591592        return "medium"593594    def _recommend_action(self, property_name: str, values: List[Any]) -> str:595        """Recommend action for conflict."""596        try:597            if len(set(values)) == 2:598                return (599                    "Compare source documents and use most recent or authoritative "600                    "source"601                )602        except TypeError:603            # Handle unhashable types (like dicts or lists)604            # Convert to string representation for set comparison605            str_values = [str(v) for v in values]606            if len(set(str_values)) == 2:607                return (608                    "Compare source documents and use most recent or authoritative "609                    "source"610                )611612        return "Multiple conflicting values detected. Manual review recommended."

1025    def detect_logical_conflicts(1026        self, entities: List[Dict[str, Any]]1027    ) -> List[Conflict]:1028        """1029        Detect logical conflicts (e.g., Person cannot be Organization).10301031        Args:1032            entities: List of entity dictionaries10331034        Returns:1035            List of detected conflicts1036        """1037        tracking_id = self.progress_tracker.start_tracking(1038            file=None,1039            module="conflicts",1040            submodule="ConflictDetector",1041            message=f"Detecting logical conflicts in {len(entities)} entities",1042        )10431044        try:1045            conflicts = []10461047            # Logical rules: incompatible type combinations1048            incompatible_types = {1049                "Person": ["Organization", "Company", "Institution"],1050                "Organization": ["Person"],1051                "Company": ["Person"],1052                "Location": ["Person", "Organization"],1053            }

70class ResolutionStrategy(str, Enum):71    """Conflict resolution strategy."""7273    VOTING = "voting"74    CREDIBILITY_WEIGHTED = "credibility_weighted"75    MOST_RECENT = "most_recent"76    FIRST_SEEN = "first_seen"77    HIGHEST_CONFIDENCE = "highest_confidence"78    MANUAL_REVIEW = "manual_review"79    EXPERT_REVIEW = "expert_review"808182@dataclass83class ResolutionResult:84    """Conflict resolution result."""8586    conflict_id: str87    resolved: bool88    resolved_value: Any = None89    resolution_strategy: Optional[str] = None90    confidence: float = 0.091    sources_used: List[str] = field(default_factory=list)92    resolution_notes: Optional[str] = None93    metadata: Dict[str, Any] = field(default_factory=dict)

214            # Check for property-specific rule if strategy was None215            if strategy is None:216                if conflict.property_name:217                    rule_key = f"{conflict.entity_id}.{conflict.property_name}"218                    if rule_key in self.resolution_rules:219                        normalized_strategy = self.resolution_rules[rule_key]220221            strategy = normalized_strategy222223            self.logger.info(224                (225                    f"Resolving conflict {conflict.conflict_id} using strategy: "226                    f"{strategy.value}"227                )228            )229230            if strategy == ResolutionStrategy.VOTING:231                result = self._resolve_by_voting(conflict)232            elif strategy == ResolutionStrategy.CREDIBILITY_WEIGHTED:233                result = self._resolve_by_credibility(conflict)234            elif strategy == ResolutionStrategy.MOST_RECENT:235                result = self._resolve_by_recency(conflict)236            elif strategy == ResolutionStrategy.FIRST_SEEN:237                result = self._resolve_by_first_seen(conflict)238            elif strategy == ResolutionStrategy.HIGHEST_CONFIDENCE:239                result = self._resolve_by_confidence(conflict)240            elif strategy == ResolutionStrategy.MANUAL_REVIEW:241                result = self._flag_for_manual_review(conflict)242            elif strategy == ResolutionStrategy.EXPERT_REVIEW:243                result = self._flag_for_expert_review(conflict)244            else:245                result = self._resolve_by_voting(conflict)246247            result.resolution_strategy = strategy.value248            conflict_type = (249                conflict.conflict_type.value250                if hasattr(conflict.conflict_type, "value")251                else conflict.conflict_type252            )253            result.metadata.setdefault("conflict_type", conflict_type)254            result.metadata.setdefault("entity_id", conflict.entity_id)255            result.metadata.setdefault("property_name", conflict.property_name)256            result.metadata.setdefault("relationship_id", conflict.relationship_id)257            if conflict.metadata:258                result.metadata.setdefault("conflict_metadata", conflict.metadata)259            self.resolution_history.append(result)

355    def _resolve_by_voting(self, conflict: Conflict) -> ResolutionResult:356        """Resolve conflict by voting (most common value wins)."""357        if not conflict.conflicting_values:358            return ResolutionResult(359                conflict_id=conflict.conflict_id,360                resolved=False,361                resolution_notes="No conflicting values to resolve",362            )363364        # Count value occurrences365        value_counts = Counter(conflict.conflicting_values)366        most_common_value, count = value_counts.most_common(1)[0]367368        # Calculate confidence based on vote ratio369        total_votes = len(conflict.conflicting_values)370        confidence = count / total_votes if total_votes > 0 else 0.0371372        sources_used = [s.get("document", "unknown") for s in conflict.sources]373374        return ResolutionResult(375            conflict_id=conflict.conflict_id,376            resolved=True,377            resolved_value=most_common_value,378            confidence=confidence,379            sources_used=sources_used,380            resolution_notes=(381                f"Resolved by voting: {count}/{total_votes} votes for this value"382            ),383        )384385    def _resolve_by_credibility(self, conflict: Conflict) -> ResolutionResult:386        """Resolve conflict by credibility-weighted voting."""387        if not conflict.conflicting_values or not conflict.sources:388            return ResolutionResult(389                conflict_id=conflict.conflict_id,390                resolved=False,391                resolution_notes="Insufficient data for credibility-based resolution",392            )393394        # Weight values by source credibility395        value_weights: Dict[Any, float] = {}396397        for i, value in enumerate(conflict.conflicting_values):398            source = conflict.sources[i] if i < len(conflict.sources) else {}399            document = source.get("document", "unknown")400            source_confidence = source.get("confidence", 0.5)401            credibility = self.source_tracker.get_source_credibility(document)402403            weight = source_confidence * credibility404405            if value not in value_weights:406                value_weights[value] = 0.0407            value_weights[value] += weight408409        # Get value with highest weight410        if not value_weights:411            return ResolutionResult(412                conflict_id=conflict.conflict_id,413                resolved=False,414                resolution_notes="Could not calculate credibility weights",415            )416417        resolved_value = max(value_weights.items(), key=lambda x: x[1])[0]418        total_weight = sum(value_weights.values())419        confidence = (420            value_weights[resolved_value] / total_weight if total_weight > 0 else 0.0421        )422423        sources_used = [s.get("document", "unknown") for s in conflict.sources]424425        return ResolutionResult(426            conflict_id=conflict.conflict_id,427            resolved=True,428            resolved_value=resolved_value,429            confidence=confidence,430            sources_used=sources_used,431            resolution_notes=(432                "Resolved by credibility-weighted voting "433                f"(weight: {value_weights[resolved_value]:.2f})"434            ),

437    def _resolve_by_recency(self, conflict: Conflict) -> ResolutionResult:438        """Resolve conflict by using most recent value."""439        if not conflict.sources:440            return ResolutionResult(441                conflict_id=conflict.conflict_id,442                resolved=False,443                resolution_notes="No source timestamps available",444            )445446        # Find most recent source (by timestamp if available)447        most_recent_idx = 0448        most_recent_time = None449450        for i, source in enumerate(conflict.sources):451            # Check metadata for timestamp452            timestamp = source.get("metadata", {}).get("timestamp")453            if timestamp:454                if not most_recent_time or timestamp > most_recent_time:455                    most_recent_time = timestamp456                    most_recent_idx = i457458        resolved_value = conflict.conflicting_values[most_recent_idx]459        sources_used = [conflict.sources[most_recent_idx].get("document", "unknown")]460461        return ResolutionResult(462            conflict_id=conflict.conflict_id,463            resolved=True,464            resolved_value=resolved_value,465            confidence=0.8,466            sources_used=sources_used,467            resolution_notes="Resolved by most recent value",468        )

74@dataclass75class SourceReference:76    """Source reference for a property value."""7778    document: str79    page: Optional[int] = None80    section: Optional[str] = None81    line: Optional[int] = None82    timestamp: Optional[datetime] = None83    confidence: float = 1.084    metadata: Dict[str, Any] = field(default_factory=dict)858687@dataclass88class PropertySource:89    """Source information for a property value."""9091    property_name: str92    value: Any93    sources: List[SourceReference] = field(default_factory=list)94    entity_id: Optional[str] = None95    metadata: Dict[str, Any] = field(default_factory=dict)

186    def track_property_source(187        self,188        entity_id: str,189        property_name: str,190        value: Any,191        source: SourceReference,192        **metadata,193    ) -> bool:194        """195        Track source for a property value.196197        Args:198            entity_id: Entity identifier199            property_name: Property name200            value: Property value201            source: Source reference202            **metadata: Additional metadata203204        Returns:205            True if tracking successful206        """207        if entity_id not in self.entity_sources:208            self.entity_sources[entity_id] = {}209210        if property_name not in self.entity_sources[entity_id]:211            self.entity_sources[entity_id][property_name] = PropertySource(212                property_name=property_name,213                value=value,214                entity_id=entity_id,215                metadata=metadata,216            )217218        property_source = self.entity_sources[entity_id][property_name]219220        # Add source if not already present221        source_exists = any(222            s.document == source.document223            and s.page == source.page224            and s.section == source.section225            for s in property_source.sources226        )227228        if not source_exists:229            property_source.sources.append(source)230231        # Update value if different (will be used for conflict detection)232        if property_source.value != value:233            property_source.value = value  # Store latest value234235        self.logger.debug(236            f"Tracked property source: {entity_id}.{property_name} = {value} "237            f"from {source.document}"238        )239        return True

379    def set_source_credibility(self, document: str, credibility: float) -> bool:380        """381        Set credibility score for a source document.382383        Args:384            document: Document identifier385            credibility: Credibility score (0.0 to 1.0)386387        Returns:388            True if set successfully389        """390        if not 0.0 <= credibility <= 1.0:391            raise ValidationError("Credibility must be between 0.0 and 1.0")392393        self.source_credibility[document] = credibility394        self.logger.info(f"Set credibility for {document}: {credibility}")395        return True396397    def get_source_credibility(self, document: str) -> float:398        """399        Get credibility score for a source document.400401        Args:402            document: Document identifier403404        Returns:405            Credibility score (default: 0.5)406        """407        return self.source_credibility.get(document, 0.5)

52@dataclass53class DuplicateCandidate:54    """Duplicate candidate representation."""5556    entity1: Dict[str, Any]57    entity2: Dict[str, Any]58    similarity_score: float59    confidence: float60    reasons: List[str] = field(default_factory=list)61    metadata: Dict[str, Any] = field(default_factory=dict)626364@dataclass65class DuplicateGroup:66    """Group of duplicate entities."""6768    entities: List[Dict[str, Any]]69    similarity_scores: Dict[Tuple[str, str], float] = field(default_factory=dict)70    representative: Optional[Dict[str, Any]] = None71    confidence: float = 0.072    metadata: Dict[str, Any] = field(default_factory=dict)

770        # Check entity type mismatch first: two entities with different771        # explicit types are not duplicates, whatever their similarity.772        entity_type1 = self._get_entity_value(entity1, "type")773        entity_type2 = self._get_entity_value(entity2, "type")774        if entity_type1 and entity_type2 and entity_type1 != entity_type2:775            return DuplicateCandidate(776                entity1=entity1,777                entity2=entity2,778                similarity_score=similarity_score,779                confidence=0.0,780                reasons=["type_mismatch"],781                metadata={782                    "name_match": False,783                    "common_properties": 0,784                    "type_match": False,785                    "type_mismatch": True,786                },787            )788789        # Check for exact name match (strong indicator)790        name1 = str(self._get_entity_value(entity1, "name", "")).lower().strip()791        name2 = str(self._get_entity_value(entity2, "name", "")).lower().strip()792        if name1 == name2 and name1:  # Non-empty exact match793            reasons.append("exact_name_match")794            confidence += 0.1795796        # Check property value matches797        props1 = self._get_entity_value(entity1, "properties", {})798        props2 = self._get_entity_value(entity2, "properties", {})799800        common_props = set(props1.keys()) & set(props2.keys())801        if common_props:802            # Count properties with matching values803            prop_matches = sum(804                1 for prop in common_props if props1.get(prop) == props2.get(prop)805            )806            if prop_matches > 0:807                reasons.append(f"{prop_matches}_property_matches")808                # Boost confidence for each matching property809                confidence += 0.05 * prop_matches810811        # Check entity type match (only boosts when types are equal; mismatch812        # is handled above)813        if entity_type1 and entity_type2 and entity_type1 == entity_type2:814            reasons.append("same_type")815            confidence += 0.05816817        # Cap confidence at 1.0818        confidence = min(1.0, confidence)

833    def _build_duplicate_groups(834        self, candidates: List[DuplicateCandidate]835    ) -> List[DuplicateGroup]:836        """Build duplicate groups from candidates."""837        # Union-find structure838        entity_to_group = {}839        groups = []840841        for candidate in candidates:842            entity1_id = self._normalize_entity_id(candidate.entity1)843            entity2_id = self._normalize_entity_id(candidate.entity2)844845            group1 = entity_to_group.get(entity1_id)846            group2 = entity_to_group.get(entity2_id)847848            if group1 is None and group2 is None:849                # Create new group850                group = DuplicateGroup(851                    entities=[candidate.entity1, candidate.entity2],852                    similarity_scores={853                        (entity1_id, entity2_id): candidate.similarity_score854                    },855                )856                groups.append(group)857                entity_to_group[entity1_id] = group858                entity_to_group[entity2_id] = group859            elif group1 is not None and group2 is None:860                # Add entity2 to group1861                if candidate.entity2 not in group1.entities:862                    group1.entities.append(candidate.entity2)863                group1.similarity_scores[(entity1_id, entity2_id)] = (864                    candidate.similarity_score865                )866                entity_to_group[entity2_id] = group1867            elif group1 is None and group2 is not None:868                # Add entity1 to group2869                if candidate.entity1 not in group2.entities:870                    group2.entities.append(candidate.entity1)871                group2.similarity_scores[(entity1_id, entity2_id)] = (872                    candidate.similarity_score873                )874                entity_to_group[entity1_id] = group2875            elif group1 != group2:876                # Merge groups877                group1.entities.extend(878                    [e for e in group2.entities if e not in group1.entities]879                )880                group1.similarity_scores.update(group2.similarity_scores)881                group1.similarity_scores[(entity1_id, entity2_id)] = (882                    candidate.similarity_score883                )884885                # Update references886                for entity in group2.entities:887                    entity_id = self._normalize_entity_id(entity)888                    entity_to_group[entity_id] = group1889890                if group2 in groups:891                    groups.remove(group2)892893        return groups

54class MergeStrategy(Enum):55    """Merge strategy types."""5657    KEEP_FIRST = "keep_first"58    KEEP_LAST = "keep_last"59    KEEP_MOST_COMPLETE = "keep_most_complete"60    KEEP_HIGHEST_CONFIDENCE = "keep_highest_confidence"61    MERGE_ALL = "merge_all"62    CUSTOM = "custom"636465@dataclass66class PropertyMergeRule:67    """Rule for merging properties."""6869    property_name: str70    strategy: MergeStrategy71    conflict_resolution: Optional[Callable] = None72    priority: int = 0737475@dataclass76class MergeResult:77    """Result of merge operation."""7879    merged_entity: Dict[str, Any]80    merged_entities: List[Dict[str, Any]]81    conflicts: List[Dict[str, Any]] = field(default_factory=list)82    metadata: Dict[str, Any] = field(default_factory=dict)

383    def _select_base_entity(384        self, entities: List[Dict[str, Any]], strategy: MergeStrategy385    ) -> Dict[str, Any]:386        """Select base entity for merge."""387        if strategy == MergeStrategy.KEEP_FIRST:388            return entities[0]389        elif strategy == MergeStrategy.KEEP_LAST:390            return entities[-1]391        elif strategy == MergeStrategy.KEEP_MOST_COMPLETE:392            return max(393                entities,394                key=lambda e: len(e.get("properties", {}))395                + len(e.get("relationships", [])),396            )397        elif strategy == MergeStrategy.KEEP_HIGHEST_CONFIDENCE:398            return max(entities, key=lambda e: e.get("confidence", 0.0))399        else:400            return entities[0]

456    def _resolve_property_conflict(457        self,458        property_name: str,459        value1: Any,460        value2: Any,461        default_strategy: MergeStrategy,462    ) -> Dict[str, Any]:463        """Resolve property conflict."""464        # Check for property-specific rule465        rule = self.property_rules.get(property_name)466467        if rule and rule.conflict_resolution:468            try:469                resolved_value = rule.conflict_resolution(value1, value2)470                return {471                    "resolved": True,472                    "value": resolved_value,473                    "resolution": "custom_rule",474                }475            except Exception as e:476                self.logger.warning(f"Custom conflict resolution failed: {e}")477478        strategy = rule.strategy if rule else default_strategy479480        if strategy == MergeStrategy.KEEP_FIRST:481            return {"resolved": True, "value": value1, "resolution": "keep_first"}482        elif strategy == MergeStrategy.KEEP_LAST:483            return {"resolved": True, "value": value2, "resolution": "keep_last"}484        elif strategy == MergeStrategy.MERGE_ALL:485            # Merge into list if not already486            if isinstance(value1, list):487                if value2 not in value1:488                    value1.append(value2)489                return {"resolved": True, "value": value1, "resolution": "merge_all"}490            else:491                return {492                    "resolved": True,493                    "value": [value1, value2],494                    "resolution": "merge_all",495                }496        else:497            # Default: keep first498            return {"resolved": True, "value": value1, "resolution": "default"}

56@dataclass57class MergeOperation:58    """Entity merge operation representation."""5960    source_entities: List[Dict[str, Any]]61    merged_entity: Dict[str, Any]62    merge_result: MergeResult63    timestamp: Optional[str] = None64    metadata: Dict[str, Any] = None

475    def _add_provenance(476        self, merged_entity: Dict[str, Any], source_entities: List[Dict[str, Any]]477    ) -> Dict[str, Any]:478        """479        Add provenance information to merged entity.480481        This method adds metadata about which entities were merged to create482        the merged entity, preserving the history of the merge operation.483484        Provenance Structure:485            metadata.provenance:486                - merged_from: List of source entity information (id, name, source)487                - merge_count: Number of entities that were merged488489        Args:490            merged_entity: The merged entity dictionary to add provenance to491            source_entities: List of source entities that were merged492493        Returns:494            Merged entity dictionary with provenance information added495        """496        # Ensure metadata structure exists497        if "metadata" not in merged_entity:498            merged_entity["metadata"] = {}499500        if "provenance" not in merged_entity["metadata"]:501            merged_entity["metadata"]["provenance"] = {}502503        provenance = merged_entity["metadata"]["provenance"]504505        # Record source entities506        provenance["merged_from"] = [507            {508                "id": get_entity_id(e),509                "name": self._get_entity_value(e, "name"),510                "source": self._get_entity_value(e, "metadata", {}).get("source") if hasattr(e, "metadata") or isinstance(e, dict) else None,511            }512            for e in source_entities513        ]514        provenance["merge_count"] = len(source_entities)515516        self.logger.debug(517            f"Added provenance for merge of {len(source_entities)} entity(ies)"518        )519520        return merged_entity

9<Info>10Run conflict detection after deduplication and before SHACL validation. Deduplication removes duplicate nodes; conflict resolution reconciles disagreeing property values on the same canonical entity. Running them out of order — detecting conflicts before deduplication — will produce spurious conflicts between entities that should have been merged first.11</Info>1213## What Is Conflict Resolution?1415When you merge data from multiple sources, the same real-world entity — a customer, a product, a threat actor, a drug compound — often appears with contradictory property values. One database says a customer's email is `alice@example.com`; another says `alice.smith@example.com`. One security feed rates a CVE at 10.0; two others rate it 9.1 and 9.5.1617**Conflict resolution** is the systematic process of deciding which value is most trustworthy and recording that decision with evidence, so the canonical entity ends up with one defensible, auditable value per property.1819### Key Concepts2021**Canonical entity** — The single authoritative record for a real-world thing. After deduplication, each entity has exactly one canonical node in your graph. Conflict resolution determines which property values belong on that node.2223**Conflicting values** — Two or more different values asserted for the same property on the same canonical entity, each reported by a different source.2425**Credibility score** — A number between 0.0 and 1.0 you attach to each source record, indicating how reliable that source is. A government registry might carry 0.99; a scraped blog might carry 0.30. You supply these; Semantica uses them during `CREDIBILITY_WEIGHTED` resolution.2627**Confidence score** — A number between 0.0 and 1.0 the resolver *computes* after resolution, reflecting how certain the outcome is. A unanimous vote produces high confidence; a close split among equally credible sources produces lower confidence. This appears on `ResolutionResult.confidence` and should be read as a signal, not a guarantee that the resolved value is correct.2829**Resolution strategy** — The rule for picking the winning value: majority vote, credibility-weighted average, latest timestamp, and so on. See [Resolution strategies at a glance](#resolution-strategies-at-a-glance) for the full list.3031**Audit trail** — The complete record of every resolution decision: conflict ID, strategy used, resolved value, sources consulted, and confidence score. Returned by `resolver.get_resolution_history()`.

38    sample_rows: list[dict[str, Any]] = []39    seen_titles: set[str] = set()40    human_readable_id = 04142    async for row in entities_table:43        title = row.get("title")44        if not title or title in seen_titles:45            continue46        seen_titles.add(title)47        row["degree"] = degree_map.get(title, 0)48        row["human_readable_id"] = human_readable_id49        row["id"] = str(uuid4())50        human_readable_id += 151        out = {col: row.get(col) for col in ENTITIES_FINAL_COLUMNS}52        await entities_table.write(out)53        if len(sample_rows) < 5:54            sample_rows.append(out)5556    return sample_rows

104def _merge_entities(entity_dfs) -> pd.DataFrame:105    all_entities = pd.concat(entity_dfs, ignore_index=True)106    return (107        all_entities108        .groupby(["title", "type"], sort=False)109        .agg(110            description=("description", list),111            text_unit_ids=("source_id", list),112            frequency=("source_id", "count"),113        )114        .reset_index()115    )116117118def _merge_relationships(relationship_dfs) -> pd.DataFrame:119    all_relationships = pd.concat(relationship_dfs, ignore_index=False)120    return (121        all_relationships122        .groupby(["source", "target"], sort=False)123        .agg(124            description=("description", list),125            text_unit_ids=("source_id", list),126            weight=("weight", "sum"),127        )128        .reset_index()129    )

49    # Group by title and resolve conflicts50    aggregated = (51        combined52        .groupby("title")53        .agg({54            "id": "first",55            "type": "first",56            "human_readable_id": "first",57            "description": lambda x: list(x.astype(str)),  # Ensure str58            # Concatenate nd.array into a single list59            "text_unit_ids": lambda x: list(itertools.chain(*x.tolist())),60            "degree": "first",  # todo: we could probably re-compute this with the entire new graph61        })62        .reset_index()63    )6465    # recompute frequency to include new text units66    aggregated["frequency"] = aggregated["text_unit_ids"].apply(len)6768    # Force the result into a DataFrame69    resolved: pd.DataFrame = pd.DataFrame(aggregated)7071    # Modify column order to keep consistency72    resolved = resolved.loc[:, ENTITIES_FINAL_COLUMNS]7374    return resolved, id_mapping