diff --git a/cookbook/use_cases/finance/03_Earnings_Call_Analysis.ipynb b/cookbook/use_cases/finance/03_Earnings_Call_Analysis.ipynb index 43a76703..dda18a38 100644 --- a/cookbook/use_cases/finance/03_Earnings_Call_Analysis.ipynb +++ b/cookbook/use_cases/finance/03_Earnings_Call_Analysis.ipynb @@ -371,8 +371,12 @@ "metadata": {}, "outputs": [], "source": [ + "from concurrent.futures import ThreadPoolExecutor, TimeoutError\n", "from semantica.semantic_extract import RelationExtractor\n", "\n", + "MAX_ENTITIES = 30\n", + "CHUNK_TIMEOUT = 60\n", + "\n", "relation_extractor = RelationExtractor(\n", " method=\"llm\",\n", " confidence_threshold=0.6,\n", @@ -392,42 +396,72 @@ " verbose=False,\n", ")\n", "\n", - "# Process ALL chunks with ALL entities\n", + "\n", + "def filter_entities(text, entities):\n", + " t = text.lower()\n", + " return [e for e in entities if e.text.lower() in t]\n", + "\n", + "\n", + "def process_chunk(idx, chunk, total):\n", + " text = get_chunk_text(chunk).strip()\n", + "\n", + " remaining = total - (idx + 1)\n", + "\n", + " if not text:\n", + " print(f\"Chunk {idx+1}/{total} | remaining {remaining} | skipped (empty)\")\n", + " return []\n", + "\n", + " chunk_entities = filter_entities(text, all_entities)[:MAX_ENTITIES]\n", + "\n", + " if len(chunk_entities) < 2:\n", + " print(\n", + " f\"Chunk {idx+1}/{total} | remaining {remaining} | \"\n", + " f\"skipped (entities={len(chunk_entities)})\"\n", + " )\n", + " return []\n", + "\n", + " print(\n", + " f\"Chunk {idx+1}/{total} | remaining {remaining} | \"\n", + " f\"entities={len(chunk_entities)}\"\n", + " )\n", + "\n", + " return relation_extractor.extract_relations(\n", + " text=text,\n", + " entities=chunk_entities,\n", + " verbose=False,\n", + " )\n", + "\n", + "\n", "relationships = []\n", "total_chunks = len(chunks)\n", "\n", - "print(f\"Processing {total_chunks} chunks with {len(all_entities)} entities each...\")\n", - "print(\"This may take a while. Starting now...\\n\")\n", + "with ThreadPoolExecutor(max_workers=1) as executor:\n", + " for i, c in enumerate(chunks):\n", + " future = executor.submit(process_chunk, i, c, total_chunks)\n", "\n", - "for i, c in enumerate(chunks):\n", - " text = get_chunk_text(c).strip()\n", - " if not text:\n", - " print(f\"Chunk {i+1}/{total_chunks}: Skipped (empty)\")\n", - " continue\n", - " \n", - " # Use ALL entities for each chunk\n", - " chunk_entities = all_entities\n", - " \n", - " # Show progress with remaining count\n", - " remaining = total_chunks - (i + 1)\n", - " print(f\"Chunk {i+1}/{total_chunks} ({remaining} remaining) - Processing {len(chunk_entities)} entities...\")\n", - " \n", - " rels = relation_extractor.extract_relations(\n", - " text=text,\n", - " entities=chunk_entities,\n", - " verbose=True, # Enable verbose to see logs\n", - " )\n", - " \n", - " relationships.extend(rels)\n", - " print(f\"Chunk {i+1}: Found {len(rels)} relations\")\n", + " try:\n", + " rels = future.result(timeout=CHUNK_TIMEOUT)\n", + " relationships.extend(rels)\n", + " print(f\" relations={len(rels)}\")\n", "\n", - "print(f\"\\nComplete! Total relationships from all {total_chunks} chunks: {len(relationships)}\")\n", + " except TimeoutError:\n", + " remaining = total_chunks - (i + 1)\n", + " print(\n", + " f\"Chunk {i+1}/{total_chunks} | remaining {remaining} | timed out\"\n", + " )\n", + "\n", + " except Exception as e:\n", + " remaining = total_chunks - (i + 1)\n", + " print(\n", + " f\"Chunk {i+1}/{total_chunks} | remaining {remaining} | failed: {e}\"\n", + " )\n", + "\n", + "print(f\"Done {total_chunks}/{total_chunks}\")\n", + "print(f\"Total relationships: {len(relationships)}\")\n", "\n", - "# Show sample relations\n", "if relationships:\n", - " print(\"\\nSample relationships:\")\n", - " for r in relationships[:15]:\n", - " print(f\"- {r.subject.text} → {r.predicate} → {r.object.text}\")" + " for r in relationships[:10]:\n", + " print(f\"{r.subject.text} → {r.predicate} → {r.object.text}\")\n" ] }, { @@ -448,39 +482,69 @@ "from semantica.conflicts import SourceTracker, SourceReference, ConflictDetector\n", "\n", "source_tracker = SourceTracker()\n", + "\n", "conflict_detector = ConflictDetector(\n", " source_tracker=source_tracker,\n", " similarity_threshold=0.8,\n", " confidence_threshold=0.7,\n", ")\n", "\n", - "for entity in all_entities:\n", - " entity_id = getattr(entity, \"id\", None) or getattr(entity, \"text\", \"\")\n", - " entity_text = getattr(entity, \"text\", \"\")\n", - " entity_label = getattr(entity, \"label\", \"UNKNOWN\")\n", + "entities = all_entities\n", + "extracted_relationships = relationships\n", "\n", + "for e in entities:\n", + " entity_id = getattr(e, \"id\", None) or e.text\n", " source_tracker.track_property_source(\n", - " entity_id,\n", - " \"name\",\n", - " entity_text,\n", - " # FIXED: Changed 'source' to 'document' to match SourceReference signature\n", + " entity_id=entity_id,\n", + " property_name=\"name\",\n", + " value=e.text,\n", " source=SourceReference(\n", - " document=\"earnings_call\", # Was incorrect: source=\"earnings_call\"\n", + " document=\"earnings_call\",\n", " timestamp=\"2024-Q1\",\n", - " metadata={\"entity_type\": entity_label},\n", + " metadata={\"entity_type\": getattr(e, \"label\", \"UNKNOWN\")},\n", " ),\n", " )\n", "\n", - "value_conflicts = conflict_detector.detect_value_conflicts(\n", - " [{\"id\": getattr(e, \"id\", \"\"), \"name\": getattr(e, \"text\", \"\")} for e in all_entities],\n", + "entity_records = [\n", + " {\n", + " \"id\": getattr(e, \"id\", None) or e.text,\n", + " \"name\": e.text,\n", + " }\n", + " for e in entities\n", + "]\n", + "\n", + "entity_value_conflicts = conflict_detector.detect_value_conflicts(\n", + " entity_records,\n", " property_name=\"name\",\n", ")\n", "\n", - "relationship_conflicts = conflict_detector.detect_relationship_conflicts(relationships)\n", + "normalized_relationships = [\n", + " {\n", + " \"id\": getattr(r, \"id\", None),\n", + " \"source_id\": getattr(r.subject, \"id\", None) or r.subject.text,\n", + " \"target_id\": getattr(r.object, \"id\", None) or r.object.text,\n", + " \"type\": r.predicate,\n", + " \"confidence\": getattr(r, \"confidence\", 1.0),\n", + " \"metadata\": {},\n", + " }\n", + " for r in extracted_relationships\n", + "]\n", + "\n", + "relationship_conflicts = conflict_detector.detect_relationship_conflicts(\n", + " normalized_relationships\n", + ")\n", "\n", "print(\"Conflict detection completed\")\n", - "print(\"Value conflicts:\", len(value_conflicts))\n", - "print(\"Relationship conflicts:\", len(relationship_conflicts))" + "print(\"Entity value conflicts:\", len(entity_value_conflicts))\n", + "print(\"Relationship conflicts:\", len(relationship_conflicts))\n", + "\n", + "if entity_value_conflicts:\n", + " print(\"\\nSample entity conflict:\")\n", + " print(entity_value_conflicts[0])\n", + "\n", + "if relationship_conflicts:\n", + " print(\"\\nSample relationship conflict:\")\n", + " print(relationship_conflicts[0])" ] }, { @@ -505,20 +569,36 @@ " source_tracker=source_tracker,\n", ")\n", "\n", - "resolved_conflicts = []\n", + "resolved_entity_value_conflicts = []\n", + "resolved_relationship_conflicts = []\n", "\n", - "for conflict in value_conflicts:\n", - " resolved_conflicts.append(\n", - " conflict_resolver.resolve_conflict(conflict, strategy=\"voting\")\n", + "for conflict in entity_value_conflicts:\n", + " resolved_entity_value_conflicts.append(\n", + " conflict_resolver.resolve_conflict(\n", + " conflict,\n", + " strategy=\"voting\",\n", + " )\n", " )\n", "\n", "for conflict in relationship_conflicts:\n", - " resolved_conflicts.append(\n", - " conflict_resolver.resolve_conflict(conflict, strategy=\"voting\")\n", + " resolved_relationship_conflicts.append(\n", + " conflict_resolver.resolve_conflict(\n", + " conflict,\n", + " strategy=\"voting\",\n", + " )\n", " )\n", "\n", "print(\"Conflict resolution completed\")\n", - "print(\"Total conflicts resolved:\", len(resolved_conflicts))\n" + "print(\"Entity value conflicts resolved:\", len(resolved_entity_value_conflicts))\n", + "print(\"Relationship conflicts resolved:\", len(resolved_relationship_conflicts))\n", + "\n", + "if resolved_entity_value_conflicts:\n", + " print(\"\\nSample resolved entity conflict:\")\n", + " print(resolved_entity_value_conflicts[0])\n", + "\n", + "if resolved_relationship_conflicts:\n", + " print(\"\\nSample resolved relationship conflict:\")\n", + " print(resolved_relationship_conflicts[0])" ] }, { @@ -538,37 +618,72 @@ "source": [ "from semantica.deduplication import DuplicateDetector, EntityMerger\n", "\n", + "# Initialize deduplication components\n", "duplicate_detector = DuplicateDetector(\n", " similarity_threshold=0.8,\n", " confidence_threshold=0.7,\n", ")\n", "\n", + "entity_merger = EntityMerger(preserve_provenance=True)\n", + "\n", + "# Prepare entities (using original entities since no entity conflicts were found)\n", "entity_dicts = [\n", " {\n", - " \"id\": getattr(e, \"id\", \"\"),\n", - " \"name\": getattr(e, \"text\", \"\"),\n", + " \"id\": getattr(e, \"id\", None) or e.text,\n", + " \"name\": e.text,\n", " \"type\": getattr(e, \"label\", \"UNKNOWN\"),\n", " \"confidence\": getattr(e, \"confidence\", 1.0),\n", " \"metadata\": getattr(e, \"metadata\", {}),\n", " }\n", - " for e in resolved_entities\n", + " for e in entities\n", "]\n", "\n", + "# Detect and merge duplicate entities\n", "duplicates = duplicate_detector.detect_duplicates(entity_dicts)\n", "\n", - "entity_merger = EntityMerger(preserve_provenance=True)\n", - "\n", "merge_operations = entity_merger.merge_duplicates(\n", - " entity_dicts,\n", + " entities=entity_dicts,\n", + " duplicates=duplicates,\n", " strategy=\"keep_most_complete\",\n", ")\n", "\n", - "merged_entities = [op.merged_entity for op in merge_operations]\n", + "deduplicated_entities = [op.merged_entity for op in merge_operations]\n", "\n", + "# Update relationships to use merged entity IDs\n", + "entity_id_mapping = {\n", + " source_id: op.merged_entity['id']\n", + " for op in merge_operations\n", + " for source_id in op.source_ids\n", + "}\n", + "\n", + "# Update relationships with merged entity IDs and resolved conflicts\n", + "deduplicated_relationships = []\n", + "for rel in normalized_relationships:\n", + " updated_rel = rel.copy()\n", + " updated_rel['source_id'] = entity_id_mapping.get(rel['source_id'], rel['source_id'])\n", + " updated_rel['target_id'] = entity_id_mapping.get(rel['target_id'], rel['target_id'])\n", + " \n", + " # Apply resolved conflict values\n", + " for resolution in resolved_relationship_conflicts:\n", + " if resolution.resolved and resolution.metadata.get('relationship_id') == rel.get('id'):\n", + " property_name = resolution.metadata.get('property_name')\n", + " updated_rel[property_name] = resolution.resolved_value\n", + " \n", + " deduplicated_relationships.append(updated_rel)\n", + "\n", + "# Results\n", "print(\"Entity deduplication completed\")\n", - "print(\"Original entities:\", len(entity_dicts))\n", - "print(\"Merged entities:\", len(merged_entities))\n", - "print(\"Duplicates removed:\", len(entity_dicts) - len(merged_entities))\n" + "print(f\"Original entities: {len(entity_dicts)}\")\n", + "print(f\"Deduplicated entities: {len(deduplicated_entities)}\")\n", + "print(f\"Duplicates removed: {len(entity_dicts) - len(deduplicated_entities)}\")\n", + "\n", + "print(\"\\nRelationship updates completed\")\n", + "print(f\"Relationships updated: {len(deduplicated_relationships)}\")\n", + "print(f\"Conflicts resolved: {len(resolved_relationship_conflicts)}\")\n", + "\n", + "if merge_operations:\n", + " print(\"\\nSample merge operation:\")\n", + " print(merge_operations[0])" ] }, {